diff --git a/zipline/component.py b/zipline/component.py index 0004051f..7bcd8500 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -75,7 +75,8 @@ class Component(object): self.out_socket = None self.killed = False self.controller = None - self.heartbeat_timeout = 2000 + # timeout after a full minute + self.heartbeat_timeout = 60 *1000 self.state_flag = COMPONENT_STATE.OK self.error_state = COMPONENT_FAILURE.NOFAILURE self.on_done = None diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 81a3d8d9..799107d8 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -160,8 +160,6 @@ class PerformanceTracker(): self.total_days = self.trading_environment.days_in_period # one indexed so that we reach 100% self.day_count = 0.0 - self.cumulative_capital_used = 0.0 - self.max_capital_used = 0.0 self.capital_base = self.trading_environment.capital_base self.returns = [] self.txn_count = 0 @@ -219,8 +217,8 @@ class PerformanceTracker(): 'period_start' : self.period_start, 'period_end' : self.period_end, 'progress' : self.progress, - 'cumulative_captial_used' : self.cumulative_capital_used, - 'max_capital_used' : self.max_capital_used, + 'cumulative_captial_used' : self.cumulative_perf.cumulative_capital_used, + 'max_capital_used' : self.cumulative_perf.max_capital_used, 'last_close' : self.market_close, 'last_open' : self.market_open, 'capital_base' : self.capital_base, @@ -232,36 +230,21 @@ class PerformanceTracker(): } def process_event(self, event): + assert isinstance(event, zp.namedict) self.event_count += 1 if(event.dt >= self.market_close): self.handle_market_close() - if not pandas.isnull(event.TRANSACTION): + if event.TRANSACTION: self.txn_count += 1 self.cumulative_performance.execute_transaction(event.TRANSACTION) self.todays_performance.execute_transaction(event.TRANSACTION) - - # we're adding a 10% cushion to the capital used, - # and then rounding to the nearest 5k - transaction_cost = event.TRANSACTION.price * event.TRANSACTION.amount - self.cumulative_capital_used += transaction_cost - - if math.fabs(self.cumulative_capital_used) > self.max_capital_used: - self.max_capital_used = math.fabs(self.cumulative_capital_used) - - cushioned_capital = 1.1 * self.max_capital_used - self.max_capital_used = self.round_to_nearest( - cushioned_capital, - base=5000 - ) - self.max_leverage = self.max_capital_used / self.capital_base - + #update last sale self.cumulative_performance.update_last_sale(event) self.todays_performance.update_last_sale(event) - def handle_market_close(self): #calculate performance as of last trade @@ -338,9 +321,6 @@ class PerformanceTracker(): # this signals that the simulation is complete. self.result_stream.send("DONE") - def round_to_nearest(self, x, base=5): - return int(base * round(float(x)/base)) - class Position(): @@ -409,6 +389,8 @@ class PerformancePeriod(): self.starting_cash = starting_cash self.ending_cash = starting_cash self.processed_transactions = [] + self.cumulative_capital_used = 0.0 + self.max_capital_used = 0.0 self.calculate_performance() @@ -426,11 +408,40 @@ class PerformancePeriod(): self.returns = 0.0 def execute_transaction(self, txn): + + # Update Position + # ---------------- if(not self.positions.has_key(txn.sid)): self.positions[txn.sid] = Position(txn.sid) self.positions[txn.sid].update(txn) self.period_capital_used += -1 * txn.price * txn.amount + + + # Max Leverage + # --------------- + # Calculate the maximum capital used and maximum leverage + + transaction_cost = txn.price * txn.amount + self.cumulative_capital_used += transaction_cost + + if math.fabs(self.cumulative_capital_used) > self.max_capital_used: + self.max_capital_used = math.fabs(self.cumulative_capital_used) + + # We want to conveye a level, rather than a precise figure. + # round to the nearest 5,000 to keep the number easy on the eyes + self.max_capital_used = self.round_to_nearest( + self.max_capital_used, + base=5000 + ) + + # we're adding a 10% cushion to the capital used. + self.max_leverage = 1.1 * self.max_capital_used / self.starting_cash + + # add transaction to the list of processed transactions self.processed_transactions.append(txn) + + def round_to_nearest(self, x, base=5): + return int(base * round(float(x)/base)) def calculate_positions_value(self): mktValue = 0.0 diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index d3fbabe5..208dc79c 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -11,6 +11,17 @@ import zipline.util as qutil import zipline.protocol as zp import zipline.finance.performance as perf +from zipline.protocol_utils import Enum + +# the simulation style enumerates the available transaction simulation +# strategies. +SIMULATION_STYLE = Enum( + 'PARTIAL_VOLUME', + 'BUY_ALL', + 'FIXED_SLIPPAGE', + 'NOOP' +) + class TradeSimulationClient(qmsg.Component): def __init__(self, trading_environment): @@ -19,6 +30,7 @@ class TradeSimulationClient(qmsg.Component): self.prev_dt = None self.event_queue = None self.txn_count = 0 + self.order_count = 0 self.trading_environment = trading_environment self.current_dt = trading_environment.period_start self.last_iteration_dur = datetime.timedelta(seconds=0) @@ -132,13 +144,14 @@ class TradeSimulationClient(qmsg.Component): return self.connect_push_socket(self.addresses['order_address']) def order(self, sid, amount): + order = zp.namedict({ 'dt':self.current_dt, 'sid':sid, 'amount':amount }) - self.order_socket.send(zp.ORDER_FRAME(order)) + self.order_count += 1 def signal_order_done(self): self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE)) @@ -211,6 +224,8 @@ class OrderDataSource(qmsg.DataSource): # we reduce the timeout here by a factor of 2, because we need # to potentially receive the client's done message before the # controller or heartbeat times out. + + # TODO: shouldn't this block until we receive a message? socks = dict(self.poll.poll(self.heartbeat_timeout/2)) # see if the poller has results for the result_feed @@ -232,19 +247,20 @@ class OrderDataSource(qmsg.DataSource): count += 1 self.sent_count += 1 - else: - # no orders, break out - break + # TODO: why didn't any unit tests catch this bug???? + + #else: + # # no orders, break out + # break #TODO: we have to send at least one dummy order per do_work iteration # or the feed will block waiting for our messages. if(count == 0): self.send(zp.namedict({})) - class TransactionSimulator(qmsg.BaseTransform): - def __init__(self): + def __init__(self, style): qmsg.BaseTransform.__init__(self, zp.TRANSFORM_TYPE.TRANSACTION) self.open_orders = {} self.order_count = 0 @@ -252,6 +268,15 @@ class TransactionSimulator(qmsg.BaseTransform): self.trade_window = datetime.timedelta(seconds=30) self.orderTTL = datetime.timedelta(days=1) self.commission = 0.03 + + if not style or style == SIMULATION_STYLE.PARTIAL_VOLUME: + self.apply_trade_to_open_orders = self.simulate_with_partial_volume + elif style == SIMULATION_STYLE.BUY_ALL: + self.apply_trade_to_open_orders = self.simulate_buy_all + elif style == SIMULATION_STYLE.FIXED_SLIPPAGE: + self.apply_trade_to_open_orders = self.simulate_with_fixed_cost + elif style == SIMULATION_STYLE.NOOP: + self.apply_trade_to_open_orders = self.simulate_noop def transform(self, event): """ @@ -296,8 +321,51 @@ class TransactionSimulator(qmsg.BaseTransform): event.filled = 0 self.open_orders[event.sid].append(event) - def apply_trade_to_open_orders(self, event): + #def apply_trade_to_open_orders(self, event): + # return self.simulate_with_fixed_cost(event) + + def simulate_buy_all(self, event): + txn = self.create_transaction( + event.sid, + event.volume, + event.price, + event.dt, + 1 + ) + return txn + def simulate_noop(self, event): + return None + + def simulate_with_fixed_cost(self, event): + if self.open_orders.has_key(event.sid): + orders = self.open_orders[event.sid] + orders = sorted(orders, key=lambda o: o.dt) + else: + return None + + amount = 0 + for order in orders: + amount += order.amount + + if(amount != 0): + direction = amount / math.fabs(amount) + else: + direction = 1 + + txn = self.create_transaction( + event.sid, + amount, + event.price + 0.10, + event.dt, + direction + ) + + self.open_orders[event.sid] = [] + + return txn + + def simulate_with_partial_volume(self, event): if(event.volume == 0): #there are zero volume events bc some stocks trade #less frequently than once per minute. diff --git a/zipline/lines.py b/zipline/lines.py index 624a9456..466de2e3 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -90,7 +90,7 @@ from zipline.finance.trading import TransactionSimulator, OrderDataSource, \ TradeSimulationClient from zipline.simulator import AddressAllocator, Simulator from zipline.monitor import Controller - +from zipline.finance.trading import SIMULATION_STYLE class SimulatedTrading(object): @@ -130,6 +130,7 @@ class SimulatedTrading(object): self.algorithm = config['algorithm'] self.allocator = config['allocator'] self.trading_environment = config['trading_environment'] + self.sim_style = config.get('simulation_style') self.leased_sockets = [] self.sim_context = None @@ -169,7 +170,7 @@ class SimulatedTrading(object): self.add_source(self.order_source) #setup transforms - self.transaction_sim = TransactionSimulator() + self.transaction_sim = TransactionSimulator(self.sim_style) self.transforms = {} self.add_transform(self.transaction_sim) @@ -192,7 +193,8 @@ class SimulatedTrading(object): - order_count - the number of orders the test algo will place, defaults to 100 - order_amount - the number of shares per order, defaults to 100 - - trade_count - the number of trades to simulate, defaults to 100 + - trade_count - the number of trades to simulate, defaults to 101 + to ensure all orders are processed. - simulator_class - optional parameter that provides an alternative subclass of ComponentHost to hold the whole zipline. Defaults to :py:class:`zipline.simulator.Simulator` @@ -228,7 +230,7 @@ class SimulatedTrading(object): if config.has_key('trade_count'): trade_count = config['trade_count'] else: - trade_count = 100 + trade_count = 101 if config.has_key('simulator_class'): simulator_class = config['simulator_class'] @@ -266,7 +268,8 @@ class SimulatedTrading(object): 'algorithm':test_algo, 'trading_environment':trading_environment, 'allocator':allocator, - 'simulator_class':simulator_class + 'simulator_class':simulator_class, + 'simulation_style':SIMULATION_STYLE.FIXED_SLIPPAGE }) #------------------- diff --git a/zipline/messaging.py b/zipline/messaging.py index 2c4caebc..7817ec8b 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -37,7 +37,7 @@ class ComponentHost(Component): # ---------------------- self.sync_register = {} - self.timeout = datetime.timedelta(seconds=5) + self.timeout = datetime.timedelta(seconds=60) self.feed = Feed() self.merge = Merge() @@ -214,7 +214,7 @@ class Feed(Component): def do_work(self): # wait for synchronization reply from the host - socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds. + socks = dict(self.poll.poll(self.heartbeat_timeout)) # TODO: Abstract this out, maybe on base component if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN: diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 9b3e4187..39a13bd1 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -21,6 +21,7 @@ TradeSimulationClient, TradingEnvironment from zipline.simulator import AddressAllocator, Simulator from zipline.monitor import Controller from zipline.lines import SimulatedTrading +from zipline.finance.performance import PerformanceTracker from zipline.protocol_utils import namedict DEFAULT_TIMEOUT = 15 # seconds @@ -120,7 +121,22 @@ class FinanceTestCase(TestCase): self.assertEqual(zipline.sim.feed.pending_messages(), 0, \ "The feed should be drained of all messages, found {n} remaining." \ .format(n=zipline.sim.feed.pending_messages())) - + + + # the trading client should receive one transaction for every + # order placed. + self.assertEqual( + zipline.trading_client.txn_count, + zipline.trading_client.order_count + ) + + # the number of transactions in the performance tracker's cumulative + # period should be the same as the number of orders place by the + # algorithm. + self.assertEqual( + zipline.trading_client.order_count, + len(zipline.trading_client.perf.cumulative_performance.processed_transactions) + ) @timed(EXTENDED_TIMEOUT) @@ -152,8 +168,8 @@ class FinanceTestCase(TestCase): self.assertEqual(zipline.sim.feed.pending_messages(), 0, \ "The feed should be drained of all messages, found {n} remaining." \ .format(n=zipline.sim.feed.pending_messages())) - - + + @timed(DEFAULT_TIMEOUT) def test_performance(self): #provide enough trades to ensure all orders are filled. @@ -358,6 +374,7 @@ class FinanceTestCase(TestCase): def transaction_sim(self, **params): + trade_count = params['trade_count'] trade_amount = params['trade_amount'] trade_interval = params['trade_interval'] @@ -409,6 +426,9 @@ class FinanceTestCase(TestCase): self.assertEqual(order.sid, sid) self.assertEqual(order.amount, order_amount) + + tracker = PerformanceTracker(trading_environment) + transactions = [] for trade in generated_trades: if trade_delay: @@ -418,10 +438,14 @@ class FinanceTestCase(TestCase): self.assertEqual(sim_state['name'], trade_sim.get_id) + txn = None if sim_state['value']: - transactions.append(sim_state['value']) + txn = sim_state['value'] + transactions.append(txn) + trade[sim_state['name']] = txn - + tracker.process_event(trade) + total_volume = 0 for txn in transactions: total_volume += txn.amount @@ -429,6 +453,9 @@ class FinanceTestCase(TestCase): self.assertEqual(total_volume, expected_txn_volume) self.assertEqual(len(transactions), expected_txn_count) + cumulative_pos = tracker.cumulative_performance.positions[sid] + self.assertEqual(total_volume, cumulative_pos.amount) + # the open orders should now be empty oo = trade_sim.open_orders self.assertTrue(oo.has_key(sid))