From 366405927d3be2cdf20d823e8428fbf0fb6e3b99 Mon Sep 17 00:00:00 2001 From: fawce Date: Sun, 11 Mar 2012 16:21:10 -0400 Subject: [PATCH] changed date serialization to use a tuple of all properties rather than the epoch time to eliminate any timezone sensitivity. add performance tracker unit tests, made various fixes to perf tracker. still have a hang on exit for zipline/test/test_finance.py:FinanceTestCase.test_orders and zipline/test/test_finance.py:FinanceTestCase.test_performance. pinging realdiehl for help... --- zipline/finance/performance.py | 49 ++++++++------- zipline/finance/trading.py | 6 +- zipline/messaging.py | 3 +- zipline/protocol.py | 62 ++++++++----------- zipline/test/client.py | 6 +- zipline/test/factory.py | 15 ++++- zipline/test/test_finance.py | 14 ++++- ...t_performance.py => test_perf_tracking.py} | 1 - zipline/test/test_risk.py | 2 +- 9 files changed, 87 insertions(+), 71 deletions(-) rename zipline/test/{test_performance.py => test_perf_tracking.py} (99%) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 66528fea..a4b1223c 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -29,19 +29,15 @@ class PerformanceTracker(): self.txn_count = 0 self.event_count = 0 self.cumulative_performance = PerformancePeriod( - self.period_start, - self.period_end, {}, capital_base, - capital_base = capital_base + starting_cash = capital_base ) self.todays_performance = PerformancePeriod( - self.market_open, - self.market_close, {}, capital_base, - capital_base = capital_base + starting_cash = capital_base ) @@ -72,26 +68,33 @@ class PerformanceTracker(): self.todays_performance.calculate_performance() def handle_market_close(self): - qutil.LOGGER.debug("###########market close###############") - self.market_open = self.market_open + self.calendar_day - while not self.trading_environment.is_trading_day(self.market_open): - if self.market_open > self.trading_environment.trading_days[-1]: - raise Exception("Attempting to backtest beyond available history.") - self.market_open = self.market_open + self.calendar_day - self.market_close = self.market_open + self.trading_day - self.day_count += 1.0 - self.progress = self.day_count / self.total_days - #add the return results from today to the list of daily return objects. - todays_date = self.todays_performance.period_end.replace(hour=0, minute=0, second=0) + #add the return results from today to the list of daily return objects. + todays_date = self.market_close.replace(hour=0, minute=0, second=0) todays_return_obj = risk.daily_return(todays_date, self.todays_performance.returns) self.returns.append(todays_return_obj) #calculate risk metrics for cumulative performance - self.cur_period_metrics = risk.RiskMetrics( - start_date=self.cumulative_performance.period_start, - end_date=self.cumulative_performance.period_end.replace(hour=0, minute=0, second=0), + self.cumulative_risk_metrics = risk.RiskMetrics( + start_date=self.period_start, + end_date=self.market_close.replace(hour=0, minute=0, second=0), returns=self.returns, - trading_environment=self.trading_environment) + trading_environment=self.trading_environment + ) + + #move the market day markers forward + self.market_open = self.market_open + self.calendar_day + while not self.trading_environment.is_trading_day(self.market_open): + if self.market_open > self.trading_environment.trading_days[-1]: + raise Exception("Attempt to backtest beyond available history.") + self.market_open = self.market_open + self.calendar_day + self.market_close = self.market_open + self.trading_day + self.day_count += 1.0 + + #calculate progress of test + self.progress = self.day_count / self.total_days + + + ###################################################################################################### #######TODO: report/relay metrics out to qexec -- values come from self.cur_period_metrics ########### @@ -101,8 +104,6 @@ class PerformanceTracker(): #roll over positions to current day. self.todays_performance.calculate_performance() self.todays_performance = PerformancePeriod( - self.market_open, - self.market_close, self.todays_performance.positions, self.todays_performance.ending_value, self.todays_performance.ending_cash @@ -201,7 +202,7 @@ class PerformancePeriod(): return mktValue def update_last_sale(self, event): - if self.positions.has_key(event.sid): + if self.positions.has_key(event.sid) and event.type == zp.DATASOURCE_TYPE.TRADE: self.positions[event.sid].last_sale = event.price self.positions[event.sid].last_date = event.dt diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 6a8b81d8..80098ff2 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -16,6 +16,7 @@ class TradeSimulationClient(qmsg.Component): self.prev_dt = None self.event_queue = [] self.event_callbacks = [] + self.txn_count = 0 @property def get_id(self): @@ -48,6 +49,9 @@ class TradeSimulationClient(qmsg.Component): event = zp.MERGE_UNFRAME(msg) + if(event.TRANSACTION != None): + self.txn_count += 1 + for cb in self.event_callbacks: cb(event) @@ -123,6 +127,7 @@ class OrderDataSource(qmsg.DataSource): order_msg = rlist[0].recv() if order_msg == str(zp.ORDER_PROTOCOL.DONE): + qutil.LOGGER.debug("Order source received done message.") self.signal_done() return @@ -144,7 +149,6 @@ class OrderDataSource(qmsg.DataSource): # or the feed will block waiting for our messages. if(count == 0): self.send(zp.namedict({})) - self.sent_count += 1 diff --git a/zipline/messaging.py b/zipline/messaging.py index bbbab784..5d3e5fd0 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -287,7 +287,8 @@ class ParallelBuffer(Component): cur_source = None earliest_source = None earliest_event = None - #iterate over the queues of events from all sources (1 queue per datasource) + #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 diff --git a/zipline/protocol.py b/zipline/protocol.py index 5bd238d6..c1ad89fa 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -492,15 +492,14 @@ def TRADE_FRAME(event): event.sid, event.price, event.volume, - event.epoch, - event.micros, + event.dt, event.type, ])) def TRADE_UNFRAME(msg): try: packed = msgpack.loads(msg) - sid, price, volume, epoch, micros, source_type = packed + sid, price, volume, dt, source_type = packed assert isinstance(sid, int) assert isinstance(price, numbers.Real) @@ -509,8 +508,7 @@ def TRADE_UNFRAME(msg): 'sid' : sid, 'price' : price, 'volume' : volume, - 'epoch' : epoch, - 'micros' : micros, + 'dt' : dt, 'type' : source_type }) UNPACK_DATE(rval) @@ -559,13 +557,12 @@ def TRANSACTION_FRAME(event): event.price, event.amount, event.commission, - event.epoch, - event.micros + event.dt ])) def TRANSACTION_UNFRAME(msg): try: - sid, price, amount, commission, epoch, micros = msgpack.loads(msg) + sid, price, amount, commission, dt = msgpack.loads(msg) assert isinstance(sid, int) assert isinstance(price, numbers.Real) @@ -576,8 +573,7 @@ def TRANSACTION_UNFRAME(msg): 'price' : price, 'amount' : amount, 'commission' : commission, - 'epoch' : epoch, - 'micros' : micros + 'dt' : dt }) UNPACK_DATE(rval) @@ -602,8 +598,7 @@ def ORDER_SOURCE_FRAME(event): return msgpack.dumps(tuple([ event.sid, event.amount, - event.epoch, - event.micros, + event.dt, event.source_id, event.type ])) @@ -611,12 +606,11 @@ def ORDER_SOURCE_FRAME(event): def ORDER_SOURCE_UNFRAME(msg): try: - sid, amount, epoch, micros, source_id, source_type = msgpack.loads(msg) + sid, amount, dt, source_id, source_type = msgpack.loads(msg) event = namedict({ "sid" : sid, "amount" : amount, - "epoch" : epoch, - "micros" : micros, + "dt" : dt, "source_id" : source_id, "type" : source_type }) @@ -639,9 +633,8 @@ 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. + The event's 'dt' property is replaced by a tuple of integers:: + - year, month, day, hour, minute, second, microsecond PACK_DATE and UNPACK_DATE are inverse operations. @@ -650,33 +643,32 @@ 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')) - event['epoch'] = epoch - event['micros'] = event.dt.microsecond - event.delete('dt') + year, month, day, hour, minute, second = event.dt.timetuple()[0:6] + micros = event.dt.microsecond + event['dt'] = tuple([year, month, day, hour, minute, second, micros]) 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 and then combining two longs: epoch and micros. - The epoch and micros properties are removed after dt is added. + The event's 'dt' property is converted to a datetime by reading and then + combining a tuple of integers. 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 + :param tuple event: event must a namedict with:: + - a property named 'dt_tuple' that is a tuple of integers + representing the date and time in UTC. dt_tumple must have year, + month, day, hour, minute, second, and microsecond :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') + assert isinstance(event.dt, tuple) + assert len(event.dt) == 7 + for item in event.dt: + assert isinstance(item, numbers.Integral) + year, month, day, hour, minute, second, micros = event.dt + dt = datetime.datetime(year, month, day, hour, minute, second) + dt = dt.replace(microsecond = micros, tzinfo = pytz.utc) event.dt = dt diff --git a/zipline/test/client.py b/zipline/test/client.py index 19bc00b5..c63f68b8 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -75,14 +75,14 @@ class TestAlgorithm(): self.sid = sid self.amount = amount self.incr = 0 + self.done = False def handle_event(self, event): - qutil.LOGGER.debug(event) #place an order for 100 shares of sid:133 if self.incr < self.count: if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE: self.trading_client.order(self.sid, self.amount) self.incr += 1 - else: + elif not self.done: self.trading_client.signal_order_done() - self.trading_client.signal_done() + self.done = True diff --git a/zipline/test/factory.py b/zipline/test/factory.py index 4728bb45..48b01bb2 100644 --- a/zipline/test/factory.py +++ b/zipline/test/factory.py @@ -11,13 +11,24 @@ def load_market_data(): bm_map = msgpack.loads(fp_bm.read()) bm_returns = [] for epoch, returns in bm_map.iteritems(): - bm_returns.append(risk.daily_return(date=datetime.datetime.fromtimestamp(epoch).replace(hour=0, minute=0, second=0, tzinfo=pytz.utc), returns=returns)) + event_dt = datetime.datetime.fromtimestamp(epoch) + event_dt = event_dt.replace( + hour=0, + minute=0, + second=0, + tzinfo=pytz.utc + ) + + daily_return = risk.daily_return(date=event_dt, returns=returns) + bm_returns.append(daily_return) bm_returns = sorted(bm_returns, key=lambda(x): x.date) fp_tr = open("./zipline/test/treasury_curves.msgpack", "rb") tr_map = msgpack.loads(fp_tr.read()) tr_curves = {} for epoch, curve in tr_map.iteritems(): - tr_curves[datetime.datetime.fromtimestamp(epoch).replace(hour=0, minute=0, second=0, tzinfo=pytz.utc)] = curve + tr_dt = datetime.datetime.fromtimestamp(epoch) + tr_dt = tr_dt.replace(hour=0, minute=0, second=0, tzinfo=pytz.utc) + tr_curves[tr_dt] = curve return bm_returns, tr_curves diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index b367299e..f8a7e7e2 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -231,9 +231,10 @@ class FinanceTestCase(TestCase): # --------------------- # TODO: Perhaps something more self-documenting for variables names? + trade_count = 100 sid = 133 - price = [10.1] * 16 - volume = [100] * 16 + price = [10.1] * trade_count + volume = [100] * trade_count start_date = datetime.strptime("02/1/2012","%m/%d/%Y") trade_time_increment = timedelta(days=1) @@ -283,6 +284,13 @@ class FinanceTestCase(TestCase): "The feed should be drained of all messages, found {n} remaining." \ .format(n=sim.feed.pending_messages()) ) + + self.assertEqual( + sim.merge.pending_messages(), + 0, + "The merge should be drained of all messages, found {n} remaining." \ + .format(n=sim.merge.pending_messages()) + ) self.assertEqual( test_algo.count, @@ -299,7 +307,7 @@ class FinanceTestCase(TestCase): transaction_sim.txn_count, perf_tracker.txn_count, "The perf tracker should handle the same number of transactions as\ - as the simulator emits." +as the simulator emits." ) self.assertEqual( diff --git a/zipline/test/test_performance.py b/zipline/test/test_perf_tracking.py similarity index 99% rename from zipline/test/test_performance.py rename to zipline/test/test_perf_tracking.py index 1efed392..b8ab30c0 100644 --- a/zipline/test/test_performance.py +++ b/zipline/test/test_perf_tracking.py @@ -5,7 +5,6 @@ import datetime import zipline.test.factory as factory import zipline.util as qutil -import zipline.protocol as zp import zipline.finance.performance as perf import zipline.finance.risk as risk class PerformanceTestCase(unittest.TestCase): diff --git a/zipline/test/test_risk.py b/zipline/test/test_risk.py index 81aa0de9..bdcf575f 100644 --- a/zipline/test/test_risk.py +++ b/zipline/test/test_risk.py @@ -41,7 +41,7 @@ class Risk(unittest.TestCase): start_date = datetime.datetime(year=2006, month=1, day=1) returns = factory.create_returns_from_list([1.0,-0.5,0.8,.17,1.0,-0.1,-0.45], start_date, self.trading_calendar) #200, 100, 180, 210.6, 421.2, 379.8, 208.494 - metrics = risk.RiskMetrics(returns[0].date, returns[-1].date, returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + metrics = risk.RiskMetrics(returns[0].date, returns[-1].date, returns, self.trading_calendar) self.assertEqual(metrics.max_drawdown, 0.505) def test_benchmark_returns_06(self):