Provides an iterative version of risk metrics.

I wrote this a little while ago as I noticed that a lot of time is spent
computing risk statistics. This is done over the complete history over
and over again while this could be done just by using the previously
computed value (iteratively).

We didn't go forward back then because for minute trade data the
difference was not significant enough. However, now with zipline
standalone I think most people will use daily (because that's
what's available) and it makes a huge difference
(speed-up of a couple of 100%).

Unfortunately, we can't just replace the existing one with an
iterative as for the final cumulative stats the batch is still
better. So that's not as nice, but the performance increase is
big enough for me to issue this PR (zipline is actually painfully
slow with daily data).

There is a unittest that compares that both produce exactly
the same outputs.

Speed measurements (for 500 trading days, daily source):

with iterative:
real 26.617 user 12.909 sys 6.112 pcpu 71.46

prior:
real 44.176 user 31.030 sys 11.381 pcpu 96.00

tests/test_risk.py

@@ -97,10 +97,10 @@ class Risk(unittest.TestCase):
         returns = factory.create_returns_from_list(
             [1.0, -0.5, 0.8, .17, 1.0, -0.1, -0.45], self.trading_env)
         #200, 100, 180, 210.6, 421.2, 379.8, 208.494
-        metrics = risk.RiskMetrics(returns[0].date,
-                                   returns[-1].date,
-                                   returns,
-                                   self.trading_env)
+        metrics = risk.RiskMetricsBatch(returns[0].date,
+                                        returns[-1].date,
+                                        returns,
+                                        self.trading_env)
         self.assertEqual(metrics.max_drawdown, 0.505)
 
     def test_benchmark_returns_06(self):

tests/test_risk_compare_batch_iterative.py

@@ -0,0 +1,140 @@
+#
+# Copyright 2012 Quantopian, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import unittest
+import datetime
+import pytz
+
+import numpy as np
+
+import zipline.finance.risk as risk
+from zipline.utils import factory
+
+from zipline.finance.trading import TradingEnvironment
+from test_risk import RETURNS
+
+
+class RiskCompareIterativeToBatch(unittest.TestCase):
+    """
+    Assert that RiskMetricsIterative and RiskMetricsBatch
+    behave in the same way.
+    """
+
+    def setUp(self):
+        self.start_date = datetime.datetime(
+            year=2006,
+            month=1,
+            day=1,
+            hour=0,
+            minute=0,
+            tzinfo=pytz.utc)
+        self.end_date = datetime.datetime(
+            year=2006, month=12, day=31, tzinfo=pytz.utc)
+        self.benchmark_returns, self.treasury_curves = \
+        factory.load_market_data()
+
+        self.trading_env = TradingEnvironment(
+            self.benchmark_returns,
+            self.treasury_curves,
+            period_start=self.start_date,
+            period_end=self.end_date,
+            capital_base=1000.0
+        )
+
+        self.oneday = datetime.timedelta(days=1)
+
+    def test_risk_metrics_returns(self):
+        risk_metrics_refactor = risk.RiskMetricsIterative(
+            self.start_date, self.trading_env)
+
+        todays_date = self.start_date
+
+        cur_returns = []
+        for i, ret in enumerate(RETURNS):
+            todays_return_obj = risk.DailyReturn(
+                todays_date,
+                ret
+            )
+
+            cur_returns.append(todays_return_obj)
+
+            try:
+                risk_metrics_original = risk.RiskMetricsBatch(
+                    start_date=self.start_date,
+                    end_date=todays_date + self.oneday,
+                    returns=cur_returns,
+                    trading_environment=self.trading_env
+                )
+            except Exception as e:
+                #assert that when original raises exception, same
+                #exception is raised by risk_metrics_refactor
+                np.testing.assert_raises(
+                    type(e), risk_metrics_refactor.update, ret, self.oneday)
+                continue
+
+            risk_metrics_refactor.update(ret, self.oneday)
+
+            todays_date += self.oneday
+
+            self.assertEqual(
+                risk_metrics_original.start_date,
+                risk_metrics_refactor.start_date)
+            self.assertEqual(
+                risk_metrics_original.end_date,
+                risk_metrics_refactor.end_date)
+            self.assertEqual(
+                risk_metrics_original.treasury_duration,
+                risk_metrics_refactor.treasury_duration)
+            self.assertEqual(
+                risk_metrics_original.treasury_curve,
+                risk_metrics_refactor.treasury_curve)
+            self.assertEqual(
+                risk_metrics_original.treasury_period_return,
+                risk_metrics_refactor.treasury_period_return)
+            self.assertEqual(
+                risk_metrics_original.benchmark_returns,
+                risk_metrics_refactor.benchmark_returns)
+            self.assertEqual(
+                risk_metrics_original.algorithm_returns,
+                risk_metrics_refactor.algorithm_returns)
+            risk_original_dict = risk_metrics_original.to_dict()
+            risk_refactor_dict = risk_metrics_refactor.to_dict()
+            self.assertEqual(set(risk_original_dict.keys()),
+                             set(risk_refactor_dict.keys()))
+
+            err_msg_format = \
+"In update step {iter}: {measure} should be {truth} but is {returned}!"
+
+            for measure in risk_original_dict.iterkeys():
+                if measure == 'max_drawdown':
+                    np.testing.assert_almost_equal(
+                        risk_refactor_dict[measure],
+                        risk_original_dict[measure],
+                        err_msg=err_msg_format.format(
+                            iter=i,
+                            measure=measure,
+                            truth=risk_original_dict[measure],
+                            returned=risk_refactor_dict[measure]))
+                else:
+                    np.testing.assert_equal(
+                        risk_original_dict[measure],
+                        risk_refactor_dict[measure],
+                        err_msg_format.format(
+                            iter=i,
+                            measure=measure,
+                            truth=risk_original_dict[measure],
+                            returned=risk_refactor_dict[measure])
+                    )

zipline/__init__.py

@@ -5,7 +5,7 @@ Zipline
 # This is *not* a place to dump arbitrary classes/modules for convenience,
 # it is a place to expose the public interfaces.
 
-from utils.protocol_utils import ndict
+from zipline.utils.protocol_utils import ndict
 
 import data
 import finance

zipline/algorithm.py

@@ -105,6 +105,9 @@ class TradingAlgorithm(object):
         """
         return self._create_generator(environment)
 
+    def initialize(self, *args, **kwargs):
+        pass
+
     # TODO: make a new subclass, e.g. BatchAlgorithm, and move
     # the run method to the subclass, and refactor to put the
     # generator creation logic into get_generator.

zipline/finance/performance.py

@@ -175,6 +175,8 @@ class PerformanceTracker(object):
         self.txn_count = 0
         self.event_count = 0
         self.last_dict = None
+        self.cumulative_risk_metrics = risk.RiskMetricsIterative(
+            self.period_start, self.trading_environment)
 
         # this performance period will span the entire simulation.
         self.cumulative_performance = PerformancePeriod(
@@ -273,13 +275,9 @@ class PerformanceTracker(object):
         )
         self.returns.append(todays_return_obj)
 
-        #calculate risk metrics for cumulative performance
-        self.cumulative_risk_metrics = risk.RiskMetrics(
-            start_date=self.period_start,
-            end_date=self.market_close.replace(hour=0, minute=0, second=0),
-            returns=self.returns,
-            trading_environment=self.trading_environment
-        )
+        #update risk metrics for cumulative performance
+        self.cumulative_risk_metrics.update(
+            self.todays_performance.returns, datetime.timedelta(days=1))
 
         # increment the day counter before we move markers forward.
         self.day_count += 1.0

zipline/finance/risk.py

@@ -77,7 +77,7 @@ def advance_by_months(dt, jump_in_months):
     return dt.replace(year=dt.year + years, month=month)
 
 
-class DailyReturn():
+class DailyReturn(object):
 
     def __init__(self, date, returns):
 
@@ -95,7 +95,7 @@ class DailyReturn():
         return str(self.date) + " - " + str(self.returns)
 
 
-class RiskMetrics():
+class RiskMetricsBase(object):
     def __init__(self, start_date, end_date, returns, trading_environment):
 
         self.treasury_curves = trading_environment.treasury_curves
@@ -216,8 +216,6 @@ class RiskMetrics():
         return period_returns, returns
 
     def calculate_volatility(self, daily_returns):
-        # TODO: we should be using an annualized number for the
-        # square root, not the days in the period.
         return np.std(daily_returns, ddof=1) * math.sqrt(self.trading_days)
 
     def calculate_sharpe(self):
@@ -228,7 +226,7 @@ class RiskMetrics():
             return 0.0
 
         return ((self.algorithm_period_returns - self.treasury_period_return) /
-                self.algorithm_volatility)
+                 self.algorithm_volatility)
 
     def calculate_beta(self):
         """
@@ -266,8 +264,7 @@ class RiskMetrics():
         http://en.wikipedia.org/wiki/Alpha_(investment)
         """
         return self.algorithm_period_returns - \
-            (self.treasury_period_return +
-             self.beta *
+            (self.treasury_period_return + self.beta *
              (self.benchmark_period_returns - self.treasury_period_return))
 
     def calculate_max_drawdown(self):
@@ -275,12 +272,13 @@ class RiskMetrics():
         cur_return = 0.0
         for r in self.algorithm_returns:
             try:
-                cur_return = math.log(1.0 + r) + cur_return
+                cur_return += math.log(1.0 + r)
             #this is a guard for a single day returning -100%
             except ValueError:
                 log.debug("{cur} return, zeroing the returns".format(
                     cur=cur_return))
                 cur_return = 0.0
+                # BUG? Shouldn't this be set to log(1.0 + 0) ?
             compounded_returns.append(cur_return)
 
         cur_max = None
@@ -327,9 +325,8 @@ class RiskMetrics():
         # in case end date is not a trading day, search for the next market
         # day for an interest rate
         for i in xrange(7):
-            day = self.end_date + i * one_day
-            if day in self.treasury_curves:
-                curve = self.treasury_curves[day]
+            if (self.end_date + i * one_day) in self.treasury_curves:
+                curve = self.treasury_curves[self.end_date + i * one_day]
                 self.treasury_curve = curve
                 rate = self.treasury_curve[self.treasury_duration]
                 # 1month note data begins in 8/2001,
@@ -349,8 +346,213 @@ class RiskMetrics():
         raise Exception(message)
 
 
-class RiskReport():
+class RiskMetricsIterative(RiskMetricsBase):
+    """Iterative version of RiskMetrics.
+    Should behave exaclty like RiskMetricsBatch.
 
+    :Usage:
+        Instantiate RiskMetricsIterative once.
+        Call update() method on each dt to update the metrics.
+    """
+
+    def __init__(self, start_date, trading_environment):
+        self.treasury_curves = trading_environment.treasury_curves
+        self.start_date = start_date
+        self.end_date = start_date
+        self.trading_environment = trading_environment
+
+        self.compounded_log_returns = []
+        self.moving_avg = []
+
+        self.algorithm_returns = []
+        self.benchmark_returns = []
+        self.algorithm_volatility = []
+        self.benchmark_volatility = []
+        self.algorithm_period_returns = []
+        self.benchmark_period_returns = []
+        self.sharpe = []
+        self.beta = []
+        self.alpha = []
+        self.max_drawdown = 0
+        self.current_max = -np.inf
+        self.excess_returns = []
+        self.last_dt = start_date
+        self.trading_days = 0
+
+        self.all_benchmark_returns = [
+                    x for x in self.trading_environment.benchmark_returns
+                    if x.date >= self.start_date
+        ]
+
+    def update(self, returns_in_period, dt):
+        if self.trading_environment.is_trading_day(self.end_date):
+            self.algorithm_returns.append(returns_in_period)
+            self.benchmark_returns.append(
+                self.all_benchmark_returns.pop(0).returns)
+            self.trading_days += 1
+            self.update_compounded_log_returns()
+
+        self.end_date += dt
+        self.end_date = self.end_date.replace(hour=0, minute=0, second=0)
+
+        self.algorithm_period_returns.append(
+            self.calculate_period_returns(self.algorithm_returns))
+        self.benchmark_period_returns.append(
+            self.calculate_period_returns(self.benchmark_returns))
+
+        if(len(self.benchmark_returns) != len(self.algorithm_returns)):
+            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=self.start_date,
+                end=self.end_date
+            )
+            raise Exception(message)
+
+        self.update_current_max()
+        self.benchmark_volatility.append(
+            self.calculate_volatility(self.benchmark_returns))
+        self.algorithm_volatility.append(
+            self.calculate_volatility(self.algorithm_returns))
+        self.treasury_period_return = self.choose_treasury()
+        self.excess_returns.append(
+            self.algorithm_period_returns[-1] - self.treasury_period_return)
+        self.beta.append(self.calculate_beta()[0])
+        self.alpha.append(self.calculate_alpha())
+        self.sharpe.append(self.calculate_sharpe())
+        self.max_drawdown = self.calculate_max_drawdown()
+
+    def to_dict(self):
+        """
+        Creates a dictionary representing the state of the risk report.
+        Returns a dict object of the form:
+        """
+        period_label = self.end_date.strftime("%Y-%m")
+        rval = {
+            'trading_days': self.trading_days,
+            'benchmark_volatility': self.benchmark_volatility[-1],
+            'algo_volatility': self.algorithm_volatility[-1],
+            'treasury_period_return': self.treasury_period_return,
+            'algorithm_period_return': self.algorithm_period_returns[-1],
+            'benchmark_period_return': self.benchmark_period_returns[-1],
+            'sharpe': self.sharpe[-1],
+            'beta': self.beta[-1],
+            'alpha': self.alpha[-1],
+            'excess_return': self.excess_returns[-1],
+            'max_drawdown': self.max_drawdown,
+            'period_label': period_label
+        }
+
+        # check if a field in rval is nan, and replace it with
+        # None.
+        def check_entry(key, value):
+            if key != 'period_label':
+                return np.isnan(value)
+            else:
+                return False
+
+        return {k: None
+                if check_entry(k, v)
+                else v for k, v in rval.iteritems()}
+
+    def __repr__(self):
+        statements = []
+        metrics = [
+            "algorithm_period_returns",
+            "benchmark_period_returns",
+            "excess_returns",
+            "trading_days",
+            "benchmark_volatility",
+            "algorithm_volatility",
+            "sharpe",
+            "algorithm_covariance",
+            "benchmark_variance",
+            "beta",
+            "alpha",
+            "max_drawdown",
+            "algorithm_returns",
+            "benchmark_returns",
+            "condition_number",
+            "eigen_values"
+        ]
+
+        for metric in metrics:
+            value = getattr(self, metric)
+            if isinstance(value, list):
+                if len(value) == 0:
+                    value = np.nan
+                else:
+                    value = value[-1]
+            statements.append("{m}:{v}".format(m=metric, v=value))
+
+        return '\n'.join(statements)
+
+    def update_compounded_log_returns(self):
+        if len(self.algorithm_returns) == 0:
+            return
+        elif len(self.compounded_log_returns) == 0:
+            self.compounded_log_returns.append(
+                math.log(1 + self.algorithm_returns[-1]))
+        else:
+            self.compounded_log_returns.append(
+                self.compounded_log_returns[-1] +
+                math.log(1 + self.algorithm_returns[-1]))
+
+    def calculate_period_returns(self, returns):
+        period_returns = 1.0
+
+        for r in returns:
+            period_returns *= (1.0 + r)
+
+        period_returns -= 1.0
+        return period_returns
+
+    def update_current_max(self):
+        if len(self.compounded_log_returns) == 0:
+            return
+        if self.current_max < self.compounded_log_returns[-1]:
+            self.current_max = self.compounded_log_returns[-1]
+
+    def calculate_max_drawdown(self):
+        if len(self.compounded_log_returns) == 0:
+            return self.max_drawdown
+
+        cur_drawdown = 1.0 - math.exp(
+            self.compounded_log_returns[-1] -
+            self.current_max)
+
+        if self.max_drawdown < cur_drawdown:
+            return cur_drawdown
+        else:
+            return self.max_drawdown
+
+    def calculate_sharpe(self):
+        """
+        http://en.wikipedia.org/wiki/Sharpe_ratio
+        """
+        if self.algorithm_volatility[-1] == 0:
+            return 0.0
+
+        return (self.algorithm_period_returns[-1] -
+                self.treasury_period_return) / self.algorithm_volatility[-1]
+
+    def calculate_alpha(self):
+        """
+        http://en.wikipedia.org/wiki/Alpha_(investment)
+        """
+        return (self.algorithm_period_returns[-1] -
+                (self.treasury_period_return + self.beta[-1] *
+                 (self.benchmark_period_returns[-1] -
+                  self.treasury_period_return)))
+
+
+class RiskMetricsBatch(RiskMetricsBase):
+    pass
+
+
+class RiskReport(object):
     def __init__(
         self,
         algorithm_returns,
@@ -372,10 +574,11 @@ class RiskReport():
             start_date = self.algorithm_returns[0].date
             end_date = self.algorithm_returns[-1].date
 
-        self.month_periods = self.periodsInRange(1, start_date, end_date)
-        self.three_month_periods = self.periodsInRange(3, start_date, end_date)
-        self.six_month_periods = self.periodsInRange(6, start_date, end_date)
-        self.year_periods = self.periodsInRange(12, start_date, end_date)
+        self.month_periods = self.periods_in_range(1, start_date, end_date)
+        self.three_month_periods = self.periods_in_range(
+            3, start_date, end_date)
+        self.six_month_periods = self.periods_in_range(6, start_date, end_date)
+        self.year_periods = self.periods_in_range(12, start_date, end_date)
 
     def to_dict(self):
         """
@@ -400,7 +603,7 @@ class RiskReport():
             'created': self.created
         }
 
