Merge branch 'optimize3' into batch_window

tests/test_optimize.py

@@ -7,7 +7,7 @@ import numpy as np
 
 from zipline.core.devsimulator import AddressAllocator
 # TODO: refactor the factory to use generators
-# from zipline.optimize.factory import create_predictable_zipline
+from zipline.optimize.factory import create_predictable_zipline
 
 DEFAULT_TIMEOUT = 15 # seconds
 EXTENDED_TIMEOUT = 90
@@ -24,7 +24,7 @@ class TestUpDown(TestCase):
 
     def setUp(self):
         self.zipline_test_config = {
-            'sid'         : 133,
+            'sid'         : [0],
             'trade_count' : 5,
             'amplitude'   : 30,
             'base_price'  : 50
@@ -44,17 +44,17 @@ class TestUpDown(TestCase):
         UpDownSource and BuySellAlgorithm interact correctly."
 
         """
-        zipline, config = create_predictable_zipline(
+
+        algo, config = create_predictable_zipline(
             self.zipline_test_config,
-            offset=0,
-            simulate=False
+            offset=0
         )
 
         #extract arguments
         base_price = self.zipline_test_config['base_price']
         amplitude = self.zipline_test_config['amplitude']
 
-        prices = np.array([event.price for event in config['trade_source'].event_list])
+        prices = config['trade_source'][0].values
         max_price_idx = np.where(prices==prices.max())[0]
         min_price_idx = np.where(prices==prices.min())[0]
         self.assertTrue(np.all(max_price_idx % 2 == 1),
@@ -70,9 +70,9 @@ class TestUpDown(TestCase):
             "Minimum price does not equal expected maximum price."
         )
 
-        zipline.simulate(blocking=True)
+        stats = algo.run(config['trade_source'])
 
-        algo = config['algorithm']
+        self.assertTrue(len(stats) != 0)
 
         orders = np.asarray(algo.orders)
         max_order_idx = np.where(orders==orders.max())[0]
@@ -108,12 +108,15 @@ class TestUpDown(TestCase):
         compound_returns = np.empty(len(test_offsets))
         ziplines = []
         for i, offset in enumerate(test_offsets):
-            zipline, config = create_predictable_zipline(
+            algo, config = create_predictable_zipline(
                 self.zipline_test_config,
                 offset=offset,
             )
-            ziplines.append(zipline)
-            compound_returns[i] = zipline.get_cumulative_performance()['returns']
+            results = algo.run(config['trade_source'])
+            ziplines.append(algo)
+
+            compound_returns[i] = results.returns.sum()
+
 
         self.assertTrue(np.all(compound_returns[supposed_max] > compound_returns[np.logical_not(supposed_max)]),
             "Maximum compound returns are not where they are supposed to be."

zipline/gens/composites.py

@@ -14,7 +14,7 @@ def date_sorted_sources(*sources):
 
     for source in sources:
         assert iter(source), "Source %s not iterable" % source
-        assert 'get_hash' in source.__class__.__dict__, "No get_hash"
+        assert hasattr(source, 'get_hash'), "No get_hash"
 
     # Get name hashes to pass to date_sort.
     names = [source.get_hash() for source in sources]

zipline/gens/tradegens.py

@@ -7,7 +7,11 @@ import pytz
 
 from itertools import chain, cycle, ifilter, izip, repeat
 from datetime import datetime, timedelta
+import pandas as pd
+from copy import copy
 
+from zipline.protocol import DATASOURCE_TYPE
+from zipline.utils import ndict
 from zipline.gens.utils import hash_args, create_trade
 
 def date_gen(start = datetime(2006, 6, 6, 12, tzinfo=pytz.utc),
@@ -78,7 +82,6 @@ class SpecificEquityTrades(object):
         self.sids = kwargs.get('sids', [1, 2])
         self.start = kwargs.get('start', datetime(2008, 6, 6, 15, tzinfo = pytz.utc))
         self.delta = kwargs.get('delta', timedelta(minutes = 1))
-        self.concurrent = kwargs.get('concurrent', False)
 
         # Default to None for event_list and filter.
         self.event_list = kwargs.get('event_list')
@@ -137,6 +140,7 @@ class SpecificEquityTrades(object):
             volumes = mock_volumes(self.count)
 
             sids = cycle(self.sids)
+
             # Combine the iterators into a single iterator of arguments
             arg_gen = izip(sids, prices, volumes, dates)
 
@@ -157,33 +161,57 @@ class SpecificEquityTrades(object):
         return filtered
 
 
-# !!!!!!! Deprecated for now !!!!!!!!!
+class DataFrameSource(SpecificEquityTrades):
+    """
+    Yields all events in event_list that match the given sid_filter.
+    If no event_list is specified, generates an internal stream of events
+    to filter.  Returns all events if filter is None.
 
-def RandomEquityTrades(object):
+    Configuration options:
 
-    def __init__(self):
-        # We shouldn't get any positional args.
-        assert args == ()
+    count  : integer representing number of trades
+    sids   : list of values representing simulated internal sids
+    start  : start date
+    delta  : timedelta between internal events
+    filter : filter to remove the sids
+    """
 
-        self.count = config.get('count', 500)
-        self.sids = config.get('sids', [1,2])
-        self.filter = config.get('filter')
+    def __init__(self, data, **kwargs):
+        assert isinstance(data.index, pd.tseries.index.DatetimeIndex)
 
-        dates = fuzzy_dates(count)
-        prices = mock_prices(count, rand = True)
-        volumes = mock_volumes(count, rand = True)
-        sids = cycle(sids)
+        self.data = data
+        # Unpack config dictionary with default values.
+        self.count = kwargs.get('count', 500)
+        self.sids = kwargs.get('sids', [0])
+        self.start = kwargs.get('start', datetime(1957, 1, 1, 0, tzinfo = pytz.utc))
+        self.end = kwargs.get('end', datetime(2010, 1, 1, tzinfo=pytz.utc))
+        self.delta = kwargs.get('delta', timedelta(days = 1))
 
-    arg_gen = izip(sids, prices, volumes, dates)
+        # Default to None for event_list and filter.
+        self.filter = kwargs.get('filter')
 
-    unfiltered = (create_trade(*args) for args in arg_gen)
+        # Hash_value for downstream sorting.
+        self.arg_string = hash_args(data, **kwargs)
 
-    if filter:
-        filtered = ifilter(lambda event: event.sid in filter, unfiltered)
-    else:
-        filtered = unfiltered
-    return filtered
+        self.generator = self.create_fresh_generator()
 
-# if __name__ == "__main__":
-#     import nose.tools; nose.tools.set_trace()
-#     trades = SpecificEquityTrades(filter = [1])
+    def create_fresh_generator(self):
+        def _generator(df=self.data):
+            for dt, series in df.iterrows():
+                if (dt < self.start) or (dt > self.end):
+                    continue
+                event = {'dt': dt,
+                         'source_id': self.get_hash(),
+                         'type': DATASOURCE_TYPE.TRADE
+                }
+
+                for sid, price in series.iterkv():
+                    event = copy(event)
+                    event['sid'] = 0
+                    event['price'] = price
+
+                    yield ndict(event)
+
+
+        # Return the filtered event stream.
+        return _generator()

zipline/gens/tradesimulation.py

@@ -193,6 +193,8 @@ class AlgorithmSimulator(object):
             'filled' : 0
         })
 
+        log.debug(order)
+
         # Tell the user if they try to buy 0 shares of something.
         if order.amount == 0:
             zero_message = "Requested to trade zero shares of {sid}".format(

zipline/optimize/algorithms.py

@@ -1,3 +1,18 @@
+import pandas as pd
+import numpy as np
+
+from datetime import datetime
+from zipline.gens.tradegens import DataFrameSource
+from zipline import ndict
+from zipline.utils.factory import create_trading_environment
+from zipline.gens.transform import StatefulTransform
+from zipline.lines import SimulatedTrading
+from zipline.finance.slippage import FixedSlippage
+
+from logbook import Logger
+
+logger = Logger('Algo')
+
 class BuySellAlgorithm(object):
     """Algorithm that buys and sells alternatingly. The amount for
     each order can be specified. In addition, an offset that will
@@ -46,3 +61,132 @@ class BuySellAlgorithm(object):
 
     def get_sid_filter(self):
         return [self.sid]
+
+# Algorithm base class, user algorithms inherit from this as they
+# don't want to have to copy and know about set_order and
+# set_portfolio
+class TradingAlgorithm(object):
+    def _setup(self):
+        assert hasattr(self, 'source'), 'source not set.'
+        assert hasattr(self, 'sids'), "sids not set."
+
+        environment = create_trading_environment(start=self.data.index[0], end=self.data.index[-1])
+
+        # Create transforms by wrapping them into StatefulTransforms
+        transforms = []
+        if hasattr(self, 'registered_transforms'):
+            for namestring, trans_descr in self.registered_transforms.iteritems():
+                sf = StatefulTransform(
+                    trans_descr['class'],
+                    *trans_descr['args'],
+                    **trans_descr['kwargs']
+                )
+                sf.namestring = namestring
+
+                transforms.append(sf)
+
+
+        self.simulated_trading = SimulatedTrading(
+            [self.source],
+            transforms,
+            self,
+            environment,
+            FixedSlippage()
+        )
+
+    def _create_daily_stats(self, perfs):
+        # create daily stats dataframe
+        daily_perfs = []
+        cum_perfs = []
+        for perf in perfs:
+            if 'daily_perf' in perf:
+                daily_perfs.append(perf['daily_perf'])
+            else:
+                cum_perfs.append(perf)
+
+        daily_dts = [np.datetime64(perf['period_close'], utc=True) for perf in daily_perfs]
+        daily_stats = pd.DataFrame(daily_perfs, index=daily_dts)
+
+        return daily_stats
+
+    def run(self, data, compute_risk_metrics=False):
+        self.source = DataFrameSource(data, sids=self.sids)
+        self.data = data
+        self._setup()
+
+        # drain simulated_trading
+        perfs = []
+        for perf in self.simulated_trading:
+            #from nose.tools import set_trace; set_trace()
+            perfs.append(perf)
+
+        #perfs = list(self.simulated_trading)
+
+        daily_stats = self._create_daily_stats(perfs)
+        return daily_stats
+
+    def set_portfolio(self, portfolio):
+        self.portfolio = portfolio
+
+    def set_order(self, order_callable):
+        self.order = order_callable
+
+    def get_sid_filter(self):
+        return self.sids
+
+    def set_logger(self, logger):
+        self.logger = logger
+
+    def initialize(self):
+        pass
+
+    def set_slippage_override(self, slippage_callable):
+        pass
+
+    def add_transform(self, transform_class, tag, *args, **kwargs):
+        if not hasattr(self, 'registered_transforms'):
+            self.registered_transforms = {}
+
+        self.registered_transforms[tag] = {'class': transform_class,
+                                           'args': args,
+                                           'kwargs': kwargs}
+
+
+class BuySellAlgorithmNew(TradingAlgorithm):
+    """Algorithm that buys and sells alternatingly. The amount for
+    each order can be specified. In addition, an offset that will
+    quadratically reduce the amount that will be bought can be
+    specified.
+
+    This algorithm is used to test the parameter optimization
+    framework. If combined with the UpDown trade source, an offset of
+    0 will produce maximum returns.
+
+    """
+
+    def __init__(self, sids, amount, offset):
+        self.sids = sids
+        self.amount = amount
+        self.incr = 0
+        self.done = False
+        self.order = None
+        self.frame_count = 0
+        self.portfolio = None
+        self.buy_or_sell = -1
+        self.offset = offset
+        self.orders = []
+        self.prices = []
+
+    def handle_data(self, data):
+        order_size = self.buy_or_sell * (self.amount - (self.offset**2))
+        self.order(self.sids[0], order_size)
+        logger.debug("ordering" + str(order_size))
+
+        #sell next time around.
+        self.buy_or_sell *= -1
+
+        self.orders.append(order_size)
+
+        self.frame_count += 1
+        self.incr += 1
+

zipline/optimize/example.py

@@ -0,0 +1,155 @@
+import pandas as pd
+
+import numpy as np
+#from mpl_toolkits.mplot3d import Axes3D
+
+import matplotlib.pyplot as plt
+import cProfile
+from zipline.gens.mavg import MovingAverage
+from zipline.optimize.algorithms import TradingAlgorithm
+from datetime import timedelta
+
+#from mpi4py_map import map
+
+# Inherits from Algorithm base class
+class DMA(TradingAlgorithm):
+    """Dual Moving Average algorithm.
+    """
+    def __init__(self, sids, amount=100, short_window=20, long_window=40):
+        self.sids = sids
+        self.amount = amount
+        self.done = False
+        self.order = None
+        self.frame_count = 0
+        self.portfolio = None
+        self.orders = []
+
+        self.prices = []
+        self.events = 0
+
+        self.invested = {}
+        for sid in self.sids:
+            self.invested[sid] = False
+
+        self.add_transform(MovingAverage, 'short_mavg', ['price'],
+                           market_aware=True,
+                           days=short_window) #timedelta(days=int(short_window)))
+
+        self.add_transform(MovingAverage, 'long_mavg', ['price'],
+                           market_aware=True,
+                           days=long_window) #timedelta(days=int(long_window)))
+
+    def handle_data(self, data):
+        self.events += 1
+
+        for sid in self.sids:
+            # access transforms via their user-defined tag
+            if (data[sid].short_mavg['price'] > data[sid].long_mavg['price']) and not self.invested[sid]:
+                self.order(sid, self.amount)
+                self.invested[sid] = True
+            elif (data[sid].short_mavg['price'] < data[sid].long_mavg['price']) and self.invested[sid]:
+                self.order(sid, -self.amount)
+                self.invested[sid] = False
+
+
+def load_close_px(indexes=None, stocks=None):
+    from pandas.io.data import DataReader
+    import pytz
+
+    if indexes is None:
+        indexes = {'SPX' : '^GSPC'}
+    if stocks is None:
+        stocks = ['AAPL'] #, 'GE', 'IBM', 'MSFT', 'XOM', 'AA', 'JNJ', 'PEP']
+
+    #start = pd.datetime(1990, 1, 1)
+    start = pd.datetime(1990, 1, 1, 0, 0, 0, 0, pytz.utc)
+    end = pd.datetime(1992, 1, 1, 0, 0, 0, 0, pytz.utc) #pd.datetime.today()
+
+    data = {}
+
+    for stock in stocks:
+        print stock
+        stkd = DataReader(stock, 'yahoo', start, end).sort_index()
+        data[stock] = stkd
+
+    for name, ticker in indexes.iteritems():
+        print name
+        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.index = df.index.tz_localize(pytz.utc)
+
+    return df
+
+
+def run((short_window, long_window)):
+    #data = pd.DataFrame.from_csv('SP500.csv')
+    data = load_close_px()
+    myalgo = DMA([0], amount=100, short_window=short_window, long_window=long_window)
+    stats = myalgo.run(data)
+    stats['sw'] = short_window
+    stats['lw'] = long_window
+    return stats
+
+def explore_params():
+    sws, lws = np.mgrid[10:20:5, 10:20:5]
+
+    stats_all = map(run, zip(sws.flatten(), lws.flatten()))
+    stats = pd.concat(stats_all)
+    returns = stats.groupby(['sw', 'lw']).sum()
+
+    plt.contourf(sws, lws, returns.returns.reshape(sws.shape))
+    plt.xlabel('Short window length')
+    plt.ylabel('Long window length')
+    plt.savefig('DMA_contour.png')
+    plt.show()
+
+#stats = run((10, 50))
+
+def get_opt_holdings_qp(univ_rets, track_rets):
+    from cvxopt import matrix
+    from cvxopt.solvers import qp
+    # set up the QP for CVXOPT
+    # .5 x' P x + q'x
+    # P = 2 * R'R
+    # q = - 2 * bmk'R
+    R = univ_rets.values
+    b = track_rets.values
+    P = matrix(2 * np.dot(R.T, R))
+    q = matrix(-2 * np.dot(R.T, b))
+    result = qp(P, q)
+    if result['status'] != 'optimal':
+        raise Exception('optimum not reached by QP')
+    return pd.Series(np.array(result['x']).ravel(), index=univ_rets.columns)
+
+def opt_portfolio(cov, budget, min_return):
+    from cvxopt import matrix
+    from cvxopt.solvers import qp
+    n = len(cov)
+    cov = matrix(2 * cov)
+    q = matrix(np.zeros(n))
+
+    h = matrix(budget) # G*x < h
+    # coneqp
+    result = qp(cov, q, h=h)
+    if result['status'] != 'optimal':
+        raise Exception('optimum not reached by QP')
+
+    return pd.Series(np.array(result['x']).ravel())
+
+def calc_te(weights, univ_rets, track_rets):
+    port_rets = (univ_rets * weights).sum(1)
+    return (port_rets - track_rets).std()
+
+def plot_returns(port_returns, bmk_returns):
+    plt.figure()
+    cum_port = ((1 + port_returns).cumprod() - 1)
+    cum_bmk = ((1 + bmk_returns).cumprod() - 1)
+    # cum_port = port_returns.cumsum()
+    # cum_bmk = bmk_returns.cumsum()
+    cum_port.plot(label='Portfolio returns')
+    cum_bmk.plot(label='Benchmark')
+    plt.title('Portfolio performance')
+    plt.legend(loc='best')

zipline/optimize/factory.py

@@ -122,7 +122,7 @@ def create_predictable_zipline(config, offset=0, simulate=True):
                                         amplitude)
 
     if 'algorithm' not in config:
