ENH: Add intraday risk and performance for minute emission.

Both risk and performance now calculate performance since inception
(cumulative) and since the open. Both periods are updated intraday
and both are reported.

Batch risk for periods starting after the end of the treasury curve
history now use most recent curve.

zipline/finance/performance.py

@@ -162,19 +162,46 @@ class PerformanceTracker(object):
         self.total_days = self.sim_params.days_in_period
         self.capital_base = self.sim_params.capital_base
         self.emission_rate = sim_params.emission_rate
-        self.cumulative_risk_metrics = \
-            risk.RiskMetricsIterative(self.sim_params)
         self.emission_rate = sim_params.emission_rate
 
+        self.perf_periods = []
+
         if self.emission_rate == 'daily':
             self.all_benchmark_returns = pd.Series(
                 index=trading.environment.trading_days)
+            self.intraday_risk_metrics = None
+            self.cumulative_risk_metrics = \
+                risk.RiskMetricsIterative(self.sim_params)
+
         elif self.emission_rate == 'minute':
             self.all_benchmark_returns = pd.Series(index=pd.date_range(
                 self.sim_params.first_open, self.sim_params.last_close,
                 freq='Min'))
+            self.intraday_risk_metrics = \
+                risk.RiskMetricsIterative(self.sim_params)
 
-        # this performance period will span the entire simulation.
+            self.cumulative_risk_metrics = \
+                risk.RiskMetricsIterative(self.sim_params)
+            self.cumulative_risk_metrics.initialize_daily_indices()
+
+            self.minute_performance = PerformancePeriod(
+                # initial cash is your capital base.
+                self.capital_base,
+                # the cumulative period will be calculated over the
+                # entire test.
+                self.period_start,
+                self.period_end,
+                # don't save the transactions for the cumulative
+                # period
+                keep_transactions=False,
+                keep_orders=False,
+                # don't serialize positions for cumualtive period
+                serialize_positions=False
+            )
+            self.perf_periods.append(self.minute_performance)
+
+        # this performance period will span the entire simulation from
+        # inception.
         self.cumulative_performance = PerformancePeriod(
             # initial cash is your capital base.
             self.capital_base,
@@ -188,6 +215,7 @@ class PerformanceTracker(object):
             # don't serialize positions for cumualtive period
             serialize_positions=False
         )
+        self.perf_periods.append(self.cumulative_performance)
 
         # this performance period will span just the current market day
         self.todays_performance = PerformancePeriod(
@@ -200,6 +228,7 @@ class PerformanceTracker(object):
             keep_orders=True,
             serialize_positions=True
         )
+        self.perf_periods.append(self.todays_performance)
 
         self.saved_dt = self.period_start
         self.returns = []
@@ -255,9 +284,11 @@ class PerformanceTracker(object):
             # Naming as intraday to make clear that these results are
             # being updated per minute
             _dict['intraday_risk_metrics'] = \
-                self.cumulative_risk_metrics.to_dict()
+                self.intraday_risk_metrics.to_dict()
             _dict['intraday_perf'] = self.todays_performance.to_dict(
                 self.saved_dt)
+            _dict['cumulative_risk_metrics'] = \
+                self.cumulative_risk_metrics.to_dict()
 
         return _dict
 
@@ -267,26 +298,24 @@ class PerformanceTracker(object):
 
         if event.type == zp.DATASOURCE_TYPE.TRADE:
             #update last sale
-            self.cumulative_performance.update_last_sale(event)
-            self.todays_performance.update_last_sale(event)
+            for perf_period in self.perf_periods:
+                perf_period.update_last_sale(event)
 
         elif event.type == zp.DATASOURCE_TYPE.TRANSACTION:
             # Trade simulation always follows a transaction with the
             # TRADE event that was used to simulate it, so we don't
             # check for end of day rollover messages here.
             self.txn_count += 1
-            self.cumulative_performance.execute_transaction(
-                event
-            )
-            self.todays_performance.execute_transaction(event)
+            for perf_period in self.perf_periods:
+                perf_period.execute_transaction(event)
 
         elif event.type == zp.DATASOURCE_TYPE.DIVIDEND:
-            self.cumulative_performance.add_dividend(event)
-            self.todays_performance.add_dividend(event)
+            for perf_period in self.perf_periods:
+                perf_period.add_dividend(event)
 
         elif event.type == zp.DATASOURCE_TYPE.ORDER:
-            self.cumulative_performance.record_order(event)
-            self.todays_performance.record_order(event)
+            for perf_period in self.perf_periods:
+                perf_period.record_order(event)
 
         elif event.type == zp.DATASOURCE_TYPE.CUSTOM:
             pass
@@ -294,14 +323,51 @@ class PerformanceTracker(object):
             self.all_benchmark_returns[event.dt] = event.returns
 
         #calculate performance as of last trade
