Merge pull request #51 from quantopian/yield-perf-data-for-missing-days

Enables performance messages on days that have no trades.

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,

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):

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)