From bc14e7e3b72f484c206ada7024f73a308ccd3d2e Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 20 Apr 2012 11:45:57 -0400 Subject: [PATCH] zipline, now a cold, heartless, http://open.spotify.com/track/1xshgoh575otNXRfeYgh9D Pretty fast too... --- zipline/finance/performance.py | 4 +- zipline/finance/trading.py | 198 +++++++---------------------- zipline/lines.py | 16 ++- zipline/test/test_finance.py | 32 ++--- zipline/test/test_perf_tracking.py | 20 +-- 5 files changed, 70 insertions(+), 200 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 770baa45..4f6b3d43 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -377,8 +377,8 @@ class PerformancePeriod(): initial_positions, starting_value, starting_cash, - period_open, - period_close, + period_open=None, + period_close=None, keep_transactions=False): self.period_open = period_open diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 228572ec..9c815156 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -27,7 +27,7 @@ SIMULATION_STYLE = Enum( class TradeSimulationClient(qmsg.Component): - def __init__(self, trading_environment): + def __init__(self, trading_environment, sim_style): qmsg.Component.__init__(self) self.received_count = 0 self.prev_dt = None @@ -40,6 +40,7 @@ class TradeSimulationClient(qmsg.Component): self.algorithm = None self.max_wait = datetime.timedelta(seconds=60) self.last_msg_dt = datetime.datetime.utcnow() + self.txn_sim = TransactionSimulator(sim_style) assert self.trading_environment.frame_index != None self.event_frame = pandas.DataFrame( @@ -63,12 +64,8 @@ class TradeSimulationClient(qmsg.Component): def open(self): self.result_feed = self.connect_result() - self.order_socket = self.connect_order() - # send a wake up call to the order data source. - self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK)) def do_work(self): - # poll all the sockets socks = dict(self.poll.poll(self.heartbeat_timeout)) @@ -99,54 +96,49 @@ class TradeSimulationClient(qmsg.Component): # update performance and relay the event to the algorithm self.process_event(event) - # signal loop is done for order source. - self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK)) - else: - # no events in the sock means the non-order sources are - # drained. Signal the order_source that we're done, and - # the done will cascade through the whole zipline. - # shutdown the feedback loop to the OrderDataSource - wait_time = datetime.datetime.utcnow() - self.last_msg_dt - if wait_time > self.max_wait: - self.signal_order_done() - + def process_event(self, event): - # track the number of transactions, for testing purposes. - if(event.TRANSACTION != None): + + # generate transactions, if applicable + txn = self.txn_sim.apply_trade_to_open_orders(event) + if txn: + event.TRANSACTION = txn + # track the number of transactions, for testing purposes. self.txn_count += 1 + else: + event.TRANSACTION = None + + # the performance class needs to process each event, without + # skipping. Algorithm should wait until the performance has been + # updated, so that down stream components can safely assume that + # performance is up to date. Note that this is done before we + # mark the time for the algorithm's processing, thereby not + # running the algo's clock for performance book keeping. + self.perf.process_event(event) - #filter order flow out of the events sent to callbacks - if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE: - - # the performance class needs to process each event, without - # skipping. Algorithm should wait until the performance has been - # updated, so that down stream components can safely assume that - # performance is up to date. Note that this is done before we - # mark the time for the algorithm's processing, thereby not - # running the algo's clock for performance book keeping. - self.perf.process_event(event) - - # mark the start time for client's processing of this event. - event_start = datetime.datetime.utcnow() - - # queue the event. - self.queue_event(event) - - # if the event is later than our current time, run the algo - # otherwise, the algorithm has fallen behind the feed - # and processing per event is longer than time between events. - if event.dt >= self.current_dt: - # compress time by moving the current_time up to the event - # time. - self.current_dt = event.dt - self.run_algorithm() - - # tally the time spent on this iteration - self.last_iteration_dur = datetime.datetime.utcnow() - event_start - # move the algorithm's clock forward to include iteration time - self.current_dt = self.current_dt + self.last_iteration_dur + # mark the start time for client's processing of this event. + event_start = datetime.datetime.utcnow() + # queue the event. + self.queue_event(event) + + + # if the event is later than our current time, run the algo + # otherwise, the algorithm has fallen behind the feed + # and processing per event is longer than time between events. + if event.dt >= self.current_dt: + # compress time by moving the current_time up to the event + # time. + self.current_dt = event.dt + self.run_algorithm() + + # tally the time spent on this iteration + self.last_iteration_dur = datetime.datetime.utcnow() - event_start + # move the algorithm's clock forward to include iteration time + self.current_dt = self.current_dt + self.last_iteration_dur + + def run_algorithm(self): """ As per the algorithm protocol: @@ -164,15 +156,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 self.perf.log_order(order) + self.txn_sim.add_open_order(order) def signal_order_done(self): self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE)) @@ -188,91 +179,11 @@ class TradeSimulationClient(qmsg.Component): self.event_frame[event['sid']] = event self.event_queue = [] return self.event_frame - -class OrderDataSource(qmsg.DataSource): - """DataSource that relays orders from the client""" + - def __init__(self): - """ - :param simulation_time: datetime in UTC timezone, sets the start - time of simulation. orders - will be timestamped relative to this datetime. - event = { - 'sid' : an integer for security id, - 'dt' : datetime object, - 'price' : float for price, - 'volume' : integer for volume - } - """ - qmsg.DataSource.__init__(self, zp.FINANCE_COMPONENT.ORDER_SOURCE) - self.sent_count = 0 - self.recv_count = Counter() - - @property - def get_type(self): - return zp.DATASOURCE_TYPE.ORDER - - def open(self): - qmsg.DataSource.open(self) - self.order_socket = self.bind_order() - - def bind_order(self): - return self.bind_pull_socket(self.addresses['order_address']) - - def do_work(self): - - self.recv_count['work_loops'] += 1 - - #pull all orders from client. - count = 0 - - # one iteration of the client could include several orders - # so iterate until the client signals a break or a close. - # while True: - # poll all the sockets - # 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. - - # this will block for timeout/2, and return an empty dict if there - # are no messages. - - socks = dict(self.poll.poll()) - - # see if the poller has results for the result_feed - if self.order_socket in socks and \ - socks[self.order_socket] == self.zmq.POLLIN: - - order_msg = self.order_socket.recv() - - if order_msg == str(zp.ORDER_PROTOCOL.DONE): - qutil.LOGGER.info("order source is done") - self.signal_done() - self.recv_count['done'] += 1 - return - - if order_msg == str(zp.ORDER_PROTOCOL.BREAK): - # send a blank message to avoid an empty buffer - # in the feed - self.recv_count['break'] += 1 - if self.sent_count == 0: - self.send(namedict({})) - self.sent_count = 0 - return - - order = zp.ORDER_UNFRAME(order_msg) - self.recv_count['order'] += 1 - #send the order along - self.send(order) - count += 1 - self.sent_count += 1 - - - -class TransactionSimulator(qmsg.BaseTransform): +class TransactionSimulator(object): def __init__(self, style=SIMULATION_STYLE.PARTIAL_VOLUME): - qmsg.BaseTransform.__init__(self, zp.TRANSFORM_TYPE.TRANSACTION) self.open_orders = {} self.order_count = 0 self.txn_count = 0 @@ -289,27 +200,6 @@ class TransactionSimulator(qmsg.BaseTransform): elif style == SIMULATION_STYLE.NOOP: self.apply_trade_to_open_orders = self.simulate_noop - def transform(self, event): - """ - Pulls one message from the event feed, then - loops on orders until client sends DONE message. - """ - if(event.type == zp.DATASOURCE_TYPE.ORDER): - self.add_open_order(event) - self.state['value'] = None - elif(event.type == zp.DATASOURCE_TYPE.TRADE): - txn = self.apply_trade_to_open_orders(event) - self.state['value'] = txn - else: - self.state['value'] = None - log = "unexpected event type in transform: {etype}".format( - etype=event.type - ) - qutil.LOGGER.info(log) - - #TODO: what to do if we get another kind of datasource event.type? - return self.state - def add_open_order(self, event): """Orders are captured in a buffer by sid. No calculations are done here. Amount is explicitly converted to an int. @@ -324,8 +214,6 @@ class TransactionSimulator(qmsg.BaseTransform): ) qutil.LOGGER.debug(log) return - - if(not self.open_orders.has_key(event.sid)): self.open_orders[event.sid] = [] diff --git a/zipline/lines.py b/zipline/lines.py index 43c2eac0..56ec2b9a 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -86,8 +86,7 @@ import zipline.messaging as zmsg from zipline.test.algorithms import TestAlgorithm from zipline.sources import SpecificEquityTrades -from zipline.finance.trading import TransactionSimulator, OrderDataSource, \ -TradeSimulationClient +from zipline.finance.trading import TradeSimulationClient from zipline.simulator import AddressAllocator, Simulator from zipline.monitor import Controller from zipline.finance.trading import SIMULATION_STYLE @@ -164,18 +163,21 @@ class SimulatedTrading(object): self.sim = config['simulator_class'](addresses) self.clients = {} - self.trading_client = TradeSimulationClient(self.trading_environment) + self.trading_client = TradeSimulationClient( + self.trading_environment, + self.sim_style + ) self.add_client(self.trading_client) # setup all sources self.sources = {} - self.order_source = OrderDataSource() - self.add_source(self.order_source) + #self.order_source = OrderDataSource() + #self.add_source(self.order_source) #setup transforms - self.transaction_sim = TransactionSimulator(self.sim_style) + #self.transaction_sim = TransactionSimulator(self.sim_style) self.transforms = {} - self.add_transform(self.transaction_sim) + #self.add_transform(self.transaction_sim) self.sim.register_controller( self.con ) self.sim.on_done = self.shutdown() diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index a85f220d..763d8024 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -16,7 +16,7 @@ import zipline.finance.performance as perf from zipline.test.algorithms import TestAlgorithm from zipline.sources import SpecificEquityTrades -from zipline.finance.trading import TransactionSimulator, OrderDataSource, \ +from zipline.finance.trading import TransactionSimulator, \ TradeSimulationClient, TradingEnvironment from zipline.simulator import AddressAllocator, Simulator from zipline.monitor import Controller @@ -214,14 +214,8 @@ class FinanceTestCase(TestCase): zipline.algorithm.incr, "The test algorithm should send as many orders as specified.") - order_source = zipline.sources[zp.FINANCE_COMPONENT.ORDER_SOURCE] - self.assertEqual( - order_source.sent_count, - zipline.algorithm.count, - "The order source should have sent as many orders as the algo." - ) - transaction_sim = zipline.transforms[zp.TRANSFORM_TYPE.TRANSACTION] + transaction_sim = zipline.trading_client.txn_sim self.assertEqual( transaction_sim.txn_count, zipline.trading_client.perf.txn_count, @@ -426,11 +420,7 @@ class FinanceTestCase(TestCase): 'dt' : start_date + i * order_interval }) - sim_state = trade_sim.transform(order) - - # there should not be a new transaction from an order. - self.assertTrue(sim_state['name'] == trade_sim.get_id) - self.assertTrue(sim_state['value'] == None) + trade_sim.add_open_order(order) # there should now be one open order list stored under the sid oo = trade_sim.open_orders @@ -446,21 +436,19 @@ class FinanceTestCase(TestCase): tracker = PerformanceTracker(trading_environment) + # this approximates the loop inside TradingSimulationClient transactions = [] for trade in generated_trades: if trade_delay: trade.dt = trade.dt + trade_delay - sim_state = trade_sim.transform(trade) - - self.assertEqual(sim_state['name'], trade_sim.get_id) - - txn = None - if sim_state['value']: - txn = sim_state['value'] + txn = trade_sim.apply_trade_to_open_orders(trade) + if txn: transactions.append(txn) - trade[sim_state['name']] = txn - + trade.TRANSACTION = txn + else: + trade.TRANSACTION = None + tracker.process_event(trade) total_volume = 0 diff --git a/zipline/test/test_perf_tracking.py b/zipline/test/test_perf_tracking.py index be7c9a25..e952fed4 100644 --- a/zipline/test/test_perf_tracking.py +++ b/zipline/test/test_perf_tracking.py @@ -10,7 +10,8 @@ import zipline.util as qutil import zipline.finance.performance as perf import zipline.finance.risk as risk import zipline.protocol as zp -from zipline.finance.trading import TradeSimulationClient, TradingEnvironment +from zipline.finance.trading import TradeSimulationClient, TradingEnvironment, \ +SIMULATION_STYLE class PerformanceTestCase(unittest.TestCase): def setUp(self): @@ -539,11 +540,7 @@ shares in position" self.trading_environment.capital_base = 1000.0 self.trading_environment.frame_index = ['sid', 'volume', 'dt', \ 'price', 'changed'] - client = TradeSimulationClient(self.trading_environment) - # the client expects an algorithm that fullfills the algorithm - # protocol, so we use the noop algorithm. - test_algo = zipline.test.algorithms.NoopAlgorithm() - client.set_algorithm(test_algo) + perf_tracker = perf.PerformanceTracker(self.trading_environment) for event in trade_history: #create a transaction for all but @@ -559,18 +556,13 @@ shares in position" else: txn = None event[zp.TRANSFORM_TYPE.TRANSACTION] = txn - client.process_event(event) - - df = client.get_frame() - - self.assertEqual(df[133]['price'], price) - self.assertEqual(df[134]['price'], price2) + perf_tracker.process_event(event) #we skip two trades, to test case of None transaction txn_count = len(trade_history) - 2 - self.assertEqual(client.perf.txn_count, txn_count) + self.assertEqual(perf_tracker.txn_count, txn_count) - cumulative_pos = client.perf.cumulative_performance.positions[sid] + cumulative_pos = perf_tracker.cumulative_performance.positions[sid] expected_size = txn_count / 2 * -25 self.assertEqual(cumulative_pos.amount, expected_size) \ No newline at end of file