-        self.cumulative_performance.calculate_performance()
-        self.todays_performance.calculate_performance()
+        for perf_period in self.perf_periods:
+            perf_period.calculate_performance()
 
     def handle_minute_close(self, dt):
-        #update risk metrics for cumulative performance
-        self.cumulative_risk_metrics.update(dt,
-                                            self.todays_performance.returns,
-                                            self.all_benchmark_returns[dt])
+
+        todays_date = self.market_close.replace(hour=0, minute=0, second=0,
+                                                microsecond=0)
+
+        minute_returns = self.minute_performance.returns
+        self.minute_performance.rollover()
+        algo_minute_returns = pd.Series({dt: minute_returns})
+        bench_minute_returns = pd.Series({dt: self.all_benchmark_returns[dt]})
+        # the intraday risk is calculated on top of minute performance
+        # returns for the bench and the algo
+        self.intraday_risk_metrics.update(dt,
+                                          algo_minute_returns,
+                                          bench_minute_returns)
+        # the intraday risk metrics compound the minutely returns of the
+        # benchmark.
+        bench_since_open = self.intraday_risk_metrics.benchmark_returns[-1]
+        benchmark_returns = pd.Series({dt: bench_since_open})
+
+        # if we've reached market close, check on dividends
+        if dt == self.market_close:
+            for perf_period in self.perf_periods:
+                perf_period.update_dividends(todays_date)
+
+        algorithm_returns = pd.Series({dt: self.todays_performance.returns})
+
+        self.intraday_risk_metrics.update(dt,
+                                          algorithm_returns,
+                                          benchmark_returns)
+
+        self.cumulative_risk_metrics.update(todays_date,
+                                            algorithm_returns,
+                                            benchmark_returns)
+
+        # if this is the close, save the returns objects for cumulative
+        # risk calculations
+        if dt == self.market_close:
+            todays_return_obj = zp.DailyReturn(
+                todays_date,
+                self.todays_performance.returns
+            )
+            self.returns.append(todays_return_obj)
 
     def handle_market_close(self):
         # add the return results from today to the list of DailyReturn objects.

zipline/finance/risk.py

@@ -284,37 +284,56 @@ that date doesn't exceed treasury history range."
 
 
 class RiskMetricsBase(object):
-    def __init__(self, start_date, end_date, returns):
+    def __init__(self, start_date, end_date, returns,
+                 benchmark_returns=None):
 
         treasury_curves = trading.environment.treasury_curves
-        mask = ((treasury_curves.index >= start_date) &
-                (treasury_curves.index <= end_date))
+        if treasury_curves.index[-1] >= start_date:
+            mask = ((treasury_curves.index >= start_date) &
+                    (treasury_curves.index <= end_date))
 
-        self.treasury_curves = treasury_curves[mask]
+            self.treasury_curves = treasury_curves[mask]
+        else:
+            # our test is beyond the treasury curve history
+            # so we'll use the last available treasury curve
+            self.treasury_curves = treasury_curves[-1:]
 
         self.start_date = start_date
         self.end_date = end_date
-        self.algorithm_period_returns, self.algorithm_returns = \
-            self.calculate_period_returns(returns)
 
-        benchmark_returns = [
-            x for x in trading.environment.benchmark_returns
-            if x.date >= returns[0].date and x.date <= returns[-1].date
-        ]
+        if not benchmark_returns:
+            benchmark_returns = [
+                x for x in trading.environment.benchmark_returns
+                if x.date >= returns[0].date and
+                x.date <= returns[-1].date
+            ]
 
-        self.benchmark_period_returns, self.benchmark_returns = \
-            self.calculate_period_returns(benchmark_returns)
+        self.runonce = True
+        self.algorithm_returns = self.mask_returns_to_period(returns)
+        self.benchmark_returns = self.mask_returns_to_period(benchmark_returns)
+        self.calculate_metrics()
 
-        if(len(self.benchmark_returns) != len(self.algorithm_returns)):
+    def calculate_metrics(self):
+
+        self.benchmark_period_returns = \
+            self.calculate_period_returns(self.benchmark_returns)
+
+        self.algorithm_period_returns = \
+            self.calculate_period_returns(self.algorithm_returns)
+
+        if not self.algorithm_returns.index.equals(
+            self.benchmark_returns.index
+        ):
             message = "Mismatch between benchmark_returns ({bm_count}) and \
             algorithm_returns ({algo_count}) in range {start} : {end}"
             message = message.format(
                 bm_count=len(self.benchmark_returns),
                 algo_count=len(self.algorithm_returns),
-                start=start_date,
-                end=end_date
+                start=self.start_date,
+                end=self.end_date
             )
             raise Exception(message)
