From b44ca63557ce3ab69512cfd37724f4388222dc4a Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 9 Mar 2012 01:07:01 -0500 Subject: [PATCH] refactoring orders to discard placeholder messages. --- zipline/finance/performance.py | 29 ++++++++++++-------- zipline/finance/risk.py | 16 +++++++----- zipline/finance/trading.py | 34 ++++++++++++++---------- zipline/messaging.py | 6 ++++- zipline/protocol.py | 48 ++++++++++++++++++++++++---------- zipline/test/client.py | 8 +++--- zipline/test/test_finance.py | 18 ++++++++----- 7 files changed, 102 insertions(+), 57 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 67b05be6..5c930715 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -86,10 +86,11 @@ class PerformanceTracker(): self.returns.append(todays_return_obj) #calculate risk metrics for cumulative performance - self.cur_period_metrics = risk.RiskMetrics(start_date=self.cumulative_performance.period_start, - end_date=self.cumulative_performance.period_end.replace(hour=0, minute=0, second=0), - returns=self.returns, - trading_environment=self.trading_environment) + self.cur_period_metrics = risk.RiskMetrics( + start_date=self.cumulative_performance.period_start, + end_date=self.cumulative_performance.period_end.replace(hour=0, minute=0, second=0), + returns=self.returns, + trading_environment=self.trading_environment) ###################################################################################################### #######TODO: report/relay metrics out to qexec -- values come from self.cur_period_metrics ########### @@ -97,14 +98,20 @@ class PerformanceTracker(): ###################################################################################################### #roll over positions to current day. - self.todays_performance = PerformancePeriod(self.market_open, - self.market_close, - self.todays_performance.positions, - self.todays_performance.ending_value, - self.capital_base) - + self.todays_performance = PerformancePeriod( + self.market_open, + self.market_close, + self.todays_performance.positions, + self.todays_performance.ending_value, + self.capital_base + ) + def handle_simulation_end(self): - self.risk_report = risk.RiskReport(self.returns, self.trading_environment) + self.risk_report = risk.RiskReport( + self.returns, + self.trading_environment + ) + ###################################################################################################### #######TODO: report/relay metrics out to qexec -- values come from self.risk_report ########### ###################################################################################################### diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index 2d6195c0..3607ec1d 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -202,13 +202,15 @@ class RiskReport(): cur_end = advance_by_months(cur_start, months_per) - one_day if(cur_end > the_end): break - #qutil.LOGGER.debug("start: {start}, end: {end}".format(start=cur_start, end=cur_end)) - cur_period_metrics = RiskMetrics(start_date=cur_start, - end_date=cur_end, - returns=self.algorithm_returns, - benchmark_returns=self.bm_returns, - treasury_curves=self.treasury_curves, - trading_calendar=self.trading_calendar) + cur_period_metrics = RiskMetrics( + start_date=cur_start, + end_date=cur_end, + returns=self.algorithm_returns, + benchmark_returns=self.bm_returns, + treasury_curves=self.treasury_curves, + trading_calendar=self.trading_calendar + ) + ends.append(cur_period_metrics) cur_start = advance_by_months(cur_start, 1) diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 34c19b62..6a8b81d8 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -47,6 +47,7 @@ class TradeSimulationClient(qmsg.Component): return event = zp.MERGE_UNFRAME(msg) + for cb in self.event_callbacks: cb(event) @@ -57,6 +58,10 @@ class TradeSimulationClient(qmsg.Component): return self.connect_push_socket(self.addresses['order_address']) def order(self, sid, amount): + qutil.LOGGER.debug("ordering {amt} of {sid}".format( + amt=amount, + sid=sid + )) self.order_socket.send(zp.ORDER_FRAME(sid, amount)) def signal_order_done(self): @@ -100,7 +105,7 @@ class OrderDataSource(qmsg.DataSource): #TODO: if this is the first iteration, break deadlock by sending a dummy order if(self.sent_count == 0): - self.send_dummy() + self.send(zp.namedict({})) #pull all orders from client. orders = [] @@ -127,25 +132,19 @@ class OrderDataSource(qmsg.DataSource): sid, amount = zp.ORDER_UNFRAME(order_msg) #send the order along - self.last_iteration_duration = datetime.datetime.utcnow() - self.event_start dt = self.simulation_dt + self.last_iteration_duration - order_event = zp.namedict({"sid":sid, "amount":amount, "dt":dt}) + order = zp.namedict({"dt":dt, 'sid':sid, 'amount':amount}) - self.send(order_event) + self.send(order) count += 1 self.sent_count += 1 #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_dummy() + self.send(zp.namedict({})) self.sent_count += 1 - - def send_dummy(self): - dt = self.simulation_dt + self.last_iteration_duration - dummy_order = zp.namedict({"sid":0, "amount":0, "dt":dt}) - self.send(dummy_order) @@ -166,7 +165,6 @@ class TransactionSimulator(qmsg.BaseTransform): Pulls one message from the event feed, then loops on orders until client sends DONE message. """ - #TODO: need a way to send a placeholder txn, to avoid blocking merge... maybe customize merge to not block on txn? if(event.type == zp.DATASOURCE_TYPE.ORDER): self.add_open_order(event) self.state['value'] = None @@ -199,7 +197,8 @@ class TransactionSimulator(qmsg.BaseTransform): def apply_trade_to_open_orders(self, event): if(event.volume == 0): - #there are zero volume events bc some stocks trade less frequently than once per minute. + #there are zero volume events bc some stocks trade + #less frequently than once per minute. return self.create_dummy_txn(event.dt) if self.open_orders.has_key(event.sid): @@ -212,7 +211,8 @@ class TransactionSimulator(qmsg.BaseTransform): dt = event.dt for order in orders: - #we're using minute bars, so allow orders within 30 seconds of the trade + #we're using minute bars, so allow orders within + #30 seconds of the trade if((order.dt - event.dt) < self.trade_windwo): total_order += order.amount if(order.dt > dt): @@ -233,7 +233,13 @@ class TransactionSimulator(qmsg.BaseTransform): volume_share = .25 amount = volume_share * event.volume * direction impact = (volume_share)**2 * .1 * direction * event.price - return self.create_transaction(event.sid, amount, event.price + impact, dt.replace(tzinfo = pytz.utc), direction) + return self.create_transaction( + event.sid, + amount, + event.price + impact, + dt.replace(tzinfo = pytz.utc), + direction + ) def create_transaction(self, sid, amount, price, dt, direction): diff --git a/zipline/messaging.py b/zipline/messaging.py index 87673c00..bbbab784 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -293,7 +293,11 @@ class ParallelBuffer(Component): continue cur_source = events first_in_list = events[0] - + if first_in_list.dt == None: + #this is a filler event, discard + events.pop(0) + continue + if (earliest_event == None) or (first_in_list.dt <= earliest_event.dt): earliest_event = first_in_list earliest_source = cur_source diff --git a/zipline/protocol.py b/zipline/protocol.py index 89726fe6..5bd238d6 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -295,11 +295,27 @@ def DATASOURCE_FRAME(event): assert isinstance(event.source_id, basestring) assert isinstance(event.type, int), 'Unexpected type %s' % (event.type) + + #datasources will send sometimes send empty msgs to feel gaps + if len(event.keys()) == 2: + return msgpack.dumps(tuple([ + event.type, + event.source_id, + DATASOURCE_TYPE.EMPTY + ])) if(event.type == DATASOURCE_TYPE.TRADE): - return msgpack.dumps(tuple([event.type, TRADE_FRAME(event)])) + return msgpack.dumps(tuple([ + event.type, + event.source_id, + TRADE_FRAME(event) + ])) elif(event.type == DATASOURCE_TYPE.ORDER): - return msgpack.dumps(tuple([event.type, ORDER_SOURCE_FRAME(event)])) + return msgpack.dumps(tuple([ + event.type, + event.source_id, + ORDER_SOURCE_FRAME(event) + ])) else: raise INVALID_DATASOURCE_FRAME(str(event)) @@ -321,15 +337,21 @@ def DATASOURCE_UNFRAME(msg): """ try: - ds_type, payload = msgpack.loads(msg) + ds_type, source_id, payload = msgpack.loads(msg) assert isinstance(ds_type, int) - if(ds_type == DATASOURCE_TYPE.TRADE): - return TRADE_UNFRAME(payload) + rval = namedict({'source_id':source_id}) + if payload == DATASOURCE_TYPE.EMPTY: + child_value = namedict({'dt':None}) + elif(ds_type == DATASOURCE_TYPE.TRADE): + child_value = TRADE_UNFRAME(payload) elif(ds_type == DATASOURCE_TYPE.ORDER): - return ORDER_SOURCE_UNFRAME(payload) + child_value = ORDER_SOURCE_UNFRAME(payload) else: raise INVALID_DATASOURCE_FRAME(msg) - + + rval.merge(child_value) + return rval + except TypeError: raise INVALID_DATASOURCE_FRAME(msg) except ValueError: @@ -461,7 +483,6 @@ def TRADE_FRAME(event): """ assert isinstance(event, namedict) - assert isinstance(event.source_id, basestring) assert event.type == DATASOURCE_TYPE.TRADE assert isinstance(event.sid, int) assert isinstance(event.price, numbers.Real) @@ -474,13 +495,12 @@ def TRADE_FRAME(event): event.epoch, event.micros, event.type, - event.source_id ])) def TRADE_UNFRAME(msg): try: packed = msgpack.loads(msg) - sid, price, volume, epoch, micros, source_type, source_id = packed + sid, price, volume, epoch, micros, source_type = packed assert isinstance(sid, int) assert isinstance(price, numbers.Real) @@ -491,8 +511,7 @@ def TRADE_UNFRAME(msg): 'volume' : volume, 'epoch' : epoch, 'micros' : micros, - 'type' : source_type, - 'source_id' : source_id + 'type' : source_type }) UNPACK_DATE(rval) return rval @@ -663,12 +682,13 @@ def UNPACK_DATE(event): DATASOURCE_TYPE = Enum( 'ORDER', - 'TRADE' + 'TRADE', + 'EMPTY', ) ORDER_PROTOCOL = Enum( 'DONE', - 'BREAK' + 'BREAK', ) diff --git a/zipline/test/client.py b/zipline/test/client.py index ccc6e397..19bc00b5 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -77,10 +77,12 @@ class TestAlgorithm(): self.incr = 0 def handle_event(self, event): + qutil.LOGGER.debug(event) #place an order for 100 shares of sid:133 - if(self.incr < self.count): - self.trading_client.order(self.sid, self.amount) - self.incr += 1 + if self.incr < self.count: + if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE: + self.trading_client.order(self.sid, self.amount) + self.incr += 1 else: self.trading_client.signal_order_done() self.trading_client.signal_done() diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 8d34938d..b367299e 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -202,8 +202,7 @@ class FinanceTestCase(TestCase): .format(n=sim.feed.pending_messages())) - def test_performance(self): - + def test_performance(self): # verify order -> transaction -> portfolio position. # -------------- @@ -278,12 +277,17 @@ class FinanceTestCase(TestCase): sim_context = sim.simulate() sim_context.join() - - - # TODO: Make more assertions about the final state of the components. - self.assertEqual(sim.feed.pending_messages(), 0, \ + self.assertEqual( + sim.feed.pending_messages(), + 0, "The feed should be drained of all messages, found {n} remaining." \ - .format(n=sim.feed.pending_messages())) + .format(n=sim.feed.pending_messages()) + ) + + self.assertEqual( + test_algo.count, + test_algo.incr, + "The test algorithm should send as many orders as specified.") self.assertEqual( order_source.sent_count,