-        config['algorithm'] = BuySellAlgorithm(sid, 100, offset)
+        algorithm = BuySellAlgorithmNew(sid, 100, offset)
 
     config['order_count'] = trade_count - 1
     config['trade_count'] = trade_count
@@ -131,9 +131,4 @@ def create_predictable_zipline(config, offset=0, simulate=True):
     config['slippage'] = FixedSlippage()
     config['devel'] = True
 
-    zipline = SimulatedTrading.create_test_zipline(**config)
-
-    if simulate:
-        zipline.simulate(blocking=True)
-
-    return zipline, config
+    return algorithm, config

zipline/test_algorithms.py

@@ -52,7 +52,6 @@ The algorithm must expose methods:
 
 """
 
-
 class TestAlgorithm():
     """
     This algorithm will send a specified number of orders, to allow unit tests
@@ -60,8 +59,7 @@ class TestAlgorithm():
     at the close of a simulation.
     """
 
-    def __init__(self, sid, amount, order_count, sid_filter=None):
-        self.count = order_count
+    def __init__(self, sid, amount, sid_filter=None):
         self.sid = sid
         self.amount = amount
         self.incr = 0
@@ -69,6 +67,7 @@ class TestAlgorithm():
         self.order = None
         self.frame_count = 0
         self.portfolio = None
+
         if sid_filter:
             self.sid_filter = sid_filter
         else:

zipline/utils/factory.py

@@ -14,7 +14,6 @@ from zipline.utils.protocol_utils import ndict
 
 import zipline.finance.risk as risk
 
-from zipline.gens.tradegens import RandomEquityTrades
 from zipline.gens.tradegens import SpecificEquityTrades
 from zipline.gens.utils import create_trade
 from zipline.finance.trading import TradingEnvironment
@@ -57,12 +56,15 @@ def load_market_data():
 
     return bm_returns, tr_curves
 
-def create_trading_environment(year=2006):
+def create_trading_environment(year=2006, start=None, end=None):
     """Construct a complete environment with reasonable defaults"""
     benchmark_returns, treasury_curves = load_market_data()
 
-    start = datetime(year, 1, 1, tzinfo=pytz.utc)
-    end   = datetime(year, 12, 31, tzinfo=pytz.utc)
+    if start is None:
+        start = datetime(year, 1, 1, tzinfo=pytz.utc)
+    if end is None:
+        end   = datetime(year, 12, 31, tzinfo=pytz.utc)
+
     trading_environment = TradingEnvironment(
         benchmark_returns,
         treasury_curves,