Large refactoring and documentation of new algorithm base class and batch transform.

tests/test_optimize.py

@@ -5,13 +5,8 @@ from collections import defaultdict
 
 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
 
-DEFAULT_TIMEOUT = 15 # seconds
-EXTENDED_TIMEOUT = 90
-
 from zipline.utils.test_utils import setup_logger, teardown_logger
 
 class TestUpDown(TestCase):
@@ -36,7 +31,6 @@ class TestUpDown(TestCase):
         teardown_logger(self)
 
     @skip
-    @timed(DEFAULT_TIMEOUT)
     def test_source_and_orders(self):
         """verify that UpDownSource is having the correct
         behavior and that BuySellAlgorithm places the buy/sell

tests/test_transforms.py

@@ -1,5 +1,5 @@
 import pytz
-import numpy
+import numpy as np
 
 from datetime import timedelta, datetime
 from unittest2 import TestCase
@@ -10,13 +10,13 @@ from zipline.utils.test_utils import setup_logger
 from zipline.utils.date_utils import utcnow
 
 from zipline.gens.tradegens import SpecificEquityTrades
-from zipline.gens.transform import StatefulTransform, EventWindow
+from zipline.gens.transform import StatefulTransform, EventWindow, BatchTransform, batch_transform
 from zipline.gens.vwap import VWAP
 from zipline.gens.mavg import MovingAverage
 from zipline.gens.stddev import MovingStandardDev
 from zipline.gens.returns import Returns
-
 import zipline.utils.factory as factory
+from zipline import TradingAlgorithm
 
 def to_dt(msg):
     return ndict({'dt': msg})
@@ -42,26 +42,26 @@ class EventWindowTestCase(TestCase):
 
     def setUp(self):
         setup_logger(self)
-        
+
         self.monday = datetime(2012, 7, 9, 16, tzinfo=pytz.utc)
-        self.eleven_normal_days = [self.monday + i*timedelta(days=1) 
+        self.eleven_normal_days = [self.monday + i*timedelta(days=1)
                                    for i in xrange(11)]
 
         # Modify the end of the period slightly to exercise the
         # incomplete day logic.
         self.eleven_normal_days[-1] -= timedelta(minutes = 1)
         self.eleven_normal_days.append(self.monday+timedelta(days=11,seconds=1))
-        
+
         # Second set of dates to test holiday handling.
         self.jul4_monday = datetime(2012, 7, 2, 16, tzinfo=pytz.utc)
         self.week_of_jul4 = [self.jul4_monday + i*timedelta(days=1)
                                    for i in xrange(5)]
 
     def test_event_window_with_timedelta(self):
-        
+
         # Keep all events within a 5 minute window.
         window = NoopEventWindow(
-            market_aware = False, 
+            market_aware = False,
             delta = timedelta(minutes = 5),
             days = None
         )
@@ -91,7 +91,7 @@ class EventWindowTestCase(TestCase):
 
     def test_market_aware_window_normal_week(self):
         window = NoopEventWindow(
-            market_aware = True, 
+            market_aware = True,
             delta = None,
             days = 3
         )
@@ -102,7 +102,7 @@ class EventWindowTestCase(TestCase):
             window.update(event)
             # Record the length of the window after each event.
             lengths.append(len(window.ticks))
-        
+
         # The window stretches out during the weekend because we wait
         # to drop events until the weekend ends. The last window is
         # briefly longer because it doesn't complete a full day.  The
@@ -113,7 +113,7 @@ class EventWindowTestCase(TestCase):
 
     def test_market_aware_window_holiday(self):
         window = NoopEventWindow(
-            market_aware = True, 
+            market_aware = True,
             delta = None,
             days = 2
         )
@@ -125,11 +125,11 @@ class EventWindowTestCase(TestCase):
             window.update(event)
             # Record the length of the window after each event.
             lengths.append(len(window.ticks))
-        
+
         assert lengths == [1, 2, 3, 3, 2]
         assert window.added == events
         assert window.removed == events[:-2]
-        
+
     def tearDown(self):
         setup_logger(self)
 
@@ -186,7 +186,7 @@ class FinanceTransformsTestCase(TestCase):
         expected = [0.0, 0.0, 0.1, 0.0]
 
         assert tnfm_vals == expected
-        
+
         # Two-day returns.  An extra kink here is that the
         # factory will automatically skip a weekend for the
         # last event. Results shouldn't notice this blip.
@@ -222,12 +222,12 @@ class FinanceTransformsTestCase(TestCase):
             fields = ['price', 'volume'],
             delta = timedelta(days = 2),
         )
-        
+
         transformed = list(mavg.transform(self.source))
         # Output values.
         tnfm_prices = [message.tnfm_value.price for message in transformed]
         tnfm_volumes = [message.tnfm_value.volume for message in transformed]
-        
+
         # "Hand-calculated" values
         expected_prices = [
             ((10.0) / 1.0),
@@ -267,16 +267,16 @@ class FinanceTransformsTestCase(TestCase):
         transformed = list(stddev.transform(self.source))
 
         vals = [message.tnfm_value for message in transformed]
-        
+
         expected = [
             None,
-            numpy.std([10.0, 15.0], ddof = 1),
-            numpy.std([10.0, 15.0, 13.0], ddof = 1),
-            numpy.std([15.0, 13.0, 12.0], ddof = 1),
+            np.std([10.0, 15.0], ddof = 1),
+            np.std([10.0, 15.0, 13.0], ddof = 1),
+            np.std([15.0, 13.0, 12.0], ddof = 1),
             ]
 
-        # numpy has odd rounding behavior, cf.
-        # http://docs.scipy.org/doc/numpy/reference/generated/numpy.std.html
+        # np has odd rounding behavior, cf.
+        # http://docs.scipy.org/doc/np/reference/generated/np.std.html
         for v1, v2 in zip(vals, expected):
 
             if v1 == None:
@@ -285,8 +285,68 @@ class FinanceTransformsTestCase(TestCase):
             assert round(v1, 5) == round(v2, 5)
 
 
+############################################################
+# Test BatchTransform
 
-        
-        
-        
+class NoopBatchTransform(BatchTransform):
+    def get_value(self, data):
+        return data.price
+
+@batch_transform
+def noop_batch_decorator(data):
+    return data.price
+
+class BatchTransformAlgorithm(TradingAlgorithm):
+    def initialize(self, *args, **kwargs):
+        self.history_class = []
+        self.history_decorator = []
+        self.days = 3
+        self.noop_class = NoopBatchTransform(sids=[0, 1],
+                                          market_aware=False,
+                                          refresh_period=2,
+                                          delta=timedelta(days=self.days))
+
+        self.noop_decorator = noop_batch_decorator(sids=[0, 1],
+                                                   market_aware=False,
+                                                   refresh_period=2,
+                                                   delta=timedelta(days=self.days))
+
+    def handle_data(self, data):
+        window_class = self.noop_class.handle_data(data)
+        window_decorator = self.noop_decorator.handle_data(data)
+        self.history_class.append(window_class)
+        self.history_decorator.append(window_decorator)
+
+class BatchTransformTestCase(TestCase):
+    def setUp(self):
+        setup_logger(self)
+        self.source, self.df = factory.create_test_df_source()
+
+    def test_batch_inherit(self):
+        algo = BatchTransformAlgorithm(sids=[0, 1])
+        algo.run(self.source)
+
+        assert algo.history_class[:2] == algo.history_decorator[:2] == [None, None]
+
+        # test overloaded class
+        assert np.all(algo.history_class[2][0].values == [4, 6, 8])
+        assert np.all(algo.history_class[2][1].values == [5, 7, 9])
+        assert np.all(algo.history_class[3][0].values == [4, 6, 8, 10])
+        assert np.all(algo.history_class[3][1].values == [5, 7, 9, 11])
+        # not updated because of refresh_period=2
+        assert np.all(algo.history_class[4][0].values == [4, 6, 8, 10])
+        assert np.all(algo.history_class[4][1].values == [5, 7, 9, 11])
+        assert np.all(algo.history_class[5][0].values == [10, 12, 14])
+        assert np.all(algo.history_class[5][1].values == [11, 13, 15])
+
+        # test decorator
+        assert np.all(algo.history_decorator[2][0].values == [4, 6, 8])
+        assert np.all(algo.history_decorator[2][1].values == [5, 7, 9])
+        assert np.all(algo.history_decorator[3][0].values == [4, 6, 8, 10])
+        assert np.all(algo.history_decorator[3][1].values == [5, 7, 9, 11])
+        # not updated because of refresh_period=2
+        assert np.all(algo.history_decorator[4][0].values == [4, 6, 8, 10])
+        assert np.all(algo.history_decorator[4][1].values == [5, 7, 9, 11])
+        assert np.all(algo.history_decorator[5][0].values == [10, 12, 14])
+        assert np.all(algo.history_decorator[5][1].values == [11, 13, 15])
 

zipline/__init__.py

@@ -6,7 +6,9 @@ Zipline
 # it is a place to expose the public interfaces.
 
 from utils.protocol_utils import ndict
+from algorithm import TradingAlgorithm
 
 __all__ = [
-    ndict
+    ndict,
+    TradingAlgorithm
 ]

zipline/algorithm.py

@@ -0,0 +1,163 @@
+import pandas as pd
+import numpy as np
+
+from zipline.gens.tradegens import DataFrameSource
+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
+
+
+class TradingAlgorithm(object):
+    """
+    Base class for trading algorithms. Inherit and overload handle_data(data).
+
+    A new algorithm could look like this:
+    ```
+    class MyAlgo(TradingAlgorithm):
+        def initialize(amount):
+            self.amount = amount
+
+        def handle_data(data):
+            sid = self.sids[0]
+            self.order(sid, amount)
+    ```
+    To then run this algorithm:
+
+    >>> my_algo = MyAlgo(100, sids=[0])
+    >>> stats = my_algo.run(data)
+
+    """
+    def __init__(self, sids, *args, **kwargs):
+        """
+        Initialize sids and other state variables.
+
+        Calls user-defined initialize and forwarding *args and **kwargs.
+        """
+        self.sids = sids
+        self.done = False
+        self.order = None
+        self.frame_count = 0
+        self.portfolio = None
+
+        self.registered_transforms = {}
+
+        # call to user-defined initialize method
+        self.initialize(*args, **kwargs)
+
+    def _create_simulator(self, source):
+        """
+        Create trading environment, transforms and SimulatedTrading object.
+
+        Gets called by self.run().
+        """
+        environment = create_trading_environment(start=source.data.index[0], end=source.data.index[-1])
+
+        # Create transforms by wrapping them into StatefulTransforms
+        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)
+
+        # SimulatedTrading is the main class handling data streaming,
+        # application of transforms and calling of the user algo.
+        return SimulatedTrading(
+            [source],
+            transforms,
+            self,
+            environment,
+            FixedSlippage()
+        )
+
+    def run(self, source):
+        """
+        Run the algorithm.
+
+        :Arguments:
+            data : zipline source or pandas.DataFrame
+               pandas.DataFrame must have the following structure:
+               * column names must consist of ints representing the different sids
+               * index must be TimeStamps
+               * array contents should be price
+
+        :Returns:
+            daily_stats : pandas.DataFrame
+              Daily performance metrics such as returns, alpha etc.
+
+        """
+        if isinstance(source, pd.DataFrame):
+            assert isinstance(source.index, pd.tseries.index.DatetimeIndex)
+            source = DataFrameSource(source, sids=self.sids)
+
+        # create transforms and zipline
+        simulated_trading = self._create_simulator(source)
+
+        # loop through simulated_trading, each iteration returns a
+        # perf ndict
+        perfs = list(simulated_trading)
+
+        # convert perf ndict to pandas dataframe
+        daily_stats = self._create_daily_stats(perfs)
+
+        return daily_stats
+
+
+    def _create_daily_stats(self, perfs):
+        # create daily and cumulative 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 add_transform(self, transform_class, tag, *args, **kwargs):
+        """Add a single-sid, sequential transform to the model.
+
+        :Arguments:
+            transform_class : class
+                Which transform to use. E.g. mavg.
+            tag : str
+                How to name the transform. Can later be access via:
+                data[sid].tag()
+
+        Extra args and kwargs will be forwarded to the transform
+        instantiation.
+
+        """
+        self.registered_transforms[tag] = {'class': transform_class,
+                                           'args': args,
+                                           'kwargs': kwargs}
+
+    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, *args, **kwargs):
+        pass
+
+    def set_slippage_override(self, slippage_callable):
+        pass
+
+
+

zipline/gens/cov.py

@@ -1,9 +0,0 @@
-from zipline.gens.transform import BatchWindow, batch_transform
-
-class CovEventWindow(BatchWindow):
-    def get_value(self, data):
-        return data.cov()
-
-@batch_transform
-def cov(data):
-    return data.cov()

zipline/gens/tradegens.py

@@ -212,6 +212,5 @@ class DataFrameSource(SpecificEquityTrades):
 
                     yield ndict(event)
 
-
         # Return the filtered event stream.
         return _generator()

zipline/gens/tradesimulation.py

@@ -219,61 +219,61 @@ class AlgorithmSimulator(object):
         # Capture any output of this generator to stdout and pipe it
         # to a logbook interface.  Also inject the current algo
         # snapshot time to any log record generated.
-        with self.processor.threadbound(), self.stdout_capture(Logger('Print'),''):
+        #with self.processor.threadbound(), self.stdout_capture(Logger('Print'),''):
 
-            # Call user's initialize method with a timeout.
-            with Timeout(INIT_TIMEOUT, message="Call to initialize timed out"):
-                self.algo.initialize()
+        # Call user's initialize method with a timeout.
+        with Timeout(INIT_TIMEOUT, message="Call to initialize timed out"):
+            self.algo.initialize()
 
-            # Group together events with the same dt field. This depends on the
-            # events already being sorted.
-            for date, snapshot in groupby(stream_in, attrgetter('dt')):
-                # Set the simulation date to be the first event we see.
-                # This should only occur once, at the start of the test.
-                if self.simulation_dt == None:
-                    self.simulation_dt = date
+        # Group together events with the same dt field. This depends on the
+        # events already being sorted.
+        for date, snapshot in groupby(stream_in, attrgetter('dt')):
+            # Set the simulation date to be the first event we see.
+            # This should only occur once, at the start of the test.
+            if self.simulation_dt == None:
+                self.simulation_dt = date
 
-                # Done message has the risk report, so we yield before exiting.
-                if date == 'DONE':
-                    for event in snapshot:
+            # Done message has the risk report, so we yield before exiting.
+            if date == 'DONE':
+                for event in snapshot:
+                    yield event.perf_message
+                raise StopIteration()
+
+            # We're still in the warmup period.  Use the event to
+            # update our universe, but don't yield any perf messages,
+            # and don't send a snapshot to handle_data.
+            elif date < self.algo_start:
+                for event in snapshot:
+                    del event['perf_message']
+                    self.update_universe(event)
+
+            # The algo has taken so long to process events that
+            # its simulated time is later than the event time.
+            # Update the universe and yield any perf messages
+            # 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 != None:
                         yield event.perf_message
-                    raise StopIteration()
+                    # Delete the message before updating so we don't send it
+                    # to the user.
+                    del event['perf_message']
+                    self.update_universe(event)
 
-                # We're still in the warmup period.  Use the event to
-                # update our universe, but don't yield any perf messages,
-                # and don't send a snapshot to handle_data.
-                elif date < self.algo_start:
-                    for event in snapshot:
-                        del event['perf_message']
-                        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 != None:
+                        yield event.perf_message
+                    del event['perf_message']
 
-                # The algo has taken so long to process events that
-                # its simulated time is later than the event time.
-                # Update the universe and yield any perf messages
-                # 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 != None:
-                            yield event.perf_message
-                        # Delete the message before updating so we don't send it
-                        # to the user.
-                        del event['perf_message']
-                        self.update_universe(event)
+                    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 != None:
-                            yield event.perf_message
-                        del event['perf_message']
-
-                        self.update_universe(event)
-
-                    # Send the current state of the universe to the user's algo.
-                    self.simulate_snapshot(date)
+                # Send the current state of the universe to the user's algo.
+                self.simulate_snapshot(date)
 
     def update_universe(self, event):
         """

