From 7846a1e1f49f3754f137617c99e6800438ef70f3 Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 1 Mar 2012 10:46:50 -0500 Subject: [PATCH] passing basic order and transaction simulation test --- zipline/finance/trading.py | 112 +++++++++++++++++++---------------- zipline/protocol.py | 34 ++++++++--- zipline/test/client.py | 5 +- zipline/test/test_finance.py | 13 ++-- 4 files changed, 99 insertions(+), 65 deletions(-) diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index a6b81d2e..9420eb96 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -1,5 +1,7 @@ import json import datetime +import pytz +import math from zmq.core.poll import select @@ -35,25 +37,26 @@ class TradeSimulationClient(qmsg.Component): self.signal_done() return - event = zp.MERGE_UNFRAME(message) + event = zp.MERGE_UNFRAME(msg) self._handle_event(event) def connect_order(self): return self.connect_push_socket(self.addresses['order_address']) def _handle_event(self, event): - self.event_queue.append(event) - if event.SIM_DT <= event.dt: - #event occurred in the present, send the queue to be processed - self.handle_events(self.event_queue) - self.order_socket.send(str(zp.CONTROL_PROTOCOL.DONE)) + self.handle_event(event) + #signal done to order source. + self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK)) - def handle_events(self, event_queue): + def handle_event(self, event): raise NotImplementedError def order(self, sid, amount): self.order_socket.send(zp.ORDER_FRAME(sid, amount)) + def signal_order_done(self): + self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE)) + class OrderDataSource(qmsg.DataSource): """DataSource that relays orders from the client""" @@ -107,7 +110,11 @@ class OrderDataSource(qmsg.DataSource): continue order_msg = rlist[0].recv() - if order_msg == str(zp.CONTROL_PROTOCOL.DONE): + if order_msg == str(zp.ORDER_PROTOCOL.DONE): + self.signal_done() + return + + if order_msg == str(zp.ORDER_PROTOCOL.BREAK): qutil.LOGGER.info("order loop finished") break @@ -156,74 +163,77 @@ class TransactionSimulator(qmsg.BaseTransform): """ #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.sid, event.amount) - self.state['value'] = self.create_transaction(0, 0, 0.0, event.dt, 1) + self.add_open_order(event) + self.state['value'] = None elif(event.type == zp.DATASOURCE_TYPE.TRADE): - txn = apply_trade_to_open_orders(event) + txn = self.apply_trade_to_open_orders(event) self.state['value'] = txn + else: + self.state['value'] = None + qutil.LOGGER.info("unexpected event type in transform: {etype}".format(etype=event.type)) #TODO: what to do if we get another kind of datasource event.type? return self.state - def add_open_order(self, sid, amount): + 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. Orders of amount zero are ignored. """ - amount = int(amount) - if amount == 0: - qutil.LOGGER.debug("requested to trade zero shares of {sid}".format(sid=sid)) + event.amount = int(event.amount) + if event.amount == 0: + qutil.LOGGER.debug("requested to trade zero shares of {sid}".format(sid=event.sid)) return self.order_count += 1 - order = zp.namedict({'sid' : sid, - 'amount' : amount, - 'dt' : self.algo_time, - 'filled': 0, - 'direction': math.fabs(amount) / amount - }) - if(not self.open_orders.has_key(sid)): - self.open_orders[sid] = [] - self.open_orders[sid].append(order) + if(not self.open_orders.has_key(event.sid)): + self.open_orders[event.sid] = [] + self.open_orders[event.sid].append(event) 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. - return + return self.create_dummy_txn(event.dt) + if self.open_orders.has_key(event.sid): orders = self.open_orders[event.sid] - remaining_orders = [] - total_order = 0 - dt = event.dt - - for order in orders: - #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): - dt = order.dt - #if the order still has time to live (TTL) keep track - elif((self.algo_time - order.dt) < self.orderTTL): - remaining_orders.append(order) - - self.open_orders[event.sid] = remaining_orders - - if(total_order != 0): - direction = total_order / math.fabs(total_order) - volume_share = (direction * total_order) / event.volume - if volume_share > .25: - volume_share = .25 - amount = volume_share * event.volume * direction - impact = (volShare)**2 * .1 * direction * event.price - return self.create_transaction(event.sid, amount, event.price + impact, dt.replace(tzinfo = pytz.utc), direction) - + else: + return None + + remaining_orders = [] + total_order = 0 + dt = event.dt + + for order in orders: + #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): + dt = order.dt + #if the order still has time to live (TTL) keep track + elif((self.algo_time - order.dt) < self.orderTTL): + remaining_orders.append(order) + + self.open_orders[event.sid] = remaining_orders + + if(total_order != 0): + direction = total_order / math.fabs(total_order) + else: + direction = 1 + + volume_share = (direction * total_order) / event.volume + if volume_share > .25: + 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) def create_transaction(self, sid, amount, price, dt, direction): txn = {'sid' : sid, - 'amount' : amount, + 'amount' : int(amount), 'dt' : dt, 'price' : price, 'commission' : self.commission * amount * direction, diff --git a/zipline/protocol.py b/zipline/protocol.py index fe074840..e2b9bae2 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -106,7 +106,7 @@ class namedict(object): return "namedict: " + str(self.__dict__) def __eq__(self, other): - return self.__dict__ == other.__dict__ + return other != None and self.__dict__ == other.__dict__ def has_attr(self, name): return self.__dict__.has_key(name) @@ -263,13 +263,9 @@ def FEED_UNFRAME(msg): INVALID_TRANSFORM_FRAME = FrameExceptionFactory('TRANSFORM') def TRANSFORM_FRAME(name, value): - """ - :event: a nameddict with at least:: - - source_id - - type - """ assert isinstance(name, basestring) - assert value != None + if value == None: + return msgpack.dumps(tuple([name, TRANSFORM_TYPE.EMPTY])) if(name == TRANSFORM_TYPE.TRANSACTION): value = TRANSACTION_FRAME(value) return msgpack.dumps(tuple([name, value])) @@ -279,13 +275,17 @@ def TRANSFORM_UNFRAME(msg): :rtype: namedict with : """ try: + name, value = msgpack.loads(msg) + if(value == TRANSFORM_TYPE.EMPTY): + return namedict({name : None}) #TODO: anything we can do to assert more about the content of the dict? assert isinstance(name, basestring) if(name == TRANSFORM_TYPE.PASSTHROUGH): value = FEED_UNFRAME(value) elif(name == TRANSFORM_TYPE.TRANSACTION): value = TRANSACTION_UNFRAME(value) + return namedict({name : value}) except TypeError: raise INVALID_TRANSFORM_FRAME(msg) @@ -305,6 +305,11 @@ def MERGE_FRAME(event): """ assert isinstance(event, namedict) PACK_DATE(event) + if(event.has_attr(TRANSFORM_TYPE.TRANSACTION)): + if(event.TRANSACTION == None): + event.TRANSACTION = TRANSFORM_TYPE.EMPTY + else: + event.TRANSACTION = TRANSACTION_FRAME(event.TRANSACTION) payload = event.__dict__ return msgpack.dumps(payload) @@ -314,6 +319,11 @@ def MERGE_UNFRAME(msg): #TODO: anything we can do to assert more about the content of the dict? assert isinstance(payload, dict) payload = namedict(payload) + if(payload.has_attr(TRANSFORM_TYPE.TRANSACTION)): + if(payload.TRANSACTION == TRANSFORM_TYPE.EMPTY): + payload.TRANSACTION = None + else: + payload.TRANSACTION = TRANSACTION_UNFRAME(payload.TRANSACTION) UNPACK_DATE(payload) return payload except TypeError: @@ -476,9 +486,17 @@ DATASOURCE_TYPE = Enum( 'TRADE' , ) +ORDER_PROTOCOL = Enum( + 'DONE', + 'BREAK' +) + + +#Transform type needs to be a namedict to facilitate merging. TRANSFORM_TYPE = namedict({ 'TRANSACTION':'TRANSACTION', #needed? - 'PASSTHROUGH':'PASSTHROUGH' + 'PASSTHROUGH':'PASSTHROUGH', + 'EMPTY':'' }) diff --git a/zipline/test/client.py b/zipline/test/client.py index 02a8e4c1..00259118 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -51,9 +51,12 @@ class TestTradingClient(TradeSimulationClient): self.amount = amount self.incr = 0 - def handle_events(self, event_queue): + def handle_event(self, event): #place an order for 100 shares of sid:133 if(self.incr < self.count): self.order(self.sid, self.amount) self.incr += 1 + else: + self.signal_order_done() + self.signal_done() diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index c95b2b52..cd5b0a32 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -127,11 +127,14 @@ class FinanceTestCase(ThreadPoolExecutorMixin, TestCase): # Simulation Components # --------------------- - set1 = SpecificEquityTrades("flat-133",factory.create_trade_history(133, - [10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0], - [100,100,100,100,100,100,100,100,100,100,100,100,100,100,100,100], - datetime.datetime.strptime("02/1/2012","%m/%d/%Y"), - datetime.timedelta(days=1))) + set1 = SpecificEquityTrades("flat-133", + factory.create_trade_history( + 133, + [10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0,10.0], + [100,100,100,100,100,100,100,100,100,100,100,100,100,100,100,100], + datetime.datetime.strptime("02/1/2012","%m/%d/%Y"), + datetime.timedelta(days=1))) + #client sill send 10 orders for 100 shares of 133 client = TestTradingClient(133, 100, 10) order_source = OrderDataSource(datetime.datetime.strptime("02/1/2012","%m/%d/%Y").replace(tzinfo=pytz.utc))