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)