zipline/gens/transform.py

@@ -175,13 +175,13 @@ class EventWindow(object):
     # Mark this as an abstract base class.
     __metaclass__ = ABCMeta
 
-    def __init__(self, market_aware, days = None, delta = None):
+    def __init__(self, market_aware, days=None, delta=None):
 
         self.market_aware = market_aware
         self.days = days
         self.delta = delta
 
-        self.ticks  = deque()
+        self.ticks = deque()
 
         # Market-aware mode only works with full-day windows.
         if self.market_aware:
@@ -284,9 +284,46 @@ class EventWindow(object):
                 "Events arrived out of order in EventWindow: %s -> %s" % (event, self.ticks[0])
 
 
-class BatchWindow(EventWindow):
-    def __init__(self, func=None, refresh_period=None, days=None, sids=None):
-        super(BatchWindow, self).__init__(True, days=days, delta=None)
+class BatchTransform(EventWindow):
+    """Base class for batch transforms with a trailing window of
+    variable length. As opposed to pure EventWindows that get a stream
+    of events and are bound to a single SID, this class creates stream
+    of pandas DataFrames with each colum representing a sid.
+
+    There are two ways to create a new batch window:
+    (i) Inherit from BatchTransform and overload get_value(data).
+        E.g.:
+        ```
+        class MyBatchTransform(BatchTransform):
+            def get_value(self, data):
+               # compute difference between the means of sid 0 and sid 1
+               return data[0].mean() - data[1].mean()
+        ```
+
+    (ii) Use the batch_transform decorator.
+        E.g.:
+        ```
+        @batch_transform
+        def my_batch_transform(data):
+            return data[0].mean() - data[1].mean()
+
+        ```
+
+    In you algorithm you would then have to instantiate this in the initialize() method:
+    ```
+    self.my_batch_transform = MyBatchTransform()
+    ```
+
+    To then use it, inside of the algorithm handle_data(), call the
+    handle_data() of the BatchTransform and pass it the current event:
+    ```
+    result = self.my_batch_transform(data)
+    ```
+
+    """
+
+    def __init__(self, func=None, refresh_period=None, market_aware=True, delta=None, days=None, sids=None):
+        super(BatchTransform, self).__init__(market_aware, days=days, delta=delta)
         self.func = func
         self.sids = sids
         self.refresh_period = refresh_period
