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

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