From aa567a18545e321fa479635e01e423ca16884450 Mon Sep 17 00:00:00 2001 From: fawce Date: Wed, 29 Feb 2012 11:08:25 -0500 Subject: [PATCH 01/16] clearing unit test references from client. --- setup.cfg | 12 ++++++------ zipline/test/client.py | 8 +++----- 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/setup.cfg b/setup.cfg index 517174ae..c64d493c 100644 --- a/setup.cfg +++ b/setup.cfg @@ -2,12 +2,12 @@ verbosity=2 detailed-errors=1 -with-xcoverage=1 -cover-package=zipline -cover-erase=1 -cover-html=1 -cover-html-dir=docs/_build/html/cover -with-xunit=1 +#with-xcoverage=1 +#cover-package=zipline +#cover-erase=1 +#cover-html=1 +#cover-html-dir=docs/_build/html/cover +#with-xunit=1 #processes=2 diff --git a/zipline/test/client.py b/zipline/test/client.py index 33b9bf29..59e4d785 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -8,11 +8,10 @@ import zipline.protocol as zp class TestClient(qmsg.Component): """no-op client - Just connects to the merge and counts messages. compares received message count to the expected count.""" - def __init__(self, utest, expected_msg_count=0): + def __init__(self, expected_msg_count=0): qmsg.Component.__init__(self) self.received_count = 0 self.expected_msg_count = expected_msg_count - self.utest = utest self.prev_dt = None self.heartbeat_timeout = 2000 @@ -32,9 +31,8 @@ class TestClient(qmsg.Component): if msg == str(zp.CONTROL_PROTOCOL.DONE): qutil.LOGGER.info("Client is DONE!") self.signal_done() - self.utest.assertEqual(self.expected_msg_count, self.received_count, - "The client should have received ({n}) the same number of messages as the feed sent ({m})." - .format(n=self.received_count, m=self.expected_msg_count)) + if(self.expected_msg_count > 0): + assert self.received_count == self.expected_msg_count return self.received_count += 1 From 00c2ccfe72c345809e980ec990ea1b0465f64092 Mon Sep 17 00:00:00 2001 From: fawce Date: Wed, 29 Feb 2012 17:18:17 -0500 Subject: [PATCH 02/16] using an enum for finance protocol constants --- zipline/finance/trading.py | 10 +++++----- zipline/protocol.py | 6 +++--- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index ee13d432..e3a97e53 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -71,12 +71,12 @@ class OrderDataSource(qmsg.DataSource): 'volume' : integer for volume } """ - zm.DataSource.__init__(self, "ORDER_SIM") + zm.DataSource.__init__(self, str(zp.FINANCE_PROTOCOL.ORDER)) self.simulation_dt = simulation_dt self.last_iteration_duration = datetime.timedelta(seconds=0) def get_type(self): - return 'ORDER_SIM' + return str(zp.FINANCE_PROTOCOL.ORDER) def open(self): qmsg.DataSource.open(self) @@ -88,7 +88,7 @@ class OrderDataSource(qmsg.DataSource): def do_work(self): #mark the start time for client's processing of this event. self.event_start = datetime.datetime.utcnow() - self.result_socket.send(zp.TRANSFORM_FRAME('ORDER_SIM', self.simulation_dt), self.zmq.NOBLOCK) + self.result_socket.send(zp.TRANSFORM_FRAME(str(zp.FINANCE_PROTOCOL.ORDER), self.simulation_dt), self.zmq.NOBLOCK) self.simulation_dt = self.simulation_dt + self.last_iteration_duration @@ -139,10 +139,10 @@ class TransactionSimulator(qmsg.BaseTransform): Pulls one message from the event feed, then loops on orders until client sends DONE message. """ - if(event.type == "ORDER_SIM"): + if(event.type == zp.FINANCE_PROTOCOL.ORDER): self.add_open_order(event.sid, event.amount) self.state['value'] = self.average - elif(event.type == "EQUITY_TRADE"): + elif(event.type == zp.FINANCE_PROTOCOL.TRADE): txn = apply_trade_to_open_orders(event) self.state['value'] = txn diff --git a/zipline/protocol.py b/zipline/protocol.py index dc3fb408..22ca9712 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -429,7 +429,7 @@ def UNPACK_DATE(payload): FINANCE_PROTOCOL = Enum( - 'ORDER' , # 0 - req - 'TRANSACTION' , # 1 - req - 'STATUS' , # 2 - req + 'ORDER' , # 0 + 'TRANSACTION' , # 1 + 'TRADE' , # 2 ) From 536a1e7fdc9f655883921da3e3ff63f902e88176 Mon Sep 17 00:00:00 2001 From: fawce Date: Wed, 29 Feb 2012 23:38:46 -0500 Subject: [PATCH 03/16] protocol tests passing, orders test almost passing, but looking like we need to extend the merge component to add transaction and order specific logic. --- zipline/finance/trading.py | 104 +++++++++++++++------------ zipline/messaging.py | 5 +- zipline/protocol.py | 131 +++++++++++++++++++++++++---------- zipline/test/client.py | 16 ++--- zipline/test/factory.py | 60 +--------------- zipline/test/test_finance.py | 53 ++++++++++++-- 6 files changed, 211 insertions(+), 158 deletions(-) diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index e3a97e53..a6b81d2e 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -17,7 +17,7 @@ class TradeSimulationClient(qmsg.Component): @property def get_id(self): - return "TRADING_CLIENT" + return str(zp.FINANCE_COMPONENT.TRADING_CLIENT) def open(self): self.result_feed = self.connect_result() @@ -25,21 +25,18 @@ class TradeSimulationClient(qmsg.Component): def do_work(self): #next feed event - (rlist, wlist, xlist) = select([self.result_feed], - [], - [self.result_feed], - timeout=self.heartbeat_timeout/100) #select timeout is in sec, use 10x - # - #no more orders, should be an error condition - if len(rlist) == 0 or len(xlist) > 0: - raise Exception("unexpected end of feed stream") - message = rlist[0].recv() - if message == str(zp.CONTROL_PROTOCOL.DONE): - self.signal_done() - return #leave open orders hanging? client requests for orders? + socks = dict(self.poll.poll(self.heartbeat_timeout)) + + if self.result_feed in socks and socks[self.result_feed] == self.zmq.POLLIN: + msg = self.result_feed.recv() + + if msg == str(zp.CONTROL_PROTOCOL.DONE): + qutil.LOGGER.info("Client is DONE!") + self.signal_done() + return - event = zp.MERGE_UNFRAME(message) - self._handle_event(event) + event = zp.MERGE_UNFRAME(message) + self._handle_event(event) def connect_order(self): return self.connect_push_socket(self.addresses['order_address']) @@ -71,12 +68,13 @@ class OrderDataSource(qmsg.DataSource): 'volume' : integer for volume } """ - zm.DataSource.__init__(self, str(zp.FINANCE_PROTOCOL.ORDER)) + qmsg.DataSource.__init__(self, zp.FINANCE_COMPONENT.ORDER_SOURCE) self.simulation_dt = simulation_dt self.last_iteration_duration = datetime.timedelta(seconds=0) + self.sent_count = 0 def get_type(self): - return str(zp.FINANCE_PROTOCOL.ORDER) + return zp.FINANCE_COMPONENT.ORDER_SOURCE def open(self): qmsg.DataSource.open(self) @@ -85,16 +83,19 @@ class OrderDataSource(qmsg.DataSource): def bind_order(self): return self.bind_pull_socket(self.addresses['order_address']) - def do_work(self): + def do_work(self): #mark the start time for client's processing of this event. self.event_start = datetime.datetime.utcnow() - self.result_socket.send(zp.TRANSFORM_FRAME(str(zp.FINANCE_PROTOCOL.ORDER), self.simulation_dt), self.zmq.NOBLOCK) - self.simulation_dt = self.simulation_dt + self.last_iteration_duration + #TODO: if this is the first iteration, break deadlock by sending a dummy order + if(self.sent_count == 0): + self.send_dummy() + #pull all orders from client. orders = [] order_dt = None + count = 0 while True: (rlist, wlist, xlist) = select([self.order_socket], [], @@ -115,21 +116,35 @@ class OrderDataSource(qmsg.DataSource): 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, source_id=self.get_id}) + order_event = zp.namedict({"sid":sid, "amount":amount, "dt":dt, "source_id":self.get_id, "type":zp.DATASOURCE_TYPE.ORDER}) - message = zp.DATASOURCE_FRAME(event) - self.data_socket.send(message) + self.send(order_event) + 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.sent_count += 1 + + def send(self, order_event): + message = zp.DATASOURCE_FRAME(order_event) + self.data_socket.send(message) + + def send_dummy(self): + dt = self.simulation_dt + self.last_iteration_duration + dummy_order = zp.namedict({"sid":0, "amount":0, "dt":dt, "source_id":self.get_id, "type":zp.DATASOURCE_TYPE.ORDER}) + self.send(dummy_order) class TransactionSimulator(qmsg.BaseTransform): def __init__(self): - qmsg.BaseTransform.__init__(self, "TRANSACTION_SIM") + qmsg.BaseTransform.__init__(self, zp.TRANSFORM_TYPE.TRANSACTION) self.open_orders = {} self.order_count = 0 - self.tradeWindow = datetime.timedelta(seconds=30) + self.trade_windwo = datetime.timedelta(seconds=30) self.orderTTL = datetime.timedelta(days=1) self.volume_share = 0.05 self.commission = 0.03 @@ -139,12 +154,14 @@ class TransactionSimulator(qmsg.BaseTransform): Pulls one message from the event feed, then loops on orders until client sends DONE message. """ - if(event.type == zp.FINANCE_PROTOCOL.ORDER): + #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.average - elif(event.type == zp.FINANCE_PROTOCOL.TRADE): + self.state['value'] = self.create_transaction(0, 0, 0.0, event.dt, 1) + elif(event.type == zp.DATASOURCE_TYPE.TRADE): txn = apply_trade_to_open_orders(event) self.state['value'] = txn + #TODO: what to do if we get another kind of datasource event.type? return self.state @@ -155,17 +172,16 @@ class TransactionSimulator(qmsg.BaseTransform): """ amount = int(amount) if amount == 0: - qutil.LOGGER.debug("{title}:{id} requested to trade zero shares of {sid}".format(sid=sid, - title=self.hostedAlgo.algo.title, - id=self.hostedAlgo.algo.id)) + qutil.LOGGER.debug("requested to trade zero shares of {sid}".format(sid=sid)) return self.order_count += 1 order = zp.namedict({'sid' : sid, 'amount' : amount, - 'dt' : self.algo_time}, + 'dt' : self.algo_time, 'filled': 0, - 'direction': math.fabs(amount) / amount) + 'direction': math.fabs(amount) / amount + }) if(not self.open_orders.has_key(sid)): self.open_orders[sid] = [] @@ -175,7 +191,7 @@ class TransactionSimulator(qmsg.BaseTransform): if(event.volume == 0): #there are zero volume events bc some stocks trade less frequently than once per minute. - continue + return if self.open_orders.has_key(event.sid): orders = self.open_orders[event.sid] remaining_orders = [] @@ -184,7 +200,7 @@ class TransactionSimulator(qmsg.BaseTransform): for order in orders: #we're using minute bars, so allow orders within 30 seconds of the trade - if((order.dt - event.dt) < self.tradeWindow): + if((order.dt - event.dt) < self.trade_windwo): total_order += order.amount if(order.dt > dt): dt = order.dt @@ -206,16 +222,14 @@ class TransactionSimulator(qmsg.BaseTransform): def create_transaction(self, sid, amount, price, dt, direction): - if(amount != 0): - txn = {'sid' : sid, - 'amount' : amount, - 'dt' : dt, - 'price' : price, - 'back_test_run_id' : self.btRun.id, - 'transaction_cost' : -1*(price * amount), - 'commision' : self.commission * amount * direction} - - return namedict(txn) + txn = {'sid' : sid, + 'amount' : amount, + 'dt' : dt, + 'price' : price, + 'commission' : self.commission * amount * direction, + 'source_id' : zp.FINANCE_COMPONENT.TRANSACTION_SIM + } + return zp.namedict(txn) diff --git a/zipline/messaging.py b/zipline/messaging.py index 70268774..7e8565e6 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -409,8 +409,9 @@ class PassthroughTransform(BaseTransform): if message == str(CONTROL_PROTOCOL.DONE): self.signal_done() return - #message is already FEED_FRAMEd, send it as the value. - self.result_socket.send(zp.TRANSFORM_FRAME("PASSTHROUGH", message), self.zmq.NOBLOCK) + + #message is already FEED_FRAMEd, send it as the value. + self.result_socket.send(zp.TRANSFORM_FRAME("PASSTHROUGH", message), self.zmq.NOBLOCK) class DataSource(Component): diff --git a/zipline/protocol.py b/zipline/protocol.py index 22ca9712..fe074840 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -190,9 +190,11 @@ def DATASOURCE_FRAME(event): ::payload:: a msgpack string carrying the payload for the frame """ assert isinstance(event.source_id, basestring) - assert isinstance(event.type, basestring) - if(event.type == "TRADE"): + assert isinstance(event.type, int) + if(event.type == DATASOURCE_TYPE.TRADE): return msgpack.dumps(tuple([event.type, TRADE_FRAME(event)])) + elif(event.type == DATASOURCE_TYPE.ORDER): + return msgpack.dumps(tuple([event.type, ORDER_SOURCE_FRAME(event)])) else: raise INVALID_DATASOURCE_FRAME(str(event)) @@ -211,9 +213,11 @@ def DATASOURCE_UNFRAME(msg): """ try: ds_type, payload = msgpack.loads(msg) - assert isinstance(ds_type, basestring) - if(ds_type == "TRADE"): + assert isinstance(ds_type, int) + if(ds_type == DATASOURCE_TYPE.TRADE): return TRADE_UNFRAME(payload) + elif(ds_type == DATASOURCE_TYPE.ORDER): + return ORDER_SOURCE_UNFRAME(payload) else: raise INVALID_DATASOURCE_FRAME(msg) @@ -266,10 +270,8 @@ def TRANSFORM_FRAME(name, value): """ assert isinstance(name, basestring) assert value != None - - if(name == 'SIM_DT'): - value = PACK_ALGO_DT(value) - + if(name == TRANSFORM_TYPE.TRANSACTION): + value = TRANSACTION_FRAME(value) return msgpack.dumps(tuple([name, value])) def TRANSFORM_UNFRAME(msg): @@ -280,26 +282,16 @@ def TRANSFORM_UNFRAME(msg): name, value = msgpack.loads(msg) #TODO: anything we can do to assert more about the content of the dict? assert isinstance(name, basestring) - if(name == "PASSTHROUGH"): + if(name == TRANSFORM_TYPE.PASSTHROUGH): value = FEED_UNFRAME(value) - elif(name == "SIM_DT"): - value = UNPACK_ALGO_DT(value) + elif(name == TRANSFORM_TYPE.TRANSACTION): + value = TRANSACTION_UNFRAME(value) return namedict({name : value}) except TypeError: raise INVALID_TRANSFORM_FRAME(msg) except ValueError: raise INVALID_TRANSFORM_FRAME(msg) -def PACK_ALGO_DT(value): - value = namedict({'dt' : value}) - PACK_DATE(value) - return value.__dict__ - -def UNPACK_ALGO_DT(value): - value = namedict(value) - UNPACK_DATE(value) - return value.dt - # ================== # Merge Protocol # ================== @@ -312,10 +304,7 @@ def MERGE_FRAME(event): - type """ assert isinstance(event, namedict) - assert isinstance(event.dt, datetime.datetime) PACK_DATE(event) - if(event.has_attr('SIM_DT')): - event.SIM_DT = PACK_ALGO_DT(event.SIM_DT) payload = event.__dict__ return msgpack.dumps(payload) @@ -325,10 +314,6 @@ 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('SIM_DT')): - payload.SIM_DT = UNPACK_ALGO_DT(payload.SIM_DT) - assert isinstance(payload.epoch, numbers.Integral) - assert isinstance(payload.micros, numbers.Integral) UNPACK_DATE(payload) return payload except TypeError: @@ -344,7 +329,7 @@ INVALID_ORDER_FRAME = FrameExceptionFactory('ORDER') INVALID_TRADE_FRAME = FrameExceptionFactory('TRADE') # ================== -# Trades +# Trades - Should only be called from inside DATASOURCE_ (UN)FRAME. # ================== def TRADE_FRAME(event): @@ -358,7 +343,7 @@ def TRADE_FRAME(event): """ assert isinstance(event, namedict) assert isinstance(event.source_id, basestring) - assert event.type == "TRADE" + assert event.type == DATASOURCE_TYPE.TRADE assert isinstance(event.sid, int) assert isinstance(event.price, float) assert isinstance(event.volume, int) @@ -372,8 +357,6 @@ def TRADE_UNFRAME(msg): assert isinstance(sid, int) assert isinstance(price, float) assert isinstance(volume, int) - assert isinstance(epoch, numbers.Integral) - assert isinstance(micros, numbers.Integral) rval = namedict({'sid' : sid, 'price' : price, 'volume' : volume, 'epoch' : epoch, 'micros' : micros, 'type' : source_type, 'source_id' : source_id}) UNPACK_DATE(rval) return rval @@ -383,7 +366,7 @@ def TRADE_UNFRAME(msg): raise INVALID_TRADE_FRAME(msg) # ========= -# Orders +# Orders - from client to order source # ========= def ORDER_FRAME(sid, amount): @@ -404,6 +387,66 @@ def ORDER_UNFRAME(msg): except ValueError: raise INVALID_ORDER_FRAME(msg) +# +# ================== +# TRANSACTIONS - Should only be called from inside TRANSFORM_(UN)FRAME. +# ================== + +def TRANSACTION_FRAME(event): + assert isinstance(event, namedict) + assert isinstance(event.sid, int) + assert isinstance(event.price, float) + assert isinstance(event.commission, float) + assert isinstance(event.amount, int) + PACK_DATE(event) + return msgpack.dumps(tuple([event.sid, event.price, event.amount, event.commission, event.epoch, event.micros])) + +def TRANSACTION_UNFRAME(msg): + try: + sid, price, amount, commission, epoch, micros = msgpack.loads(msg) + + assert isinstance(sid, int) + assert isinstance(price, float) + assert isinstance(commission, float) + assert isinstance(amount, int) + rval = namedict({'sid' : sid, 'price' : price, 'amount' : amount, 'commission':commission, 'epoch' : epoch, 'micros' : micros}) + UNPACK_DATE(rval) + return rval + except TypeError: + raise INVALID_TRADE_FRAME(msg) + except ValueError: + raise INVALID_TRADE_FRAME(msg) + + +# ========= +# Orders - from order source to feed +# - should only be called from inside DATASOURCE_(UN)FRAME +# ========= + +def ORDER_SOURCE_FRAME(event): + assert isinstance(event.sid, int) + assert isinstance(event.amount, int) #no partial shares... + assert isinstance(event.source_id, basestring) + assert event.type == DATASOURCE_TYPE.ORDER + PACK_DATE(event) + return msgpack.dumps(tuple([event.sid, event.amount, event.epoch, event.micros, event.source_id, event.type])) + + +def ORDER_SOURCE_UNFRAME(msg): + try: + sid, amount, epoch, micros, source_id, source_type = msgpack.loads(msg) + event = namedict({"sid":sid, "amount":amount, "epoch":epoch, "micros":micros, "source_id":source_id, "type":source_type}) + assert isinstance(sid, int) + assert isinstance(amount, int) + assert isinstance(source_id, basestring) + assert isinstance(source_type, int) + UNPACK_DATE(event) + return event + except TypeError: + raise INVALID_ORDER_FRAME(msg) + except ValueError: + raise INVALID_ORDER_FRAME(msg) + # ================= # Date Helpers # ================= @@ -428,8 +471,20 @@ def UNPACK_DATE(payload): return payload -FINANCE_PROTOCOL = Enum( - 'ORDER' , # 0 - 'TRANSACTION' , # 1 - 'TRADE' , # 2 - ) +DATASOURCE_TYPE = Enum( + 'ORDER' , + 'TRADE' , +) + +TRANSFORM_TYPE = namedict({ + 'TRANSACTION':'TRANSACTION', #needed? + 'PASSTHROUGH':'PASSTHROUGH' + }) + + +FINANCE_COMPONENT = namedict({ + 'TRADING_CLIENT':'TRADING_CLIENT', + 'PORTFOLIO_CLIENT':'PORTFOLIO_CLIENT', + 'ORDER_SOURCE':'ORDER_SOURCE', + 'TRANSACTION_SIM':'TRANSACTION_SIM' + }) diff --git a/zipline/test/client.py b/zipline/test/client.py index e9649977..02a8e4c1 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -6,12 +6,11 @@ from zipline.finance.trading import TradeSimulationClient import zipline.protocol as zp class TestClient(qmsg.Component): - """no-op client - Just connects to the merge and counts messages. compares received message count to the expected count.""" + """no-op client - Just connects to the merge and counts messages.""" - def __init__(self, expected_msg_count=0): + def __init__(self): qmsg.Component.__init__(self) self.received_count = 0 - self.expected_msg_count = expected_msg_count self.prev_dt = None self.heartbeat_timeout = 2000 @@ -31,9 +30,6 @@ class TestClient(qmsg.Component): if msg == str(zp.CONTROL_PROTOCOL.DONE): qutil.LOGGER.info("Client is DONE!") self.signal_done() - if(self.expected_msg_count > 0): - assert self.received_count == self.expected_msg_count - return self.received_count += 1 @@ -48,14 +44,16 @@ class TestClient(qmsg.Component): class TestTradingClient(TradeSimulationClient): - def __init__(self, count): + def __init__(self, sid, amount, order_count): TradeSimulationClient.__init__(self) - self.count = count + self.count = order_count + self.sid = sid + self.amount = amount self.incr = 0 def handle_events(self, event_queue): #place an order for 100 shares of sid:133 if(self.incr < self.count): - self.order(133, 100) + self.order(self.sid, self.amount) self.incr += 1 diff --git a/zipline/test/factory.py b/zipline/test/factory.py index a30979e5..2a460506 100644 --- a/zipline/test/factory.py +++ b/zipline/test/factory.py @@ -2,67 +2,13 @@ import datetime import pytz import zipline.util as qutil import zipline.finance.risk as risk +import zipline.protocol as zp -def createReturns(daycount, start): - i = 0 - test_range = [] - current = start.replace(tzinfo=pytz.utc) - one_day = datetime.timedelta(days = 1) - while i < daycount: - i += 1 - r = daily_return(current, random.random()) - test_range.append(r) - current = current + one_day - return [ x for x in test_range if(risk.trading_calendar.is_trading_day(x.date)) ] - -def createReturnsFromRange(start, end): - current = start.replace(tzinfo=pytz.utc) - end = end.replace(tzinfo=pytz.utc) - one_day = datetime.timedelta(days = 1) - test_range = [] - i = 0 - while current <= end: - current = current + one_day - if(not risk.trading_calendar.is_trading_day(current)): - continue - r = daily_return(current, random.random()) - i += 1 - test_range.append(r) - - return test_range -def createReturnsFromList(returns, start): - current = start.replace(tzinfo=pytz.utc) - one_day = datetime.timedelta(days = 1) - test_range = [] - i = 0 - while len(test_range) < len(returns): - if(risk.trading_calendar.is_trading_day(current)): - r = daily_return(current, returns[i]) - i += 1 - test_range.append(r) - current = current + one_day - return test_range - - -def createAlgo(filename): - algo = Algorithm() - algo.code = getCodeFromFile(filename) - algo.title = filename - algo._id = pymongo.objectid.ObjectId() - hostedAlgo = HostedAlgorithm(algo) - return hostedAlgo - -def getCodeFromFile(filename): - rVal = None - with open('./test/algo_samples/' + filename, 'r') as f: - rVal = f.read() - return rVal - - + def create_trade(sid, price, amount, datetime): row = {} row['source_id'] = "test_factory" - row['type'] = "TRADE" + row['type'] = zp.DATASOURCE_TYPE.TRADE row['sid'] = sid row['dt'] = datetime row['price'] = price diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 5bb3fdfa..c95b2b52 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -13,7 +13,7 @@ import zipline.protocol as zp from zipline.test.client import TestTradingClient from zipline.test.dummy import ThreadPoolExecutorMixin from zipline.sources import SpecificEquityTrades -from zipline.finance.trading import TradeSimulator +from zipline.finance.trading import TransactionSimulator, OrderDataSource class FinanceTestCase(ThreadPoolExecutorMixin, TestCase): @@ -36,7 +36,7 @@ class FinanceTestCase(ThreadPoolExecutorMixin, TestCase): #do a transform trans_msg = zp.TRANSFORM_FRAME('helloworld', 2345.6) #simulate passthrough transform -- passthrough shouldn't even unpack the msg, just resend. - passthrough_msg = zp.TRANSFORM_FRAME('PASSTHROUGH', feed_msg) + passthrough_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.PASSTHROUGH, feed_msg) #merge unframes transform and passthrough trans_recovered = zp.TRANSFORM_UNFRAME(trans_msg) pt_recovered = zp.TRANSFORM_UNFRAME(passthrough_msg) @@ -54,11 +54,48 @@ class FinanceTestCase(ThreadPoolExecutorMixin, TestCase): self.assertEqual(zp.namedict(trade), event) def test_order_protocol(self): + #client places an order order_msg = zp.ORDER_FRAME(133, 100) + + #order datasource receives sid, amount = zp.ORDER_UNFRAME(order_msg) self.assertEqual(sid, 133) self.assertEqual(amount, 100) - + + #order datasource datasource frames the order + order_dt = datetime.datetime.utcnow().replace(tzinfo=pytz.utc) + order_event = zp.namedict({"sid" : sid, + "amount" : amount, + "dt" : order_dt, + "source_id" : zp.FINANCE_COMPONENT.ORDER_SOURCE, + "type" : zp.DATASOURCE_TYPE.ORDER + }) + + order_ds_msg = zp.DATASOURCE_FRAME(order_event) + + #transaction transform unframes + recovered_order = zp.DATASOURCE_UNFRAME(order_ds_msg) + + self.assertEqual(order_dt, recovered_order.dt) + + #create a transaction from the order + txn = zp.namedict({ + 'sid' : recovered_order.sid, + 'amount' : recovered_order.amount, + 'dt' : recovered_order.dt, + 'price' : 10.0, + 'commission' : 0.50 + }) + + #frame that transaction + txn_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.TRANSACTION, txn) + + #unframe + recovered_tx = zp.TRANSFORM_UNFRAME(txn_msg).TRANSACTION + self.assertEqual(recovered_tx.sid, 133) + self.assertEqual(recovered_tx.amount, 100) + + def test_trading_calendar(self): known_trading_day = datetime.datetime.strptime("02/24/2012","%m/%d/%Y") known_holiday = datetime.datetime.strptime("02/20/2012", "%m/%d/%Y") #president's day @@ -95,10 +132,12 @@ class FinanceTestCase(ThreadPoolExecutorMixin, TestCase): [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 = TestTradingClient(10) - order_sim = TradeSimulator(expected_orders=10) - - sim.register_components([client, order_sim, set1]) + #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)) + transaction_sim = TransactionSimulator() + + sim.register_components([client, order_source, transaction_sim, set1]) sim.register_controller( con ) # Simulation From 7846a1e1f49f3754f137617c99e6800438ef70f3 Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 1 Mar 2012 10:46:50 -0500 Subject: [PATCH 04/16] 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)) From 2aa4af86aa170a067c0466d12afa261e2c02bb30 Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 1 Mar 2012 23:44:35 -0500 Subject: [PATCH 05/16] merged master branch that contains control component, error handling with finance branch. --- zipline/finance/trading.py | 12 ++--- zipline/messaging.py | 92 ++++++++++++++++++++++++------------ zipline/protocol.py | 8 ++-- zipline/simulator.py | 85 +++++++++++++++++++++++++++++++++ zipline/test/client.py | 15 +++--- zipline/test/test_finance.py | 9 +--- 6 files changed, 165 insertions(+), 56 deletions(-) create mode 100644 zipline/simulator.py diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 9420eb96..2ecdc197 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -1,4 +1,3 @@ -import json import datetime import pytz import math @@ -76,8 +75,9 @@ class OrderDataSource(qmsg.DataSource): self.last_iteration_duration = datetime.timedelta(seconds=0) self.sent_count = 0 + @property def get_type(self): - return zp.FINANCE_COMPONENT.ORDER_SOURCE + return zp.DATASOURCE_TYPE.ORDER def open(self): qmsg.DataSource.open(self) @@ -123,7 +123,7 @@ class OrderDataSource(qmsg.DataSource): 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, "source_id":self.get_id, "type":zp.DATASOURCE_TYPE.ORDER}) + order_event = zp.namedict({"sid":sid, "amount":amount, "dt":dt}) self.send(order_event) count += 1 @@ -133,14 +133,10 @@ class OrderDataSource(qmsg.DataSource): if(count == 0): self.send_dummy() self.sent_count += 1 - - def send(self, order_event): - message = zp.DATASOURCE_FRAME(order_event) - self.data_socket.send(message) def send_dummy(self): dt = self.simulation_dt + self.last_iteration_duration - dummy_order = zp.namedict({"sid":0, "amount":0, "dt":dt, "source_id":self.get_id, "type":zp.DATASOURCE_TYPE.ORDER}) + dummy_order = zp.namedict({"sid":0, "amount":0, "dt":dt}) self.send(dummy_order) diff --git a/zipline/messaging.py b/zipline/messaging.py index 5be276b8..79296b1e 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -3,10 +3,10 @@ Commonly used messaging components. """ import datetime -import ujson as json import zipline.util as qutil from zipline.component import Component +import zipline.protocol as zp from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \ COMPONENT_STATE @@ -220,20 +220,27 @@ class ParallelBuffer(Component): self.signal_done() else: try: - event = json.loads(message) - - # JSON deserialization error - except ValueError as exc: + event = self.unframe(message) + # deserialization error + except zp.InvalidFrame as exc: return self.signal_exception(exc) try: - self.append(event[u'id'], event) + self.append(event) self.send_next() # Invalid message - except KeyError as exc: + except zp.InvalidFrame as exc: return self.signal_exception(exc) + # + def unframe(self, msg): + return zp.DATASOURCE_UNFRAME(msg) + + def frame(self, event): + return zp.FEED_FRAME(event) + + # ------------- # Flow Control # ------------- @@ -254,17 +261,16 @@ class ParallelBuffer(Component): return event = self.next() - - if event != None: - self.feed_socket.send(json.dumps(event), self.zmq.NOBLOCK) + if(event != None): + self.feed_socket.send(self.frame(event), self.zmq.NOBLOCK) self.sent_count += 1 - def append(self, source_id, value): + def append(self, event): """ Add an event to the buffer for the source specified by source_id. """ - self.data_buffer[source_id].append(value) + self.data_buffer[event.source_id].append(event) self.received_count += 1 def next(self): @@ -280,7 +286,7 @@ class ParallelBuffer(Component): if len(events) == 0: continue cur = events - if (earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']): + if (earliest == None) or (cur[0].dt <= earliest[0].dt): earliest = cur if earliest != None: @@ -350,15 +356,32 @@ class MergedParallelBuffer(ParallelBuffer): if(not(self.is_full() or self.draining)): return + # #get the raw event from the passthrough transform. - result = self.data_buffer["PASSTHROUGH"].pop(0)['value'] + result = self.data_buffer[zp.TRANSFORM_TYPE.PASSTHROUGH].pop(0).PASSTHROUGH for source, events in self.data_buffer.iteritems(): - if source == "PASSTHROUGH": + if source == zp.TRANSFORM_TYPE.PASSTHROUGH: continue if len(events) > 0: cur = events.pop(0) - result[source] = cur['value'] + result.merge(cur) return result + + def unframe(self, msg): + return zp.TRANSFORM_UNFRAME(msg) + + def frame(self, event): + return zp.MERGE_FRAME(event) + + def append(self, event): + """ + :param event: a namedict with one entry. key is the name of the transform, value is the transformed value. + Add an event to the buffer for the source specified by + source_id. + """ + + self.data_buffer[event.__dict__.keys()[0]].append(event) + self.received_count += 1 class BaseTransform(Component): @@ -425,14 +448,12 @@ class BaseTransform(Component): return try: - event = json.loads(message) - except ValueError as exc: + event = self.unframe(message) + except zp.InvalidFrame as exc: return self.signal_exception(exc) try: cur_state = self.transform(event) - cur_state['dt'] = event['dt'] - cur_state['id'] = self.state['name'] # This is overloaded, so it can fail in all sorts of # unknown ways. Its best to catch it in the @@ -441,12 +462,18 @@ class BaseTransform(Component): return self.signal_exception(exc) try: - json_frame = json.dumps(cur_state) - except ValueError as exc: + transform_frame = self.frame(cur_state) + except zp.InvalidFrame as exc: return self.signal_exception(exc) - self.result_socket.send(json_frame, self.zmq.NOBLOCK) - + self.result_socket.send(transform_frame, self.zmq.NOBLOCK) + + def frame(self, cur_state): + return zp.TRANSFORM_FRAME(cur_state['name'], cur_state['value']) + + def unframe(self, msg): + return zp.FEED_UNFRAME(msg) + def transform(self, event): """ Must return the transformed value as a map with:: @@ -487,7 +514,7 @@ class PassthroughTransform(BaseTransform): #TODO, could save some cycles by skipping the _UNFRAME call and just setting value to original msg string. def transform(self, event): - return {'name':zp.TRANSFORM_TYPE.PASSTHROUGH, 'value': zp.DATASOURCE_FRAME(event) } + return {'name':zp.TRANSFORM_TYPE.PASSTHROUGH, 'value': zp.FEED_FRAME(event) } class DataSource(Component): @@ -520,14 +547,17 @@ class DataSource(Component): """ Emit data. """ - assert isinstance(event, dict) + assert isinstance(event, zp.namedict) - event['id'] = self.id - event['type'] = self.get_type() + event.__dict__['source_id'] = self.get_id + event.__dict__['type'] = self.get_type try: - json_frame = json.dumps(event) - except ValueError as exc: + ds_frame = self.frame(event) + except zp.InvalidFrame as exc: return self.signal_exception(exc) - self.data_socket.send(json_frame) + self.data_socket.send(ds_frame) + + def frame(self, event): + return zp.DATASOURCE_FRAME(event) diff --git a/zipline/protocol.py b/zipline/protocol.py index 5e007715..24a9a243 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -61,13 +61,15 @@ def Enum(*options): _fields_ = [(o, c_ubyte) for o in options] return cstruct(*range(len(options))) +class InvalidFrame(Exception): + def __init__(self, got): + self.got = got + def FrameExceptionFactory(name): """ Exception factory with a closure around the frame class name. """ - class InvalidFrame(Exception): - def __init__(self, got): - self.got = got + class NamedInvalidFrame(InvalidFrame): def __str__(self): return "Invalid {framecls} Frame: {got}".format( framecls = name, diff --git a/zipline/simulator.py b/zipline/simulator.py new file mode 100644 index 00000000..1e3e3f5f --- /dev/null +++ b/zipline/simulator.py @@ -0,0 +1,85 @@ +""" +Simulator hosts all the components necessary to execute a simluation. See :py:method"" +""" + +import threading +import mock +from collections import defaultdict +from zipline.monitor import Controller +from zipline.messaging import ComponentHost +import zipline.util as qutil + +class AddressAllocator(object): + + def __init__(self, ns): + self.idx = 0 + self.sockets = [ + 'tcp://127.0.0.1:%s' % (10000 + n) + for n in xrange(ns) + ] + + def lease(self, n): + sockets = self.sockets[self.idx:self.idx+n] + self.idx += n + return sockets + + def reaquire(self, *conn): + pass + +# +class Simulator(ComponentHost): + + def __init__(self, addresses): + ComponentHost.__init__(self, addresses) + self.subthreads = [] + self.running = False + + def launch_controller(self): + thread = threading.Thread(target=self.controller.run) + thread.start() + + self.subthreads.append(thread) + return thread + + def simulate(self): + thread = threading.Thread(target=self.run) + thread.start() + + self.subthreads.append(thread) + self.running = True + + return thread + + def did_clean_shutdown(self): + return not any([t.isAlive() for t in self.subthreads]) + + def shutdown(self): + """ + Destroy all tracked components. + """ + + if not self.running: + return + + try: + self.controller.shutdown(context=self.context) + except: + import pdb; pdb.set_trace() + + for component in self.components.itervalues(): + component.shutdown() + + for thread in self.subthreads: + if thread.is_alive(): + thread._Thread__stop() + + self.running = False + + assert self.did_clean_shutdown() + + def launch_component(self, component): + thread = threading.Thread(target=component.run) + thread.start() + + self.subthreads.append(thread) + return thread diff --git a/zipline/test/client.py b/zipline/test/client.py index 20bb8df7..2b1ac006 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -1,6 +1,3 @@ -#import logging -import ujson as json - import zipline.util as qutil import zipline.messaging as qmsg from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE @@ -45,10 +42,10 @@ class TestClient(qmsg.Component): self.received_count += 1 try: - event = json.loads(msg) + event = self.unframe(msg) - # JSON deserialization error - except ValueError as exc: + # deserialization error + except zp.InvalidFrame as exc: return self.signal_exception(exc) if self.prev_dt != None: @@ -59,10 +56,14 @@ class TestClient(qmsg.Component): ) ) else: - self.prev_dt = event['dt'] + self.prev_dt = event.dt if self.received_count % 100 == 0: qutil.LOGGER.info("received {n} messages".format(n=self.received_count)) + + def unframe(self, msg): + return zp.MERGE_UNFRARME(msg) + class TestTradingClient(TradeSimulationClient): diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 0a9869e0..d1f35271 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -157,13 +157,8 @@ class FinanceTestCase(TestCase): # ---------- sim.simulate().join() - # Stop Running - # ------------ - - # TODO: less abrupt later, just shove a StopIteration - # down the pipe to make it stop spinning - sim.cuc._Thread__stop() - + + # TODO: Make more assertions about the final state of the components. self.assertEqual(sim.feed.pending_messages(), 0, "The feed should be drained of all messages, found {n} remaining." .format(n=sim.feed.pending_messages()) From 12be6730113d054d615a91733bdbd89a1f82d418 Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 2 Mar 2012 13:10:06 -0500 Subject: [PATCH 06/16] switched to using the subclass of InvalidFrame in the messaging module exception handling. --- zipline/messaging.py | 10 +++++----- zipline/protocol.py | 9 ++++----- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/zipline/messaging.py b/zipline/messaging.py index 79296b1e..65ef93b0 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -222,7 +222,7 @@ class ParallelBuffer(Component): try: event = self.unframe(message) # deserialization error - except zp.InvalidFrame as exc: + except zp.INVALID_DATASOURE_FRAME as exc: return self.signal_exception(exc) try: @@ -230,7 +230,7 @@ class ParallelBuffer(Component): self.send_next() # Invalid message - except zp.InvalidFrame as exc: + except zp.INVALID_DATASOURCE_FRAME as exc: return self.signal_exception(exc) # @@ -449,7 +449,7 @@ class BaseTransform(Component): try: event = self.unframe(message) - except zp.InvalidFrame as exc: + except zp.INVALID_FEED_FRAME as exc: return self.signal_exception(exc) try: @@ -463,7 +463,7 @@ class BaseTransform(Component): try: transform_frame = self.frame(cur_state) - except zp.InvalidFrame as exc: + except zp.INVALID_TRANSFORM_FRAME as exc: return self.signal_exception(exc) self.result_socket.send(transform_frame, self.zmq.NOBLOCK) @@ -554,7 +554,7 @@ class DataSource(Component): try: ds_frame = self.frame(event) - except zp.InvalidFrame as exc: + except zp.INVALID_DATASOURCE_FRAME as exc: return self.signal_exception(exc) self.data_socket.send(ds_frame) diff --git a/zipline/protocol.py b/zipline/protocol.py index 24a9a243..e7ffed1a 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -61,15 +61,14 @@ def Enum(*options): _fields_ = [(o, c_ubyte) for o in options] return cstruct(*range(len(options))) -class InvalidFrame(Exception): - def __init__(self, got): - self.got = got - def FrameExceptionFactory(name): """ Exception factory with a closure around the frame class name. """ - class NamedInvalidFrame(InvalidFrame): + class InvalidFrame(Exception): + def __init__(self, got): + self.got = got + def __str__(self): return "Invalid {framecls} Frame: {got}".format( framecls = name, From 5c0bb6acaedb08ddf739733253cdb62ee5ab81b5 Mon Sep 17 00:00:00 2001 From: fawce Date: Sat, 3 Mar 2012 11:45:04 -0500 Subject: [PATCH 07/16] updated namedict based on sdiehl's feedback, modified ParallelBuffer.get_next() method to be more readable. --- zipline/messaging.py | 25 +++++++++++++++---------- zipline/protocol.py | 25 +++++++++++++++++++------ zipline/test/test_finance.py | 2 +- 3 files changed, 35 insertions(+), 17 deletions(-) diff --git a/zipline/messaging.py b/zipline/messaging.py index 65ef93b0..fbeadd90 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -280,17 +280,22 @@ class ParallelBuffer(Component): if not(self.is_full() or self.draining): return - cur = None - earliest = None + cur_source = None + earliest_source = None + earliest_event = None + #iterate over the queues of events from all sources (1 queue per datasource) for events in self.data_buffer.values(): if len(events) == 0: continue - cur = events - if (earliest == None) or (cur[0].dt <= earliest[0].dt): - earliest = cur + cur_source = events + first_in_list = events[0] + + if (earliest_event == None) or (first_in_list.dt <= earliest_event.dt): + earliest_event = first_in_list + earliest_source = cur_source - if earliest != None: - return earliest.pop(0) + if earliest_event != None: + return earliest_source.pop(0) def is_full(self): """ @@ -380,7 +385,7 @@ class MergedParallelBuffer(ParallelBuffer): source_id. """ - self.data_buffer[event.__dict__.keys()[0]].append(event) + self.data_buffer[event.keys()[0]].append(event) self.received_count += 1 @@ -549,8 +554,8 @@ class DataSource(Component): """ assert isinstance(event, zp.namedict) - event.__dict__['source_id'] = self.get_id - event.__dict__['type'] = self.get_type + event['source_id'] = self.get_id + event['type'] = self.get_type try: ds_frame = self.frame(event) diff --git a/zipline/protocol.py b/zipline/protocol.py index e7ffed1a..090a28cd 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -98,6 +98,19 @@ class namedict(object): self.__dict__['id'] = value else: self.__dict__[key] = value + + def __getitem__(self, key): + return self.__dict__[key] + + def keys(self): + return self.__dict__.keys() + + def as_dict(self): + #TODO: make a copy? + return self.__dict__ + + def delete(self, key): + del(self.__dict__[key]) def merge(self, other_nd): assert isinstance(other_nd, namedict) @@ -248,7 +261,7 @@ def FEED_FRAME(event): source_id = event.source_id ds_type = event.type PACK_DATE(event) - payload = event.__dict__ + payload = event.as_dict() return msgpack.dumps(payload) def FEED_UNFRAME(msg): @@ -317,7 +330,7 @@ def MERGE_FRAME(event): event.TRANSACTION = TRANSFORM_TYPE.EMPTY else: event.TRANSACTION = TRANSACTION_FRAME(event.TRANSACTION) - payload = event.__dict__ + payload = event.as_dict() return msgpack.dumps(payload) def MERGE_UNFRAME(msg): @@ -474,7 +487,7 @@ def PACK_DATE(event): epoch = long(event.dt.strftime('%s')) event['epoch'] = epoch event['micros'] = event.dt.microsecond - del(event.__dict__['dt']) + event.delete('dt') return event def UNPACK_DATE(payload): @@ -482,9 +495,9 @@ def UNPACK_DATE(payload): assert isinstance(payload.micros, numbers.Integral) dt = datetime.datetime.fromtimestamp(payload.epoch) dt = dt.replace(microsecond = payload.micros, tzinfo = pytz.utc) - del(payload.__dict__['epoch']) - del(payload.__dict__['micros']) - payload['dt'] = dt + payload.delete('epoch') + payload.delete('micros') + payload.dt = dt return payload DATASOURCE_TYPE = Enum( diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index d1f35271..60f88ad3 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -53,7 +53,7 @@ class FinanceTestCase(TestCase): #check the transformed value, should only be in event, not trade. self.assertTrue(event.helloworld == 2345.6) - del(event.__dict__['helloworld']) + event.delete('helloworld') self.assertEqual(zp.namedict(trade), event) From 2faf1736b1af6457e257e0c13b7cc0e86639bc15 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Sat, 3 Mar 2012 12:53:58 -0500 Subject: [PATCH 08/16] Documentation & best practices documentation. --- zipline/protocol.py | 80 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 78 insertions(+), 2 deletions(-) diff --git a/zipline/protocol.py b/zipline/protocol.py index 090a28cd..39d6deed 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -39,12 +39,88 @@ that anything outside of UTF8 can cause serious problems, so if you have a strong desire to JSON encode ancient Sanskrit ( admit it, we all do ), just say no. +Data Structures +=============== + +Enum +---- + +Classic C style enumeration:: + + opts = Enum('FOO', 'BAR') + + opts.FOO # 0 + opts.BAR # 1 + opts.FOO = opts.BAR # False + +Oh, and if you do this:: + + protocol.Enum([1,2,3]) + +Your interpreter will segfault, think of this like an extreme assert. + +Namedict +-------- + +Namedicts are dict like objects that have fields accessible by attribute lookup +as well as being indexable and iterable:: + + HEARTBEAT_PROTOCOL = namedict({ + 'REQ' : b'\x01', + 'REP' : b'\x02', + }) + + HEARTBEAT_PROTOCOL.REQ # syntatic sugar + HEARTBEAT_PROTOCOL.REP # oh suga suga + + HEARTBEAT_PROTOCOL['REQ'] # classic dictionary index + +Nmaedtuple +---------- + +From the standard library, namedtuples are great for specifing +containers for spec'ing data container objects:: + + from collections import namedtuple + + Person = namedtuple('Person', 'name age gender') + bob = Person(name='Bob', age=30, gender='male') + + bob.name # 'Bob' + bob.age # 30 + bob.gender # male + + # The slots on the tuple are also finite and read-only. This + # is a good thing, keeps us honest! + + bob.hobby = 'underwater archery' + # Will raise: + # AttributeError: 'Person' object has no attribute 'hobby' + + bob.name = 'joe' + # Will raise: + # AttributeError: can't set attribute + + # Namedtuples are normally read-only, but you can change the + # internals using a private operation. + bob._replace(gender='female') + + # You can also dump out to dictionary form: + OrderedDict([('name', 'Bob'), ('age', 30), ('gender', 'male')]) + + # Or JSON. + json.dumps(bob._asdict()) + '{"gender":"male","age":30,"name":"Bob"}' + """ import msgpack import numbers import datetime import pytz +import copy +from collections import namedtuple + import zipline.util as qutil #import ujson #import ultrajson_numpy @@ -106,8 +182,8 @@ class namedict(object): return self.__dict__.keys() def as_dict(self): - #TODO: make a copy? - return self.__dict__ + # shallow copy is O(n) + return copy.copy(self.__dict__) def delete(self, key): del(self.__dict__[key]) From 47aee5c13dc97968f9a4a690168631478657db47 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Sat, 3 Mar 2012 13:06:16 -0500 Subject: [PATCH 09/16] I spellz good. --- zipline/protocol.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/zipline/protocol.py b/zipline/protocol.py index 39d6deed..38b75e36 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -10,7 +10,7 @@ Notes Msgpack ------- -Msgpack is the fastest seriaization protocol in Python at the +Msgpack is the fastest serialization protocol in Python at the moment. Its 100% C is typically orders of magnitude faster than json and pickle making it awesome for ZeroMQ. @@ -70,15 +70,15 @@ as well as being indexable and iterable:: 'REP' : b'\x02', }) - HEARTBEAT_PROTOCOL.REQ # syntatic sugar + HEARTBEAT_PROTOCOL.REQ # syntactic sugar HEARTBEAT_PROTOCOL.REP # oh suga suga HEARTBEAT_PROTOCOL['REQ'] # classic dictionary index -Nmaedtuple +Namedtuple ---------- -From the standard library, namedtuples are great for specifing +From the standard library, namedtuples are great for specifying containers for spec'ing data container objects:: from collections import namedtuple From 6c581bef3989d68e7ab88d05c8d659fc270e037a Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Sat, 3 Mar 2012 21:28:41 -0500 Subject: [PATCH 10/16] Fix PEP8 warnings. --- zipline/protocol.py | 186 +++++++++++++++++++++++++++++--------------- 1 file changed, 123 insertions(+), 63 deletions(-) diff --git a/zipline/protocol.py b/zipline/protocol.py index 38b75e36..9631cfc9 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -62,14 +62,14 @@ Your interpreter will segfault, think of this like an extreme assert. Namedict -------- -Namedicts are dict like objects that have fields accessible by attribute lookup +Namedicts are dict like objects that have fields accessible by attribute lookup as well as being indexable and iterable:: HEARTBEAT_PROTOCOL = namedict({ 'REQ' : b'\x01', 'REP' : b'\x02', }) - + HEARTBEAT_PROTOCOL.REQ # syntactic sugar HEARTBEAT_PROTOCOL.REP # oh suga suga @@ -144,13 +144,13 @@ def FrameExceptionFactory(name): class InvalidFrame(Exception): def __init__(self, got): self.got = got - + def __str__(self): return "Invalid {framecls} Frame: {got}".format( framecls = name, got = self.got, ) - + return InvalidFrame class namedict(object): @@ -167,37 +167,42 @@ class namedict(object): def __init__(self, dct=None): if(dct): self.__dict__.update(dct) - + def __setitem__(self, key, value): - """Required for use by pymongo as_class parameter to find.""" + """ + Required for use by pymongo as_class parameter to find. + """ if(key == '_id'): self.__dict__['id'] = value else: self.__dict__[key] = value - + def __getitem__(self, key): return self.__dict__[key] - + def keys(self): return self.__dict__.keys() - + def as_dict(self): # shallow copy is O(n) return copy.copy(self.__dict__) - + def delete(self, key): del(self.__dict__[key]) - + def merge(self, other_nd): assert isinstance(other_nd, namedict) self.__dict__.update(other_nd.__dict__) - + def __repr__(self): return "namedict: " + str(self.__dict__) - + def __eq__(self, other): + # !!!!!!!!!!!!!!!!!!!! + # !!!! DANGEROUS !!!!! + # !!!!!!!!!!!!!!!!!!!! return other != None and self.__dict__ == other.__dict__ - + def has_attr(self, name): return self.__dict__.has_key(name) @@ -277,36 +282,44 @@ INVALID_DATASOURCE_FRAME = FrameExceptionFactory('DATASOURCE') def DATASOURCE_FRAME(event): """ - wraps any datasource payload with id and type, so that unpacking may choose the write - UNFRAME for the payload. - ::ds_id:: an identifier that is unique to the datasource in the context of a component host (e.g. Simulator + Wraps any datasource payload with id and type, so that unpacking may choose + the write UNFRAME for the payload. + + ::ds_id:: an identifier that is unique to the datasource in the context of + a component host (e.g. Simulator ::ds_type:: a string denoting the datasource type. Must be on of:: TRADE (others to follow soon) ::payload:: a msgpack string carrying the payload for the frame """ + assert isinstance(event.source_id, basestring) - assert isinstance(event.type, int) + assert isinstance(event.type, int), 'Unexpected type %s' % (event.type) + if(event.type == DATASOURCE_TYPE.TRADE): return msgpack.dumps(tuple([event.type, TRADE_FRAME(event)])) elif(event.type == DATASOURCE_TYPE.ORDER): return msgpack.dumps(tuple([event.type, ORDER_SOURCE_FRAME(event)])) else: raise INVALID_DATASOURCE_FRAME(str(event)) - + def DATASOURCE_UNFRAME(msg): """ - extracts payload, and calls correct UNFRAME method based on the datasource type passed along + Extracts payload, and calls correct UNFRAME method based on the datasource + type passed along. + returns a dict containing at least:: - source_id - type + other properties are added based on the datasource type:: - TRADE:: - sid - int security identifier - price - float - - volume - int + - volume - int - dt - a datetime object """ + try: ds_type, payload = msgpack.loads(msg) assert isinstance(ds_type, int) @@ -316,12 +329,12 @@ def DATASOURCE_UNFRAME(msg): return ORDER_SOURCE_UNFRAME(payload) else: raise INVALID_DATASOURCE_FRAME(msg) - + except TypeError: raise INVALID_DATASOURCE_FRAME(msg) except ValueError: raise INVALID_DATASOURCE_FRAME(msg) - + # ================== # Feed Protocol # ================== @@ -330,7 +343,7 @@ INVALID_FEED_FRAME = FrameExceptionFactory('FEED') def FEED_FRAME(event): """ :event: a nameddict with at least:: - - source_id + - source_id - type """ assert isinstance(event, namedict) @@ -339,7 +352,7 @@ def FEED_FRAME(event): PACK_DATE(event) payload = event.as_dict() return msgpack.dumps(payload) - + def FEED_UNFRAME(msg): try: payload = msgpack.loads(msg) @@ -352,7 +365,7 @@ def FEED_UNFRAME(msg): raise INVALID_FEED_FRAME(msg) except ValueError: raise INVALID_FEED_FRAME(msg) - + # ================== # Transform Protocol # ================== @@ -365,13 +378,13 @@ def TRANSFORM_FRAME(name, value): if(name == TRANSFORM_TYPE.TRANSACTION): value = TRANSACTION_FRAME(value) return msgpack.dumps(tuple([name, value])) - + def TRANSFORM_UNFRAME(msg): """ :rtype: namedict with : """ try: - + name, value = msgpack.loads(msg) if(value == TRANSFORM_TYPE.EMPTY): return namedict({name : None}) @@ -381,7 +394,7 @@ def TRANSFORM_UNFRAME(msg): value = FEED_UNFRAME(value) elif(name == TRANSFORM_TYPE.TRANSACTION): value = TRANSACTION_UNFRAME(value) - + return namedict({name : value}) except TypeError: raise INVALID_TRANSFORM_FRAME(msg) @@ -396,7 +409,7 @@ INVALID_MERGE_FRAME = FrameExceptionFactory('MERGE') def MERGE_FRAME(event): """ :event: a nameddict with at least:: - - source_id + - source_id - type """ assert isinstance(event, namedict) @@ -408,7 +421,7 @@ def MERGE_FRAME(event): event.TRANSACTION = TRANSACTION_FRAME(event.TRANSACTION) payload = event.as_dict() return msgpack.dumps(payload) - + def MERGE_UNFRAME(msg): try: payload = msgpack.loads(msg) @@ -427,7 +440,7 @@ def MERGE_UNFRAME(msg): except ValueError: raise INVALID_MERGE_FRAME(msg) - + # ================== # Finance Protocol # ================== @@ -445,7 +458,7 @@ def TRADE_FRAME(event): - price -- float of the price printed for the trade - volume -- int for shares in the trade - dt -- datetime for the trade - + """ assert isinstance(event, namedict) assert isinstance(event.source_id, basestring) @@ -454,16 +467,33 @@ def TRADE_FRAME(event): assert isinstance(event.price, float) assert isinstance(event.volume, int) PACK_DATE(event) - return msgpack.dumps(tuple([event.sid, event.price, event.volume, event.epoch, event.micros, event.type, event.source_id])) - + return msgpack.dumps(tuple([ + event.sid, + event.price, + event.volume, + event.epoch, + event.micros, + event.type, + event.source_id + ])) + def TRADE_UNFRAME(msg): try: - sid, price, volume, epoch, micros, source_type, source_id = msgpack.loads(msg) - + packed = msgpack.loads(msg) + sid, price, volume, epoch, micros, source_type, source_id = packed + assert isinstance(sid, int) assert isinstance(price, float) assert isinstance(volume, int) - rval = namedict({'sid' : sid, 'price' : price, 'volume' : volume, 'epoch' : epoch, 'micros' : micros, 'type' : source_type, 'source_id' : source_id}) + rval = namedict({ + 'sid' : sid, + 'price' : price, + 'volume' : volume, + 'epoch' : epoch, + 'micros' : micros, + 'type' : source_type, + 'source_id' : source_id + }) UNPACK_DATE(rval) return rval except TypeError: @@ -477,9 +507,9 @@ def TRADE_UNFRAME(msg): def ORDER_FRAME(sid, amount): assert isinstance(sid, int) - assert isinstance(amount, int) #no partial shares... + assert isinstance(amount, int) #no partial shares... return msgpack.dumps(tuple([sid, amount])) - + def ORDER_UNFRAME(msg): try: @@ -492,7 +522,7 @@ def ORDER_UNFRAME(msg): raise INVALID_ORDER_FRAME(msg) except ValueError: raise INVALID_ORDER_FRAME(msg) - + # # ================== # TRANSACTIONS - Should only be called from inside TRANSFORM_(UN)FRAME. @@ -505,17 +535,32 @@ def TRANSACTION_FRAME(event): assert isinstance(event.commission, float) assert isinstance(event.amount, int) PACK_DATE(event) - return msgpack.dumps(tuple([event.sid, event.price, event.amount, event.commission, event.epoch, event.micros])) - + return msgpack.dumps(tuple([ + event.sid, + event.price, + event.amount, + event.commission, + event.epoch, + event.micros + ])) + def TRANSACTION_UNFRAME(msg): try: sid, price, amount, commission, epoch, micros = msgpack.loads(msg) - + assert isinstance(sid, int) assert isinstance(price, float) assert isinstance(commission, float) assert isinstance(amount, int) - rval = namedict({'sid' : sid, 'price' : price, 'amount' : amount, 'commission':commission, 'epoch' : epoch, 'micros' : micros}) + rval = namedict({ + 'sid' : sid, + 'price' : price, + 'amount' : amount, + 'commission' : commission, + 'epoch' : epoch, + 'micros' : micros + }) + UNPACK_DATE(rval) return rval except TypeError: @@ -526,22 +571,36 @@ def TRANSACTION_UNFRAME(msg): # ========= # Orders - from order source to feed -# - should only be called from inside DATASOURCE_(UN)FRAME +# - should only be called from inside DATASOURCE_(UN)FRAME # ========= def ORDER_SOURCE_FRAME(event): assert isinstance(event.sid, int) - assert isinstance(event.amount, int) #no partial shares... + assert isinstance(event.amount, int) #no partial shares... assert isinstance(event.source_id, basestring) assert event.type == DATASOURCE_TYPE.ORDER PACK_DATE(event) - return msgpack.dumps(tuple([event.sid, event.amount, event.epoch, event.micros, event.source_id, event.type])) - + return msgpack.dumps(tuple([ + event.sid, + event.amount, + event.epoch, + event.micros, + event.source_id, + event.type + ])) + def ORDER_SOURCE_UNFRAME(msg): try: sid, amount, epoch, micros, source_id, source_type = msgpack.loads(msg) - event = namedict({"sid":sid, "amount":amount, "epoch":epoch, "micros":micros, "source_id":source_id, "type":source_type}) + event = namedict({ + "sid" : sid, + "amount" : amount, + "epoch" : epoch, + "micros" : micros, + "source_id" : source_id, + "type" : source_type + }) assert isinstance(sid, int) assert isinstance(amount, int) assert isinstance(source_id, basestring) @@ -552,12 +611,12 @@ def ORDER_SOURCE_UNFRAME(msg): raise INVALID_ORDER_FRAME(msg) except ValueError: raise INVALID_ORDER_FRAME(msg) - + # ================= # Date Helpers # ================= -def PACK_DATE(event): +def PACK_DATE(event): assert isinstance(event.dt, datetime.datetime) assert event.dt.tzinfo == pytz.utc #utc only please epoch = long(event.dt.strftime('%s')) @@ -576,8 +635,9 @@ def UNPACK_DATE(payload): payload.dt = dt return payload + DATASOURCE_TYPE = Enum( - 'ORDER' , + 'ORDER' , 'TRADE' , ) @@ -589,15 +649,15 @@ ORDER_PROTOCOL = Enum( #Transform type needs to be a namedict to facilitate merging. TRANSFORM_TYPE = namedict({ - 'TRANSACTION':'TRANSACTION', #needed? - 'PASSTHROUGH':'PASSTHROUGH', - 'EMPTY':'' - }) - + 'TRANSACTION' : 'TRANSACTION', #needed? + 'PASSTHROUGH' : 'PASSTHROUGH', + 'EMPTY' : '' +}) + FINANCE_COMPONENT = namedict({ - 'TRADING_CLIENT':'TRADING_CLIENT', - 'PORTFOLIO_CLIENT':'PORTFOLIO_CLIENT', - 'ORDER_SOURCE':'ORDER_SOURCE', - 'TRANSACTION_SIM':'TRANSACTION_SIM' - }) + 'TRADING_CLIENT' : 'TRADING_CLIENT', + 'PORTFOLIO_CLIENT' : 'PORTFOLIO_CLIENT', + 'ORDER_SOURCE' : 'ORDER_SOURCE', + 'TRANSACTION_SIM' : 'TRANSACTION_SIM' +}) From 58eba2d100036d1a4583522b67fdfcb4e82303ce Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Sat, 3 Mar 2012 21:30:45 -0500 Subject: [PATCH 11/16] Attempt to fix protocol assertion error. --- zipline/sources.py | 63 +++++++++++++++++++++------------- zipline/test/test_messaging.py | 10 +++--- 2 files changed, 46 insertions(+), 27 deletions(-) diff --git a/zipline/sources.py b/zipline/sources.py index a0ba21e2..a6631c69 100644 --- a/zipline/sources.py +++ b/zipline/sources.py @@ -5,23 +5,28 @@ import datetime import random import pytz -import zipline.util as qutil import zipline.messaging as zm import zipline.protocol as zp + class TradeDataSource(zm.DataSource): - + def send(self, event): - """ :param dict event: is a trade event with data as per :py:func: `zipline.protocol.TRADE_FRAME` - :rtype: None + """ + :param dict event: is a trade event with data as per + :py:func: `zipline.protocol.TRADE_FRAME` + :rtype: None """ event.source_id = self.get_id message = zp.DATASOURCE_FRAME(event) self.data_socket.send(message) + class RandomEquityTrades(TradeDataSource): - """Generates a random stream of trades for testing.""" - + """ + Generates a random stream of trades for testing. + """ + def __init__(self, sid, source_id, count): zm.DataSource.__init__(self, source_id) self.count = count @@ -30,32 +35,44 @@ class RandomEquityTrades(TradeDataSource): self.trade_start = datetime.datetime.now().replace(tzinfo=pytz.utc) self.minute = datetime.timedelta(minutes=1) self.price = random.uniform(5.0, 50.0) - - + + def get_type(self): - return 'EQUITY_TRADE' - - + zp.COMPONENT_TYPE.SOURCE + def do_work(self): if(self.incr == self.count): self.signal_done() return - - self.price = self.price + random.uniform(-0.05, 0.05) - self._send(self.sid, self.price, random.randrange(100,10000,100), self.trade_start + (self.minute * self.incr)) - self.incr += 1 - def _send(self, sid, price, volume, dt): - event = zp.namedict({'source_id': self.get_id, "type" : "TRADE", "sid":sid, "price":price, "volume":volume, "dt":dt}) - self.send(event) + self.price = self.price + random.uniform(-0.05, 0.05) + volume = random.randrange(100,10000,100) + + event = zp.namedict({ + 'source_id' : self.get_id, + "type" : zp.DATASOURCE_TYPE.ORDER, + "sid" : self.sid, + "price" : self.price, + "volume" : volume, + "dt" : self.trade_start + (self.minute * self.incr), + }) + + message = zp.DATASOURCE_FRAME(event) + self.send(message) + + self.incr += 1 class SpecificEquityTrades(TradeDataSource): - """Generates a random stream of trades for testing.""" + """ + Generates a random stream of trades for testing. + """ def __init__(self, source_id, event_list): """ - :event_list: should be a chronologically ordered list of dictionaries in the following form: + :event_list: should be a chronologically ordered list of dictionaries + in the following form: + event = { 'sid' : an integer for security id, 'dt' : datetime object, @@ -67,14 +84,14 @@ class SpecificEquityTrades(TradeDataSource): self.event_list = event_list def get_type(self): - return 'EQUITY_TRADE' + zp.COMPONENT_TYPE.SOURCE def do_work(self): if(len(self.event_list) == 0): self.signal_done() return - + event = self.event_list.pop(0) self.send(zp.namedict(event)) - + diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index afa9a5a3..187b68b8 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -11,6 +11,7 @@ from zipline.sources import RandomEquityTrades from zipline.test.client import TestClient from zipline.test.transform import DivideByZeroTransform +from nose.tools import timed # Should not inherit form TestCase since test runners will pick # it up as a test. Its a Mixin of sorts at this point. @@ -73,6 +74,7 @@ class SimulatorTestCase(object): # Cases # ------- + @timed(2) def test_simple(self): # Simple test just to make sure that the archiecture is @@ -100,7 +102,7 @@ class SimulatorTestCase(object): ret1 = RandomEquityTrades(133, "ret1", 1) ret2 = RandomEquityTrades(134, "ret2", 1) - client = TestClient(expected_msg_count=(ret1.count + ret2.count)) + client = TestClient() sim.register_controller( con ) sim.register_components([ret1, ret2, client]) @@ -149,7 +151,7 @@ class SimulatorTestCase(object): ret1 = RandomEquityTrades(133, "ret1", 1) ret2 = RandomEquityTrades(134, "ret2", 1) fail_transform = DivideByZeroTransform("fail") - client = TestClient(self, expected_msg_count=ret1.count + ret2.count) + client = TestClient() sim.register_controller( con ) sim.register_components([ret1, ret2, fail_transform, client]) @@ -194,7 +196,7 @@ class SimulatorTestCase(object): ret1 = RandomEquityTrades(133, "ret1", 400) ret2 = RandomEquityTrades(134, "ret2", 400) - client = TestClient(expected_msg_count=ret1.count + ret2.count) + client = TestClient() sim.register_controller( con ) sim.register_components([ret1, ret2, client]) @@ -240,7 +242,7 @@ class SimulatorTestCase(object): ret2 = RandomEquityTrades(134, "ret2", 5000) mavg1 = MovingAverage("mavg1", 30) mavg2 = MovingAverage("mavg2", 60) - client = TestClient(expected_msg_count=10000) + client = TestClient() sim.register_components([ret1, ret2, mavg1, mavg2, client]) sim.register_controller( con ) From 5900e5effcd8006b97d85937bd575179fe9271a9 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Sat, 3 Mar 2012 21:47:38 -0500 Subject: [PATCH 12/16] PEP8ify finance tests. --- zipline/test/test_finance.py | 134 ++++++++++++++++++++--------------- 1 file changed, 75 insertions(+), 59 deletions(-) diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 60f88ad3..fe59d90f 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -1,9 +1,10 @@ """Tests for the zipline.finance package""" -import datetime import mock import pytz import host_settings from unittest2 import TestCase +from datetime import datetime, timedelta + import zipline.test.factory as factory import zipline.util as qutil import zipline.db as db @@ -18,16 +19,22 @@ from zipline.monitor import Controller class FinanceTestCase(TestCase): - + def setUp(self): qutil.configure_logging() - + def test_trade_feed_protocol(self): - trades = factory.create_trade_history(133, - [10.0,10.0,10.0,10.0], - [100,100,100,100], - datetime.datetime.strptime("02/15/2012","%m/%d/%Y"), - datetime.timedelta(days=1)) + + # TODO: Perhaps something more self-documenting for variables names? + a = 133 + b = [10] * 4 + c = [100] * 4 + + ts = datetime.strptime("02/15/2012","%m/%d/%Y") + dt = timedelta(days=1) + + trades = factory.create_trade_history( a, b, c, ts, dt ) + for trade in trades: #simulate data source sending frame msg = zp.DATASOURCE_FRAME(zp.namedict(trade)) @@ -39,8 +46,12 @@ class FinanceTestCase(TestCase): recovered_feed = zp.FEED_UNFRAME(feed_msg) #do a transform trans_msg = zp.TRANSFORM_FRAME('helloworld', 2345.6) - #simulate passthrough transform -- passthrough shouldn't even unpack the msg, just resend. - passthrough_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.PASSTHROUGH, feed_msg) + #simulate passthrough transform -- passthrough shouldn't even + # unpack the msg, just resend. + + passthrough_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.PASSTHROUGH,\ + feed_msg) + #merge unframes transform and passthrough trans_recovered = zp.TRANSFORM_UNFRAME(trans_msg) pt_recovered = zp.TRANSFORM_UNFRAME(passthrough_msg) @@ -50,64 +61,65 @@ class FinanceTestCase(TestCase): merged_msg = zp.MERGE_FRAME(pt_recovered.PASSTHROUGH) #unframe the merge and validate values event = zp.MERGE_UNFRAME(merged_msg) - + #check the transformed value, should only be in event, not trade. self.assertTrue(event.helloworld == 2345.6) event.delete('helloworld') - + self.assertEqual(zp.namedict(trade), event) - + def test_order_protocol(self): #client places an order order_msg = zp.ORDER_FRAME(133, 100) - + #order datasource receives sid, amount = zp.ORDER_UNFRAME(order_msg) self.assertEqual(sid, 133) self.assertEqual(amount, 100) - + #order datasource datasource frames the order - order_dt = datetime.datetime.utcnow().replace(tzinfo=pytz.utc) - order_event = zp.namedict({"sid" : sid, - "amount" : amount, - "dt" : order_dt, - "source_id" : zp.FINANCE_COMPONENT.ORDER_SOURCE, - "type" : zp.DATASOURCE_TYPE.ORDER - }) - + order_dt = datetime.utcnow().replace(tzinfo=pytz.utc) + order_event = zp.namedict({ + "sid" : sid, + "amount" : amount, + "dt" : order_dt, + "source_id" : zp.FINANCE_COMPONENT.ORDER_SOURCE, + "type" : zp.DATASOURCE_TYPE.ORDER + }) + + order_ds_msg = zp.DATASOURCE_FRAME(order_event) - + #transaction transform unframes recovered_order = zp.DATASOURCE_UNFRAME(order_ds_msg) - + self.assertEqual(order_dt, recovered_order.dt) - + #create a transaction from the order txn = zp.namedict({ - 'sid' : recovered_order.sid, - 'amount' : recovered_order.amount, - 'dt' : recovered_order.dt, - 'price' : 10.0, - 'commission' : 0.50 - }) - + 'sid' : recovered_order.sid, + 'amount' : recovered_order.amount, + 'dt' : recovered_order.dt, + 'price' : 10.0, + 'commission' : 0.50 + }) + #frame that transaction txn_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.TRANSACTION, txn) - + #unframe recovered_tx = zp.TRANSFORM_UNFRAME(txn_msg).TRANSACTION self.assertEqual(recovered_tx.sid, 133) self.assertEqual(recovered_tx.amount, 100) - - + def test_trading_calendar(self): - known_trading_day = datetime.datetime.strptime("02/24/2012","%m/%d/%Y") - known_holiday = datetime.datetime.strptime("02/20/2012", "%m/%d/%Y") #president's day - saturday = datetime.datetime.strptime("02/25/2012", "%m/%d/%Y") + known_trading_day = datetime.strptime("02/24/2012","%m/%d/%Y") + known_holiday = datetime.strptime("02/20/2012", "%m/%d/%Y") #president's day + saturday = datetime.strptime("02/25/2012", "%m/%d/%Y") self.assertTrue(risk.trading_calendar.is_trading_day(known_trading_day)) self.assertFalse(risk.trading_calendar.is_trading_day(known_holiday)) self.assertFalse(risk.trading_calendar.is_trading_day(saturday)) - + def test_orders(self): # Just verify sending and receiving orders. @@ -116,7 +128,7 @@ class FinanceTestCase(TestCase): # Allocate sockets for the simulator components allocator = AddressAllocator(8) sockets = allocator.lease(8) - + addresses = { 'sync_address' : sockets[0], 'data_address' : sockets[1], @@ -125,7 +137,7 @@ class FinanceTestCase(TestCase): 'result_address' : sockets[4], 'order_address' : sockets[5] } - + con = Controller( sockets[6], sockets[7], @@ -137,30 +149,34 @@ class FinanceTestCase(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))) - + # TODO: Perhaps something more self-documenting for variables names? + a = 133 + b = [10.0] * 16 + c = [100] * 16 + ts = datetime.strptime("02/1/2012","%m/%d/%Y") + dt = timedelta(days=1) + + trade_history = factory.create_trade_history( a, b, c, ts, dt ) + + set1 = SpecificEquityTrades("flat-133",) + #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)) + ts = datetime.strptime("02/1/2012","%m/%d/%Y").replace(tzinfo=pytz.utc) + + order_source = OrderDataSource(ts) transaction_sim = TransactionSimulator() - + sim.register_components([client, order_source, transaction_sim, set1]) sim.register_controller( con ) # Simulation # ---------- - sim.simulate().join() + 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, - "The feed should be drained of all messages, found {n} remaining." - .format(n=sim.feed.pending_messages()) - ) - \ No newline at end of file + self.assertEqual(sim.feed.pending_messages(), 0, \ + "The feed should be drained of all messages, found {n} remaining." \ + .format(n=sim.feed.pending_messages())) From 3100d7c308d525671b4178a72e4ff0d3d1e5ae58 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Sat, 3 Mar 2012 22:01:05 -0500 Subject: [PATCH 13/16] i++ sweep -> Pythonic iterator / zip --- zipline/test/factory.py | 47 ++++++++++++++++++++++------------------- 1 file changed, 25 insertions(+), 22 deletions(-) diff --git a/zipline/test/factory.py b/zipline/test/factory.py index 2a460506..abbcd593 100644 --- a/zipline/test/factory.py +++ b/zipline/test/factory.py @@ -4,45 +4,48 @@ import zipline.util as qutil import zipline.finance.risk as risk import zipline.protocol as zp - + def create_trade(sid, price, amount, datetime): - row = {} - row['source_id'] = "test_factory" - row['type'] = zp.DATASOURCE_TYPE.TRADE - row['sid'] = sid - row['dt'] = datetime - row['price'] = price - row['volume'] = amount + row = { + 'source_id' : "test_factory", + 'type' : zp.DATASOURCE_TYPE.TRADE, + 'sid' : sid, + 'dt' : datetime, + 'price' : price, + 'volume' : amount + } return row - + def create_trade_history(sid, prices, amounts, start_time, interval): i = 0 trades = [] current = start_time.replace(tzinfo = pytz.utc) while i < len(prices): - if(risk.trading_calendar.is_trading_day(current)): + if(risk.trading_calendar.is_trading_day(current)): trades.append(create_trade(sid, prices[i], amounts[i], current)) current = current + interval i += 1 else: current = current + datetime.timedelta(days=1) - - return trades - + + return trades + def createTxn(sid, price, amount, datetime, btrid=None): - txn = Transaction(sid=sid, amount=amount, dt = datetime, + txn = Transaction(sid=sid, amount=amount, dt = datetime, price=price, transaction_cost=-1*price*amount) return txn - + def createTxnHistory(sid, priceList, amtList, startTime, interval): - i = 0 txns = [] current = startTime - while i < len(priceList): - if(risk.trading_calendar.is_trading_day(current)): - txns.append(createTxn(sid,priceList[i],amtList[i], current)) + + for price, amount in zip(priceList, amtList): + + if risk.trading_calendar.is_trading_day(current): + txns.append(createTxn(sid, price, amount, current)) current = current + interval - i += 1 + else: - current = current + datetime.timedelta(days=1) - return txns \ No newline at end of file + current = current + datetime.timedelta(days=1) + + return txns From 48fa33de682c71c229b4e8a6a37d8f57d2d5214a Mon Sep 17 00:00:00 2001 From: fawce Date: Sun, 4 Mar 2012 11:30:45 -0500 Subject: [PATCH 14/16] commented the (UN)PACK_DATE methods --- zipline/protocol.py | 48 +++++++++++++++++++++++++++++++++------------ 1 file changed, 36 insertions(+), 12 deletions(-) diff --git a/zipline/protocol.py b/zipline/protocol.py index 9631cfc9..f7397132 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -617,28 +617,52 @@ def ORDER_SOURCE_UNFRAME(msg): # ================= def PACK_DATE(event): + """ + Packs the datetime property of event into msgpack'able longs. + This function should be called purely for its side effects. + The event's 'dt' property is replaced by two longs: epoch and micros. + Epoch is the unix epoch time in UTC, and micros is the microsecond + property of the original event.dt datetime object. + + PACK_DATE and UNPACK_DATE are inverse operations. + + :param event: event must a namedict with a property named 'dt' that is a datetime. + :rtype: None + """ assert isinstance(event.dt, datetime.datetime) assert event.dt.tzinfo == pytz.utc #utc only please epoch = long(event.dt.strftime('%s')) event['epoch'] = epoch event['micros'] = event.dt.microsecond event.delete('dt') - return event -def UNPACK_DATE(payload): - assert isinstance(payload.epoch, numbers.Integral) - assert isinstance(payload.micros, numbers.Integral) - dt = datetime.datetime.fromtimestamp(payload.epoch) - dt = dt.replace(microsecond = payload.micros, tzinfo = pytz.utc) - payload.delete('epoch') - payload.delete('micros') - payload.dt = dt - return payload +def UNPACK_DATE(event): + """ + Unpacks the datetime property of event from msgpack'able longs. + This function should be called purely for its side effects. + The event's 'dt' property is created by reading two longs: epoch and micros. + + UNPACK_DATE and PACK_DATE are inverse operations. + + :param event: event must a namedict with:: + - a property named 'epoch' that is an integral representing the unix \ + epoch time in UTC + - a property named 'micros' that is an integral the microsecond \ + property of the original event.dt datetime object + :rtype: None + """ + assert isinstance(event.epoch, numbers.Integral) + assert isinstance(event.micros, numbers.Integral) + dt = datetime.datetime.fromtimestamp(event.epoch) + dt = dt.replace(microsecond = event.micros, tzinfo = pytz.utc) + event.delete('epoch') + event.delete('micros') + event.dt = dt DATASOURCE_TYPE = Enum( - 'ORDER' , - 'TRADE' , + 'ORDER', + 'TRADE' ) ORDER_PROTOCOL = Enum( From 1cd66a091d8cf23450d48b31405570f28199fe6f Mon Sep 17 00:00:00 2001 From: fawce Date: Sun, 4 Mar 2012 11:58:10 -0500 Subject: [PATCH 15/16] fixed bug in client --- zipline/sources.py | 7 +++---- zipline/test/client.py | 7 ++++--- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/zipline/sources.py b/zipline/sources.py index a6631c69..f7b3eada 100644 --- a/zipline/sources.py +++ b/zipline/sources.py @@ -49,16 +49,15 @@ class RandomEquityTrades(TradeDataSource): volume = random.randrange(100,10000,100) event = zp.namedict({ - 'source_id' : self.get_id, - "type" : zp.DATASOURCE_TYPE.ORDER, + #'source_id' : self.get_id, + "type" : zp.DATASOURCE_TYPE.TRADE, "sid" : self.sid, "price" : self.price, "volume" : volume, "dt" : self.trade_start + (self.minute * self.incr), }) - message = zp.DATASOURCE_FRAME(event) - self.send(message) + self.send(event) self.incr += 1 diff --git a/zipline/test/client.py b/zipline/test/client.py index 2b1ac006..205cabff 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -1,5 +1,6 @@ import zipline.util as qutil import zipline.messaging as qmsg +import zipline.protocol as zp from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE from zipline.finance.trading import TradeSimulationClient @@ -45,7 +46,7 @@ class TestClient(qmsg.Component): event = self.unframe(msg) # deserialization error - except zp.InvalidFrame as exc: + except zp.INVALID_MERGE_FRAME as exc: return self.signal_exception(exc) if self.prev_dt != None: @@ -61,8 +62,8 @@ class TestClient(qmsg.Component): if self.received_count % 100 == 0: qutil.LOGGER.info("received {n} messages".format(n=self.received_count)) - def unframe(self, msg): - return zp.MERGE_UNFRARME(msg) + def unframe(self, msg): + return zp.MERGE_UNFRAME(msg) class TestTradingClient(TradeSimulationClient): From 53b4876a1948391618ae1db3c8b9b8f2b100bc7e Mon Sep 17 00:00:00 2001 From: fawce Date: Sun, 4 Mar 2012 11:59:15 -0500 Subject: [PATCH 16/16] removed the commented out line that was setting the source id. parent class ensures id is set. --- zipline/sources.py | 1 - 1 file changed, 1 deletion(-) diff --git a/zipline/sources.py b/zipline/sources.py index f7b3eada..28d356b9 100644 --- a/zipline/sources.py +++ b/zipline/sources.py @@ -49,7 +49,6 @@ class RandomEquityTrades(TradeDataSource): volume = random.randrange(100,10000,100) event = zp.namedict({ - #'source_id' : self.get_id, "type" : zp.DATASOURCE_TYPE.TRADE, "sid" : self.sid, "price" : self.price,