-    def periodsInRange(self, months_per, start, end):
+    def periods_in_range(self, months_per, start, end):
         one_day = datetime.timedelta(days=1)
         ends = []
         cur_start = start.replace(day=1)
@@ -417,7 +620,7 @@ class RiskReport():
             cur_end = advance_by_months(cur_start, months_per) - one_day
             if(cur_end > the_end):
                 break
-            cur_period_metrics = RiskMetrics(
+            cur_period_metrics = RiskMetricsBatch(
                 start_date=cur_start,
                 end_date=cur_end,
                 returns=self.algorithm_returns,

zipline/gens/tradegens.py

@@ -249,6 +249,7 @@ class DataFrameSource(SpecificEquityTrades):
                     event = copy(event)
                     event['sid'] = sid
                     event['price'] = price
+                    event['volume'] = 1000
 
                     yield ndict(event)
 

zipline/optimize/example.py

@@ -24,6 +24,7 @@ import numpy as np
 import matplotlib.pyplot as plt
 from zipline.gens.mavg import MovingAverage
 from zipline.algorithm import TradingAlgorithm
+from zipline.gens.transform import batch_transform
 
 
 class DMA(TradingAlgorithm):
@@ -62,6 +63,37 @@ class DMA(TradingAlgorithm):
                 self.invested[sid] = False
 
 
+class DualMovingAverage(TradingAlgorithm):
+    """Dual Moving Average algorithm.
+    """
+    def initialize(self, short_window=200, long_window=400):
+        self.short_mavg = []
+        self.long_mavg = []
+
+        self.invested = False
+
+        self.add_transform(MovingAverage, 'short_mavg', ['price'],
+                           market_aware=True,
+                           days=short_window)
+
+        self.add_transform(MovingAverage, 'long_mavg', ['price'],
+                           market_aware=True,
+                           days=long_window)
+
+    def handle_data(self, data):
+        self.short_mavg.append(data['AAPL'].short_mavg['price'])
+        self.long_mavg.append(data['AAPL'].long_mavg['price'])
+
+        if (data['AAPL'].short_mavg['price'] >
+            data['AAPL'].long_mavg['price']) and not self.invested:
+            self.order('AAPL', 100)
+            self.invested = True
+        elif (data['AAPL'].short_mavg['price'] <
+              data['AAPL'].long_mavg['price']) and self.invested:
+            self.order('AAPL', -100)
+            self.invested = False
+
+
 def load_close_px(indexes=None, stocks=None):
     from pandas.io.data import DataReader
     import pytz
@@ -70,10 +102,10 @@ def load_close_px(indexes=None, stocks=None):
     if indexes is None:
         indexes = {'SPX': '^GSPC'}
     if stocks is None:
-        stocks = ['AAPL', 'GE', 'IBM', 'MSFT', 'XOM', 'AA', 'JNJ', 'PEP']
+        stocks = ['AAPL', 'GE', 'IBM', 'MSFT', 'XOM', 'AA', 'JNJ', 'PEP', 'KO']
 
     start = pd.datetime(1990, 1, 1, 0, 0, 0, 0, pytz.utc)
-    end = pd.datetime(1992, 1, 1, 0, 0, 0, 0, pytz.utc)
+    end = pd.datetime(2000, 1, 1, 0, 0, 0, 0, pytz.utc)
 
     data = OrderedDict()
 
@@ -87,8 +119,8 @@ def load_close_px(indexes=None, stocks=None):
         stkd = DataReader(ticker, 'yahoo', start, end).sort_index()
         data[name] = stkd
 
-    #df = pd.DataFrame({key: d['Close'] for key, d in data.iteritems()})
-    df = pd.DataFrame({i: d['Close'] for i, d in enumerate(data.itervalues())})
+    df = pd.DataFrame({key: d['Close'] for key, d in data.iteritems()})
+
     df.index = df.index.tz_localize(pytz.utc)
 
     df.save('close_px.dat')
@@ -171,4 +203,55 @@ def plot_returns(port_returns, bmk_returns):
     plt.title('Portfolio performance')
     plt.legend(loc='best')
 
-print run((10, 20))
+#print run((10, 20))
+
+import statsmodels.api as sm
+
+
+@batch_transform
+def ols_transform(data, spreads):
+    p0 = data.price['PEP']
+    p1 = sm.add_constant(data.price['KO'])
+    beta, intercept = sm.OLS(p0, p1).fit().params
+
+    spread = (data.price['PEP'] - (beta * data.price['KO'] + intercept))[-1]
+
+    if len(spreads) > 10:
+        z_score = (spread - np.mean(spreads[-10:])) / np.std(spreads[-10:])
+    else:
+        z_score = np.nan
+
+    spreads.append(spread)
+
+    return z_score
+
+
+class Pairtrade(TradingAlgorithm):
+    def initialize(self):
+        self.spreads = []
+        self.invested = False
+        self.ols_transform = ols_transform(refresh_period=10, days=10)
+
+    def handle_data(self, data):
+        zscore = self.ols_transform.handle_data(data, self.spreads)
+
+        if zscore == np.nan:
+            return
+
+        if zscore >= 2.0 and not self.invested:
+            self.order('PEP', int(100 / data['PEP'].price))
+            self.order('KO', -int(100 / data['KO'].price))
+        elif zscore <= -2.0 and not self.invested:
+            self.order('KO', -int(100 / data['KO'].price))
+            self.order('PEP', int(100 / data['PEP'].price))
+        elif abs(zscore) < .5 and self.invested:
+            pass
+
+
+def run_pairtrade():
+    data = load_close_px()
+    data.save('close_px.dat')
+    #data = pd.load('close_px.dat')
+    myalgo = Pairtrade()
+    stats = myalgo.run(data)
+    return stats