From f7e4f574251cae01d0c3c5b087c0f9ca3f56f60a Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 28 Dec 2012 09:55:55 -0500 Subject: [PATCH] Enables performance messages on days that have no trades. Previously, on days that were trading days, but there with no event data to process for that day, performance metrics were not emitted, since the handling was based on having an event trigger the daily performance metric. Handled by grouping together performance messages, on market open, for all days since the last market close. Also, changes perf_tracker unit test to simulate missing data. Taken from @richafrank's branch handling the same case. --- tests/test_perf_tracking.py | 34 ++++++++++++++++++++++++--------- zipline/finance/performance.py | 30 +++++++++++++++++------------ zipline/gens/tradesimulation.py | 23 +++++++++++----------- 3 files changed, 54 insertions(+), 33 deletions(-) diff --git a/tests/test_perf_tracking.py b/tests/test_perf_tracking.py index 3e57c0e2..aea79660 100644 --- a/tests/test_perf_tracking.py +++ b/tests/test_perf_tracking.py @@ -537,7 +537,8 @@ shares in position" ) @parameterized.expand([ - # This date range covers Columbus day + # This date range covers Columbus day, + # however Columbus day is not a market holiday # # October 2008 # Su Mo Tu We Th Fr Sa @@ -598,6 +599,10 @@ shares in position" source_id="factory1" ) + # Removes second day of trading. + # To simulate days that don't have events. + del trade_history[-1] + sid2 = 134 price2 = 12.12 price2_list = [price2] * trade_count @@ -610,10 +615,15 @@ shares in position" source_id="factory2" ) + # Removes second day of trading. + # To simulate days that don't have events. + del trade_history2[-1] + trade_history.extend(trade_history2) - trading_environment.period_start = trade_history[0].dt - trading_environment.period_end = trade_history[-1].dt + trading_environment.period_start = \ + trade_history[0].dt.replace(hour=0, minute=0, second=0) + trading_environment.period_end = trade_history2[-1].dt trading_environment.first_open = \ trading_environment.calculate_first_open() trading_environment.last_close = \ @@ -645,11 +655,15 @@ shares in position" events = itertools.chain(events, [ndict({'dt': 'DONE'})]) - events = [self.event_with_txn(event, trading_environment) + events = [self.event_with_txn(event, trade_history[0].dt) for event in events] - list(perf_tracker.transform( - itertools.groupby(events, attrgetter('dt')))) + perf_messages = \ + [msg for date, snapshot in + perf_tracker.transform( + itertools.groupby(events, attrgetter('dt'))) + for event in snapshot + for msg in event.perf_messages] #we skip two trades, to test case of None transaction txn_count = len(trade_history) - 2 @@ -662,11 +676,13 @@ shares in position" self.assertEqual(perf_tracker.last_close, perf_tracker.cumulative_risk_metrics.end_date) - def event_with_txn(self, event, env): + self.assertEqual(len(perf_messages), + trading_environment.days_in_period) + + def event_with_txn(self, event, no_txn_dt): #create a transaction for all but #first trade in each sid, to simulate None transaction - if event.dt != env.period_start \ - and event.dt != 'DONE': + if event.dt != no_txn_dt and event.dt != 'DONE': txn = ndict({ 'sid': event.sid, 'amount': -25, diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 6f11a62b..67428ede 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -220,14 +220,11 @@ class PerformanceTracker(object): for event in snapshot: if date != "DONE": - event.perf_message = self.process_event(event) + event.perf_messages = self.process_event(event) event.portfolio = self.get_portfolio() else: - # the stream will end on the last trading day, but will - # not trigger an end of day, so we trigger the final - # market close here - event.perf_message = self.handle_market_close() - event.risk_message = self.handle_simulation_end() + event.perf_messages, event.risk_message = \ + self.handle_simulation_end() del event['TRANSACTION'] new_snapshot.append(event) @@ -255,12 +252,12 @@ class PerformanceTracker(object): def process_event(self, event): - message = None + messages = [] self.event_count += 1 - if(event.dt > self.market_close): - message = self.handle_market_close() + while event.dt > self.market_close: + messages.append(self.handle_market_close()) if event.TRANSACTION: self.txn_count += 1 @@ -275,7 +272,7 @@ class PerformanceTracker(object): self.cumulative_performance.calculate_performance() self.todays_performance.calculate_performance() - return message + return messages def handle_market_close(self): @@ -337,9 +334,18 @@ Last successful date: %s" % self.market_open) When the simulation is complete, run the full period risk report and send it out on the results socket. """ + # the stream will end on the last trading day, but will + # not trigger an end of day, so we trigger the final + # market close(s) here + perf_messages = [] + + while self.last_close > self.market_close: + perf_messages.append(self.handle_market_close()) + + perf_messages.append(self.handle_market_close()) log_msg = "Simulated {n} trading days out of {m}." - log.info(log_msg.format(n=self.day_count, m=self.total_days)) + log.info(log_msg.format(n=int(self.day_count), m=self.total_days)) log.info("first open: {d}".format( d=self.trading_environment.first_open)) log.info("last close: {d}".format( @@ -351,7 +357,7 @@ Last successful date: %s" % self.market_open) ) risk_dict = self.risk_report.to_dict() - return risk_dict + return perf_messages, risk_dict class Position(object): diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 4aab868d..5a843aab 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -97,8 +97,8 @@ class TradeSimulationClient(object): # Pipe the events with transactions to perf. This will remove # the TRANSACTION field added by TransactionSimulator and replace it # with a portfolio field to be passed to the user's - # algorithm. Also adds a perf_message field which is usually - # none, but contains an update message once per day. + # algorithm. Also adds a perf_messages field which is usually + # empty, but contains update messages once per day. with_portfolio = self.perf_tracker.transform(with_filled_orders) # Pass the messages from perf to the user's algorithm for simulation. @@ -208,7 +208,8 @@ class AlgorithmSimulator(object): # Done message has the risk report, so we yield before exiting. if date == 'DONE': for event in snapshot: - yield event.perf_message + for perf_message in event.perf_messages: + yield perf_message yield event.risk_message raise StopIteration @@ -217,7 +218,7 @@ class AlgorithmSimulator(object): # and don't send a snapshot to handle_data. elif date < self.algo_start: for event in snapshot: - del event['perf_message'] + del event['perf_messages'] self.update_universe(event) # The algo has taken so long to process events that @@ -226,22 +227,20 @@ class AlgorithmSimulator(object): # encountered, but don't call handle_data. elif date < self.simulation_dt: for event in snapshot: - # Only yield if we have something interesting to say. - if event.perf_message is not None: - yield event.perf_message + for perf_message in event.perf_messages: + yield perf_message # Delete the message before updating, # so we don't send it to the user. - del event['perf_message'] + del event['perf_messages'] self.update_universe(event) # Regular snapshot. Update the universe and send a snapshot # to handle data. else: for event in snapshot: - # Only yield if we have something interesting to say. - if event.perf_message is not None: - yield event.perf_message - del event['perf_message'] + for perf_message in event.perf_messages: + yield perf_message + del event['perf_messages'] self.update_universe(event)