+        self.runonce = False
 
         self.num_trading_days = len(self.benchmark_returns)
         self.benchmark_volatility = self.calculate_volatility(
@@ -399,7 +418,7 @@ class RiskMetricsBase(object):
 
         return '\n'.join(statements)
 
-    def calculate_period_returns(self, daily_returns):
+    def mask_returns_to_period(self, daily_returns):
         returns = pd.Series([x.returns for x in daily_returns],
                             index=[x.date for x in daily_returns])
 
@@ -410,9 +429,11 @@ class RiskMetricsBase(object):
                 (returns.index <= self.end_date) & trade_day_mask)
 
         returns = returns[mask]
-        period_returns = (1. + returns).prod() - 1
+        return returns
 
-        return period_returns, returns
+    def calculate_period_returns(self, returns):
+        period_returns = (1. + returns).prod() - 1
+        return period_returns
 
     def calculate_volatility(self, daily_returns):
         return np.std(daily_returns, ddof=1) * math.sqrt(self.num_trading_days)
@@ -536,21 +557,18 @@ class RiskMetricsIterative(RiskMetricsBase):
                 (all_trading_days <= self.end_date))
 
         self.trading_days = all_trading_days[mask]
+        if sim_params.period_end not in self.trading_days:
+            last_day = pd.tseries.index.DatetimeIndex(
+                [sim_params.period_end]
+            )
+            self.trading_days = self.trading_days.append(last_day)
 
         self.sim_params = sim_params
 
         if sim_params.emission_rate == 'daily':
-            self.algorithm_returns_cont = pd.Series(index=self.trading_days)
-            self.benchmark_returns_cont = pd.Series(index=self.trading_days)
-
+            self.initialize_daily_indices()
         elif sim_params.emission_rate == 'minute':
-
-            self.algorithm_returns_cont = pd.Series(index=pd.date_range(
-                sim_params.first_open, sim_params.last_close,
-                freq="Min"))
-            self.benchmark_returns_cont = pd.Series(index=pd.date_range(
-                sim_params.first_open, sim_params.last_close,
-                freq="Min"))
+            self.initialize_minute_indices(sim_params)
 
         self.algorithm_returns = None
         self.benchmark_returns = None
@@ -576,6 +594,19 @@ class RiskMetricsIterative(RiskMetricsBase):
         self.max_drawdown = 0
         self.current_max = -np.inf
         self.excess_returns = []
+        self.daily_treasury = {}
+
+    def initialize_minute_indices(self, sim_params):
+        self.algorithm_returns_cont = pd.Series(index=pd.date_range(
+            sim_params.first_open, sim_params.last_close,
+            freq="Min"))
+        self.benchmark_returns_cont = pd.Series(index=pd.date_range(
+            sim_params.first_open, sim_params.last_close,
+            freq="Min"))
+
+    def initialize_daily_indices(self):
+        self.algorithm_returns_cont = pd.Series(index=self.trading_days)
+        self.benchmark_returns_cont = pd.Series(index=self.trading_days)
 
     @property
     def last_return_date(self):
@@ -597,7 +628,9 @@ class RiskMetricsIterative(RiskMetricsBase):
         self.benchmark_period_returns.append(
             self.calculate_period_returns(self.benchmark_returns))
 
-        if(len(self.benchmark_returns) != len(self.algorithm_returns)):
+        if not self.algorithm_returns.index.equals(
+            self.benchmark_returns.index
+        ):
             message = "Mismatch between benchmark_returns ({bm_count}) and \
 algorithm_returns ({algo_count}) in range {start} : {end} on {dt}"
             message = message.format(
@@ -614,11 +647,22 @@ algorithm_returns ({algo_count}) in range {start} : {end} on {dt}"
             self.calculate_volatility(self.benchmark_returns))
         self.algorithm_volatility.append(
             self.calculate_volatility(self.algorithm_returns))
-        self.treasury_period_return = choose_treasury(
-            self.treasury_curves,
-            self.start_date,
-            self.algorithm_returns.index[-1]
-        )
+
+        # caching the treasury rates for the live case is a
+        # big speedup, because it avoids searching the treasury
+        # curves on every minute.
+        treasury_end = self.algorithm_returns.index[-1].replace(
+            hour=0, minute=0)
+        if treasury_end not in self.daily_treasury:
+            treasury_period_return = choose_treasury(
+                self.treasury_curves,
+                self.start_date,
+                self.algorithm_returns.index[-1]
+            )
+            self.daily_treasury[treasury_end] =\
+                treasury_period_return
+        self.treasury_period_return = \
+            self.daily_treasury[treasury_end]
         self.excess_returns.append(
             self.algorithm_period_returns[-1] - self.treasury_period_return)
         self.beta.append(self.calculate_beta()[0])

zipline/finance/trading.py

@@ -88,6 +88,8 @@ class TradingEnvironment(object):
             load(self.bm_symbol)
 
         self.treasury_curves = pd.Series(treasury_curves_map)
+        if max_date:
+            self.treasury_curves = self.treasury_curves[:max_date]
 
         self._period_trading_days = None
         self._trading_days_series = None