@@ -310,7 +347,8 @@ class BatchWindow(EventWindow):
         # couple of seconds shouldn't matter
         data.dt = max(dts)
 
-        # append data frame to window
+        # append data frame to window. update() will call handle_add() and
+        # handle_remove() appropriately
         self.update(data)
 
         # return newly computed or cached value
@@ -323,15 +361,30 @@ class BatchWindow(EventWindow):
 
         age = event.dt - self.last_refresh
         if age.days >= self.refresh_period:
-            # create Series price object
-            data_sids = {}
-            for sid in self.sids:
-                dts = [tick[sid].dt for tick in self.ticks]
-                prices = [tick[sid].price for tick in self.ticks]
-                data_sids[sid] = pd.Series(prices, index=dts)
+            # Create a pandas.Panel (i.e. 3d DataFrame) from the
+            # events in the current window.
+            #
+            # The resulting panel looks like this:
+            # index : field_name (e.g. price)
+            # major axis/rows : dt
+            # minor axis/colums : sid
+            #
+            # This Panel data structure ultimately gets passed to the
+            # user-overloaded get_value() method.
+            fields = {}
+            for field_name in ['price', 'volume']:
+                # Skip non-existant fields
+                if field_name not in self.ticks[0][self.sids[0]]:
+                    continue
 
