From 02d7f0a4c8988a32f849b5ebaa78f6b8a8ec3f6f Mon Sep 17 00:00:00 2001 From: fawce Date: Sat, 14 Apr 2012 15:09:52 -0400 Subject: [PATCH] heavy work on the feedback loop from trade client to order source. tests are passing using a busy wait inside the trade client. hopefully we can find a more elegant approach. --- zipline/finance/trading.py | 44 ++++++++++++++++++++++++++++-------- zipline/messaging.py | 2 +- zipline/sources.py | 1 + zipline/test/test_finance.py | 41 ++++++++++++++++++++++++--------- 4 files changed, 67 insertions(+), 21 deletions(-) diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index f9c0adee..cab1051a 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -3,6 +3,8 @@ import pytz import math import pandas +from collections import Counter + # from gevent.select import select from zmq.core.poll import select @@ -11,7 +13,7 @@ import zipline.util as qutil import zipline.protocol as zp import zipline.finance.performance as perf -from zipline.protocol_utils import Enum +from zipline.protocol_utils import Enum, namedict # the simulation style enumerates the available transaction simulation # strategies. @@ -35,6 +37,8 @@ class TradeSimulationClient(qmsg.Component): self.current_dt = trading_environment.period_start self.last_iteration_dur = datetime.timedelta(seconds=0) self.algorithm = None + self.attempts = 0 + self.max_attempts = 1000 assert self.trading_environment.frame_index != None self.event_frame = pandas.DataFrame( @@ -59,8 +63,11 @@ 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)) @@ -68,6 +75,8 @@ class TradeSimulationClient(qmsg.Component): if self.result_feed in socks and \ socks[self.result_feed] == self.zmq.POLLIN: + self.attempts = 0 + # get the next message from the result feed msg = self.result_feed.recv() @@ -77,8 +86,7 @@ class TradeSimulationClient(qmsg.Component): # signal the performance tracker that the simulation has # ended. Perf will internally calculate the full risk report. self.perf.handle_simulation_end() - # shutdown the feedback loop to the OrderDataSource - self.signal_order_done() + # signal Simulator, our ComponentHost, that this component is # done and Simulator needn't block exit on this component. self.signal_done() @@ -86,13 +94,21 @@ class TradeSimulationClient(qmsg.Component): # result_feed is a merge component, so unframe accordingly event = zp.MERGE_UNFRAME(msg) - + self.received_count += 1 # update performance and relay the event to the algorithm self.process_event(event) - # signal done to order source. + # 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 + if self.attempts > self.max_attempts: + self.signal_order_done() + else: + self.attempts += 1 def process_event(self, event): # track the number of transactions, for testing purposes. if(event.TRANSACTION != None): @@ -189,6 +205,7 @@ class OrderDataSource(qmsg.DataSource): """ qmsg.DataSource.__init__(self, zp.FINANCE_COMPONENT.ORDER_SOURCE) self.sent_count = 0 + self.recv_count = Counter() self.works = 0 @property @@ -202,7 +219,7 @@ class OrderDataSource(qmsg.DataSource): This datasource is in a loop with the TradingSimulationClient, so we don't want it to block processing. """ - return False + return True def open(self): qmsg.DataSource.open(self) @@ -215,11 +232,11 @@ class OrderDataSource(qmsg.DataSource): self.works += 1 - #pull all orders from client. - orders = [] 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 @@ -236,17 +253,26 @@ class OrderDataSource(qmsg.DataSource): 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 count == 0: + self.send(namedict({})) break 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): diff --git a/zipline/messaging.py b/zipline/messaging.py index 75e181b0..b36dea95 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -347,7 +347,7 @@ class Feed(Component): def is_full(self): """ Indicates whether the buffer has messages in buffer for - all un-DONE sources. + all un-DONE, blocking sources. """ for source_id, events in self.data_buffer.iteritems(): if not self.is_blocking_map[source_id]: diff --git a/zipline/sources.py b/zipline/sources.py index b1385754..bf08644c 100644 --- a/zipline/sources.py +++ b/zipline/sources.py @@ -94,6 +94,7 @@ class SpecificEquityTrades(TradeDataSource): def get_type(self): zp.COMPONENT_TYPE.SOURCE + def do_work(self): if(len(self.event_list) == 0): self.signal_done() diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 34c0a050..2ff7da61 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -107,6 +107,10 @@ class FinanceTestCase(TestCase): self.assertTrue(env.last_close.month == 12) self.assertTrue(env.last_close.day == 31) + # The following two tests appear broken no that the order source is + # non blocking. HUNCH: The trades are streaming through before the orders + # are placed. + @timed(DEFAULT_TIMEOUT) def test_orders(self): @@ -128,13 +132,12 @@ class FinanceTestCase(TestCase): "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 - #) + 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 @@ -150,10 +153,18 @@ class FinanceTestCase(TestCase): # Simulation # ---------- - trade_count = 10 * 1000 - self.zipline_test_config['order_count'] = 5 * 1000 - self.zipline_test_config['trade_count'] = trade_count - self.zipline_test_config['order_amount'] = 100 + + # TODO: for some reason the orders aren't filled without an extra + # trade. + trade_count = 5001 + self.zipline_test_config['order_count'] = trade_count - 1 + self.zipline_test_config['trade_count'] = trade_count + self.zipline_test_config['order_amount'] = 1 + + # tell the simulator to fill the orders in individual transactions + # matching the order volume exactly. + self.zipline_test_config['simulation_style'] = \ + SIMULATION_STYLE.FIXED_SLIPPAGE self.zipline_test_config['environment'] = factory.create_trading_environment() sid_list = [self.zipline_test_config['sid']] @@ -169,12 +180,20 @@ class FinanceTestCase(TestCase): self.assertTrue(zipline.sim.ready()) self.assertFalse(zipline.sim.exception) - - # TODO: Make more assertions about the final state of the components. + 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 + ) + + @timed(DEFAULT_TIMEOUT) def test_performance(self):