-            # concatenate different sids into one df
-            self.data = pd.concat(data_sids, axis=1)
+                values_per_sid = {}
+                for sid in self.sids:
+                    values_per_sid[sid] = pd.Series({tick[sid].dt: tick[sid][field_name] for tick in self.ticks})
+
+                # concatenate different sids into one df
+                fields[field_name] = pd.DataFrame.from_dict(values_per_sid)
+
+            self.data = pd.Panel.from_dict(fields, orient='items')
 
             self.updated = True
             self.last_refresh = event.dt
@@ -347,7 +400,7 @@ class BatchWindow(EventWindow):
 
     def compute(self, *args, **kwargs):
         if self.data is None:
-            return False
+            return None
 
         if self.updated:
             if self.func is not None:
@@ -360,9 +413,14 @@ class BatchWindow(EventWindow):
         return self.cached
 
 
-# decorator for BatchWindow
 def batch_transform(func):
-    def create_transform(*args, **kwargs):
-        return BatchWindow(*args, func=func, **kwargs)
+    """Decorator function to use instead of inheriting from BatchTransform.
+    For an example on how to use this, see the doc string of BatchTransform.
+    """
 
-    return create_transform
+    def create_window(*args, **kwargs):
+        # passes the user defined function to BatchTransform which it
+        # will call instead of self.get_value()
+        return BatchTransform(*args, func=func, **kwargs)
+
+    return create_window

zipline/optimize/algorithms.py

@@ -1,15 +1,5 @@
-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
+from zipline import TradingAlgorithm
 
 logger = Logger('Algo')
 
@@ -64,159 +54,6 @@ class BuySellAlgorithm(object):
         return [self.sid]
 
 
-class TradingAlgorithm(object):
-    """
-    Base class for trading algorithms. Inherit and overload handle_data(data).
-
-    A new algorithm could look like this:
-    ```
-    class MyAlgo(TradingAlgorithm):
-        def initialize(amount):
-            self.amount = amount
-
-        def handle_data(data):
-            sid = self.sids[0]
-            self.order(sid, amount)
-    ```
-    To then run this algorithm:
-
-    >>> my_algo = MyAlgo(100)
-    >>> stats = my_algo.run(data)
-
-    """
-    def __init__(self, sids, *args, **kwargs):
-        """
-        Initialize sids and other state variables.
-
-        Calls user-defined initialize and forwarding *args and **kwargs.
-        """
-        self.sids = sids
-        self.done = False
-        self.order = None
-        self.frame_count = 0
-        self.portfolio = None
-
-        self.registered_transforms = {}
-
-        # call to user-defined initialize method
-        self.initialize(*args, **kwargs)
-
-    def _create_simulator(self, source):
-        """
-        Create trading environment, transforms and SimulatedTrading object.
-
-        Gets called by self.run().
-        """
-        environment = create_trading_environment(start=source.data.index[0], end=source.data.index[-1])
-
-        # Create transforms by wrapping them into StatefulTransforms
-        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)
-
-        # SimulatedTrading is the main class handling data streaming,
-        # application of transforms and calling of the user algo.
-        return SimulatedTrading(
-            [source],
-            transforms,
-            self,
-            environment,
-            FixedSlippage()
-        )
-
-    def run(self, data):
-        """
-        Run the algorithm.
-
-        :Arguments:
-            data : pandas.DataFrame
-               * columns must consist of ints representing the different sids
-               * index must be TimeStamps
-               * array contents should be price
-
-        :Returns:
-            daily_stats : pandas.DataFrame
-              Daily performance metrics such as returns, alpha etc.
-
-        """
-        assert isinstance(data, pd.DataFrame)
-        assert isinstance(data.index, pd.tseries.index.DatetimeIndex)
-
-        source = DataFrameSource(data, sids=self.sids)
-
-        # create transforms and zipline
-        simulated_trading = self._create_simulator(source)
-
-        # loop through simulated_trading, each iteration returns a
-        # perf ndict
-        perfs = list(self.simulated_trading)
-
-        # convert perf ndict to pandas dataframe
-        daily_stats = self._create_daily_stats(perfs)
-
-        return daily_stats
-
-
-    def _create_daily_stats(self, perfs):
-        # create daily and cumulative 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 add_transform(self, transform_class, tag, *args, **kwargs):
-        """Add a single-sid, sequential transform to the model.
-
-        :Arguments:
-            transform_class : class
-                Which transform to use. E.g. mavg.
-            tag : str
-                How to name the transform. Can later be access via:
-                data[sid].tag()
-
-        Extra args and kwargs will be forwarded to the transform
-        instantiation.
-
-        """
-        self.registered_transforms[tag] = {'class': transform_class,
-                                           'args': args,
-                                           'kwargs': kwargs}
-
-    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, *args, **kwargs):
-        pass
-
-    def set_slippage_override(self, slippage_callable):
-        pass
-
-
-
 class BuySellAlgorithmNew(TradingAlgorithm):
     """Algorithm that buys and sells alternatingly. The amount for
     each order can be specified. In addition, an offset that will

zipline/optimize/example.py

@@ -1,25 +1,26 @@
+# WARNING: This file is still work in progress and contains rather
+# random code snippets.
+
 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.gens.cov import  CovEventWindow, cov
-from zipline.optimize.algorithms import TradingAlgorithm
-from datetime import timedelta
+from zipline.gens.cov import  CovTransform, cov
+from zipline.algorithm import TradingAlgorithm
+from zipline.gens.transform import BatchTransform, batch_transform
 
-#from mpi4py_map import map
+@batch_transform
+def cov(data):
+    return data.price.cov()
 
-# Inherits from Algorithm base class
 class DMA(TradingAlgorithm):
     """Dual Moving Average algorithm.
     """
     def initialize(self, amount=100, short_window=20, long_window=40):
-        self.orders = []
         self.amount = amount
-        self.prices = []
         self.events = 0
 
         self.invested = {}
@@ -34,14 +35,12 @@ class DMA(TradingAlgorithm):
                            market_aware=True,
                            days=long_window)
 
-        self.cov = CovEventWindow(sids=self.sids, refresh_period=1, days=5)
-        self.cov2 = cov(sids=self.sids, refresh_period=1, days=5)
+        self.cov = cov(sids=self.sids, refresh_period=1, days=5)
 
     def handle_data(self, data):
         self.events += 1
 
         cov = self.cov.handle_data(data)
-        cov = self.cov2.handle_data(data)
         print cov
 
         for sid in self.sids:

zipline/optimize/factory.py

@@ -10,7 +10,7 @@ import zipline.protocol as zp
 
 from zipline.utils.factory import get_next_trading_dt, create_trading_environment
 from zipline.finance.sources import SpecificEquityTrades
-from zipline.optimize.algorithms import BuySellAlgorithm
+from zipline.optimize.algorithms import BuySellAlgorithmNew
 from zipline.lines import SimulatedTrading
 from zipline.finance.slippage import FixedSlippage
 

zipline/utils/factory.py

@@ -8,13 +8,14 @@ import random
 from os.path import join, abspath, dirname
 from operator import attrgetter
 
+import pandas as pd
+import numpy as np
 from datetime import datetime, timedelta
-from zipline.utils.date_utils import tuple_to_date
-from zipline.utils.protocol_utils import ndict
 
 import zipline.finance.risk as risk
-
-from zipline.gens.tradegens import SpecificEquityTrades
+from zipline.utils.date_utils import tuple_to_date
+from zipline.utils.protocol_utils import ndict
+from zipline.gens.tradegens import SpecificEquityTrades, DataFrameSource
 from zipline.gens.utils import create_trade
 from zipline.finance.trading import TradingEnvironment
 
@@ -90,7 +91,6 @@ def create_trade_history(sid, prices, amounts, interval, trading_calendar):
     current = trading_calendar.first_open
 
     for price, amount in zip(prices, amounts):
-
         trade = create_trade(sid, price, amount, current)
         trades.append(trade)
         current = get_next_trading_dt(current, interval, trading_calendar)
@@ -235,3 +235,12 @@ def create_trade_source(sids, trade_count, trade_time_increment, trading_environ
     #trading_environment.period_end = trade_history[-1].dt
 
     return source
+
+def create_test_df_source():
+    start = pd.datetime(1990, 1, 1, 0, 0, 0, 0, pytz.utc)
+    end = pd.datetime(1990, 1, 10, 0, 0, 0, 0, pytz.utc)
+    index = pd.DatetimeIndex(start=start, end=end)
+    x = np.arange(0, 16).reshape((8, 2))
+    df = pd.DataFrame(x, index=index, columns=[0, 1])
+
+    return DataFrameSource(df), df