From 1576447c764ebcb3bdc7825ac2b22ece164be0f1 Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Fri, 10 Aug 2012 11:08:11 -0400 Subject: [PATCH 01/23] Refactored optimize factory. Enabled optimize unittests. --- tests/test_optimize.py | 8 ++++---- zipline/gens/examples.py | 6 +++--- zipline/optimize/factory.py | 5 ++--- 3 files changed, 9 insertions(+), 10 deletions(-) diff --git a/tests/test_optimize.py b/tests/test_optimize.py index c16454b1..0b132b8f 100644 --- a/tests/test_optimize.py +++ b/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 @@ -38,7 +38,7 @@ class TestUpDown(TestCase): def tearDown(self): teardown_logger(self) - @skip +# @skip @timed(DEFAULT_TIMEOUT) def test_source_and_orders(self): """verify that UpDownSource is having the correct @@ -94,7 +94,7 @@ class TestUpDown(TestCase): "Algorithm did not sell when price was going to increase." ) - @skip +# @skip def test_concavity_of_returns(self): """verify concave relationship between free parameter and returns in certain region around the max. Moreover, @@ -136,7 +136,7 @@ class TestUpDown(TestCase): idx[0] -= 1 idx[1] += 1 - @skip +# @skip def test_optimize(self): """verify that gradient descent (Powell's method) can find the optimal free parameter under which the BuySellAlgorithm produces diff --git a/zipline/gens/examples.py b/zipline/gens/examples.py index f3a0dd0b..21d9cde4 100644 --- a/zipline/gens/examples.py +++ b/zipline/gens/examples.py @@ -9,10 +9,10 @@ from itertools import izip from zipline.utils.factory import create_trading_environment from zipline.test_algorithms import TestAlgorithm -from zipline.gens.composites import SourceBundle, TransformBundle, \ - date_sorted_sources, merged_transforms, sequential_transforms +from zipline.gens.composites import date_sorted_sources, merged_transforms, sequential_transforms from zipline.gens.tradegens import SpecificEquityTrades -from zipline.gens.transform import MovingAverage, Passthrough, StatefulTransform +from zipline.gens.mavg import MovingAverage +from zipline.gens.transform import Passthrough, StatefulTransform from zipline.gens.tradesimulation import TradeSimulationClient as tsc import zipline.protocol as zp diff --git a/zipline/optimize/factory.py b/zipline/optimize/factory.py index cb0de069..3a7fdfff 100644 --- a/zipline/optimize/factory.py +++ b/zipline/optimize/factory.py @@ -4,12 +4,11 @@ Factory functions to prepare useful data for optimize tests. Author: Thomas V. Wiecki (thomas.wiecki@gmail.com), 2012 """ from datetime import timedelta -import pandas as pd 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.gens.tradegens import SpecificEquityTrades from zipline.optimize.algorithms import BuySellAlgorithm from zipline.lines import SimulatedTrading from zipline.finance.trading import SIMULATION_STYLE @@ -66,7 +65,7 @@ def create_updown_trade_source(sid, trade_count, trading_environment, base_price trading_environment.period_end = cur - source = SpecificEquityTrades(events) + source = SpecificEquityTrades(event_list=events) return source From d88b1756726f790dd0c932b74c997a081b88e7fc Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 10 Aug 2012 11:55:30 -0400 Subject: [PATCH 02/23] SimulatedTrading is now a generator. --- zipline/lines.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/zipline/lines.py b/zipline/lines.py index 9bc8b3ac..2a59190f 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -244,10 +244,17 @@ class SimulatedTrading(object): else: return [] + def __iter__(self): + return self + + def next(self): + return self.gen.next() + @staticmethod def create_test_zipline(**config): """ - :param config: A configuration object that is a dict with: + :param config: A configuration object that is a dict with + (all optional): - environment - a \ :py:class:`zipline.finance.trading.TradingEnvironment` @@ -269,7 +276,7 @@ class SimulatedTrading(object): of StatefulTransform objects. """ assert isinstance(config, dict) - sid = config['sid'] + sid = config.get('sid', 133) #-------------------- # Trading Environment From 51f01c0f1f15eeb8b684ffff064e2bf0f975010f Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Fri, 10 Aug 2012 17:23:08 -0400 Subject: [PATCH 03/23] ENH: Implemented new Zipline interface. Implemented Algorithm base class. Implemented example algorithm. Implemented example.py code. --- zipline/gens/composites.py | 23 ++++++++ zipline/gens/tradegens.py | 3 +- zipline/gens/transform.py | 6 +- zipline/lines.py | 104 +++++++++++++++++++++++++++++++++ zipline/optimize/algorithms.py | 57 ++++++++++++++++++ zipline/optimize/example.py | 19 ++++++ zipline/test_algorithms.py | 16 +++++ 7 files changed, 224 insertions(+), 4 deletions(-) create mode 100644 zipline/optimize/example.py diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index b3fa7576..4a74b8f6 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -88,6 +88,29 @@ def sequential_transforms(stream_in, *transforms): dt_aliased = alias_dt(stream_out) return add_done(dt_aliased) +def sequential_transforms_dict(stream_in, transforms): + """ + Apply each transform in transforms sequentially to each event in stream_in. + Each transform application will add a new entry indexed to the transform's + hash string. + """ + + assert isinstance(transforms, dict) + + for tnfm in transforms.itervalues(): + tnfm.forward_all = False + tnfm.update_in_place = False + tnfm.append_value = True + + # Recursively apply all transforms to the stream. + stream_out = reduce(lambda stream, tnfm: tnfm.transform(stream), + transforms, + stream_in) + + dt_aliased = alias_dt(stream_out) + return add_done(dt_aliased) + + def alias_dt(stream_in): """ Alias the dt field to datetime on each message. diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index 2e8f6bea..fad8bc29 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -97,8 +97,9 @@ class SpecificEquityTrades(object): return self.__class__.__name__ + "-" + self.arg_string def create_fresh_generator(self): - if self.event_list: + for event in self.event_list: + event['source_id'] = self.get_hash() unfiltered = (event for event in self.event_list) # Set up iterators for each expected field. diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index 651c337d..099fc1eb 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -93,10 +93,10 @@ class StatefulTransform(object): #TODO: refactor this to avoid unnecessary copying. assert_sort_unframe_protocol(message) - message_copy = deepcopy(message) - + #message_copy = deepcopy(message) + message_copy = message # Same shared pointer issue here as above. - tnfm_value = self.state.update(deepcopy(message_copy)) + tnfm_value = self.state.update(message_copy) # FORWARDER flag means we want to keep all original # values, plus append tnfm_id and tnfm_value. Used for diff --git a/zipline/lines.py b/zipline/lines.py index 2a59190f..c05c5c1c 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -63,6 +63,11 @@ import sys import zmq import os from signal import SIGHUP, SIGINT +import datetime +import pytz +import pandas as pd +import numpy as np + import multiprocessing from setproctitle import setproctitle @@ -70,6 +75,10 @@ from zipline.test_algorithms import TestAlgorithm from zipline.finance.trading import SIMULATION_STYLE from zipline.utils.log_utils import ZeroMQLogHandler, stdout_only_pipe from zipline.utils import factory +from zipline.utils.factory import create_trading_environment +from zipline.gens.tradegens import SpecificEquityTrades +from zipline import ndict +from zipline.protocol import DATASOURCE_TYPE from zipline.test_algorithms import TestAlgorithm @@ -358,3 +367,98 @@ class SimulatedTrading(object): #------------------- return sim + + +def create_sp_source(start_dt=None, end_dt=None): + if start_dt is None: + start_dt = datetime.datetime(2002, 1, 1, tzinfo=pytz.utc) + if end_dt is None: + end_dt = datetime.datetime(2008, 1, 1, tzinfo=pytz.utc) + + sp_events, _ = factory.load_market_data() + sp_transformed = [] + for event in sp_events: + transformed = ndict(event.to_dict()) + if (transformed.dt < start_dt) or (transformed.dt > end_dt): + continue + transformed['sid'] = 0 + transformed['price'] = transformed['returns'] + transformed['type'] = DATASOURCE_TYPE.TRADE + sp_transformed.append(transformed) + + source = SpecificEquityTrades(event_list=sp_transformed) + + return source + +class Zipline(object): + def __init__(self, **kwargs): + algorithm = kwargs.get('algorithm', TestAlgorithm) + source_descrs = kwargs.get('sources', ['S&P']) + if isinstance(source_descrs, str): + source_descrs = [source_descrs] + + sources = [] + for source_descr in source_descrs: + if isinstance(source_descr, str): + if source_descr == 'S&P': + source = create_sp_source() + else: + raise NotImplementedError, "Source with name {source_descr} not known.".format(source_descr=source_descr) + else: + source = source_descr + + sources.append(source) + + environment = kwargs.get('environment', create_trading_environment()) + + try: + transform_descrs = kwargs.get('transforms', algorithm.registered_transforms) + except: + print "Couldn't load any registered_transforms." + transform_descrs = {} + + # Create transforms by wrapping them into StatefulTransforms + transforms = [] + for namestring, trans_descr in transform_descrs.iteritems(): + sf = StatefulTransform( + trans_descr['class'], + *trans_descr['args'], + **trans_descr['kwargs'] + ) + sf.namestring = namestring + + transforms.append(sf) + + results_socket_uri = None + context = None + sim_id = None + style = SIMULATION_STYLE.FIXED_SLIPPAGE + + self.simulated_trading = SimulatedTrading( + sources, + transforms, + algorithm, + environment, + style, + results_socket_uri, + context, + sim_id) + + + def run(self): + # drain simulated_trading + perfs = [perf for perf in self.simulated_trading] + + # 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 diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index c4e3fb9d..119c0f76 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -1,3 +1,6 @@ +from zipline.gens.mavg import MovingAverage +from datetime import datetime, timedelta + 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 +49,57 @@ 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 Algorithm(object): + def set_order(self, order_callable): + self.order = order_callable + + def get_sid_filter(self): + return [self.sid] + + def set_logger(self, logger): + self.logger = logger + + def initialize(self): + 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} + + +# Inherits from Algorithm base class +class DMA(Algorithm): + """Dual Moving Average algorithm. + """ + + def __init__(self, sid, amount, short_window=20, long_window=40): + self.sid = sid + self.amount = amount + self.done = False + self.order = None + self.frame_count = 0 + self.portfolio = None + self.orders = [] + self.market_entered = False + self.prices = [] + self.events = 0 + self.add_transform(MovingAverage, 'short_mavg', ['price'], market_aware=False, delta=timedelta(days=short_window)) + self.add_transform(MovingAverage, 'long_mavg', ['price'], market_aware=False, delta=timedelta(days=long_window)) + + def handle_data(self, data): + self.events += 1 + # access transforms via their user-defined tag + if (data[self.sid].short_mavg > data[self.sid].long_mavg) and not self.market_entered: + self.order(self.sid, 100) + self.market_entered = True + elif (data[self.sid].short_mavg < data[self.sid].long_mavg) and self.market_entered: + self.order(self.sid, -100) + self.market_entered = False diff --git a/zipline/optimize/example.py b/zipline/optimize/example.py new file mode 100644 index 00000000..0119161b --- /dev/null +++ b/zipline/optimize/example.py @@ -0,0 +1,19 @@ +from zipline.lines import Zipline +from zipline.optimize.algorithms import DMA +import pandas as pd +import matplotlib.pyplot as plt +import cProfile + +def run(): + myalgo = DMA(sid=0, amount=100) + zp = Zipline(algorithm=myalgo, sources='S&P') + stats = zp.run() + print stats + return stats + + +#cProfile.run('run()') + +stats = run() +stats.returns.plot() +plt.show() \ No newline at end of file diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index a7881fa8..723c5fb9 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -46,6 +46,22 @@ The algorithm must expose methods: """ +# 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 Algorithm(object): + def set_order(self, order_callable): + self.order = order_callable + + def get_sid_filter(self): + return [self.sid] + + def add_transform(self, transform_class, tag, **kwargs): + if not hasattr(self, 'registered_transforms'): + self.registered_transforms = {} + + self.registered_transforms[tag] = transform_class(**kwargs) + class TestAlgorithm(): """ From 3884b15eb6414e0f5b39a4b7ad2cb441028255b3 Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Sun, 12 Aug 2012 20:24:34 -0400 Subject: [PATCH 04/23] ENH: Changed algorithm class to provide .run() method. Testing of multiple param combos of DMA. --- zipline/finance/performance.py | 15 +++-- zipline/gens/composites.py | 2 +- zipline/gens/tradegens.py | 62 +++++++++++++++++++ zipline/optimize/algorithms.py | 110 +++++++++++++++++++++++---------- zipline/optimize/example.py | 71 ++++++++++++++++++--- 5 files changed, 212 insertions(+), 48 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 9fee208b..a1fbf65c 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -167,6 +167,8 @@ class PerformanceTracker(object): self.last_dict = None self.exceeded_max_loss = False + self.compute_risk_metrics = True + self.results_socket = None self.results_addr = None @@ -297,12 +299,13 @@ 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 - ) + if self.compute_risk_metrics: + 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 + ) # increment the day counter before we move markers forward. self.day_count += 1.0 diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index 4a74b8f6..0a99fc0f 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -16,7 +16,7 @@ def date_sorted_sources(*sources): for source in sources: assert iter(source), "Source %s not iterable" % source - assert source.__class__.__dict__.has_key('get_hash'), "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] diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index fad8bc29..5c22ef93 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -4,6 +4,11 @@ and zipline development """ import random import pytz +from copy import copy + +import pandas as pd +from zipline import ndict +from zipline.protocol import DATASOURCE_TYPE from itertools import chain, cycle, ifilter, izip from datetime import datetime, timedelta @@ -132,6 +137,63 @@ class SpecificEquityTrades(object): return filtered +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. + + Configuration options: + + 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 + """ + + def __init__(self, data, **kwargs): + assert isinstance(data.index, pd.tseries.index.DatetimeIndex) + + self.data = data + # Unpack config dictionary with default values. + self.count = kwargs.get('count', 500) + self.sids = kwargs.get('sids', [1, 2]) + 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)) + + # Default to None for event_list and filter. + self.filter = kwargs.get('filter') + + # Hash_value for downstream sorting. + self.arg_string = hash_args(data, **kwargs) + + self.generator = self.create_fresh_generator() + + def create_fresh_generator(self): + def _generator(df=self.data): + for dt, series in df.iterrows(): + dt = dt.tz_localize('UTC') + if (self.start > dt) 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() + + # !!!!!!! Deprecated for now !!!!!!!!! def RandomEquityTrades(object): diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index 119c0f76..85dac494 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -1,5 +1,16 @@ +import pandas as pd +import numpy as np + from zipline.gens.mavg import MovingAverage from datetime import datetime, timedelta +from zipline.finance.trading import SIMULATION_STYLE +from zipline.utils import factory +from zipline.gens.tradegens import SpecificEquityTrades, DataFrameSource +from zipline.protocol import DATASOURCE_TYPE +from zipline import ndict +from zipline.utils.factory import create_trading_environment +from zipline.gens.transform import StatefulTransform +from zipline.lines import SimulatedTrading class BuySellAlgorithm(object): """Algorithm that buys and sells alternatingly. The amount for @@ -53,12 +64,77 @@ class BuySellAlgorithm(object): # 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 Algorithm(object): +class TradingAlgorithm(object): + def _setup(self, compute_risk_metrics=False): + assert hasattr(self, 'source'), 'source not set.' + assert hasattr(self, 'sids'), "sids not set." + + environment = create_trading_environment() + + # 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) + + results_socket_uri = None + context = None + sim_id = None + style = SIMULATION_STYLE.FIXED_SLIPPAGE + + self.simulated_trading = SimulatedTrading( + [self.source], + transforms, + self, + environment, + style, + results_socket_uri, + context, + sim_id) + + #self.simulated_trading.trading_client.performance_tracker.compute_risk_metrics = compute_risk_metrics + + + 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._setup(compute_risk_metrics=compute_risk_metrics) + + # drain simulated_trading + perfs = [perf for perf in 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.sid] + return self.sids def set_logger(self, logger): self.logger = logger @@ -73,33 +149,3 @@ class Algorithm(object): self.registered_transforms[tag] = {'class': transform_class, 'args': args, 'kwargs': kwargs} - - -# Inherits from Algorithm base class -class DMA(Algorithm): - """Dual Moving Average algorithm. - """ - - def __init__(self, sid, amount, short_window=20, long_window=40): - self.sid = sid - self.amount = amount - self.done = False - self.order = None - self.frame_count = 0 - self.portfolio = None - self.orders = [] - self.market_entered = False - self.prices = [] - self.events = 0 - self.add_transform(MovingAverage, 'short_mavg', ['price'], market_aware=False, delta=timedelta(days=short_window)) - self.add_transform(MovingAverage, 'long_mavg', ['price'], market_aware=False, delta=timedelta(days=long_window)) - - def handle_data(self, data): - self.events += 1 - # access transforms via their user-defined tag - if (data[self.sid].short_mavg > data[self.sid].long_mavg) and not self.market_entered: - self.order(self.sid, 100) - self.market_entered = True - elif (data[self.sid].short_mavg < data[self.sid].long_mavg) and self.market_entered: - self.order(self.sid, -100) - self.market_entered = False diff --git a/zipline/optimize/example.py b/zipline/optimize/example.py index 0119161b..7c0e58b4 100644 --- a/zipline/optimize/example.py +++ b/zipline/optimize/example.py @@ -1,19 +1,72 @@ from zipline.lines import Zipline -from zipline.optimize.algorithms import DMA 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 -def run(): - myalgo = DMA(sid=0, amount=100) - zp = Zipline(algorithm=myalgo, sources='S&P') - stats = zp.run() - print stats +from mpi4py_map import map + +# Inherits from Algorithm base class +class DMA(TradingAlgorithm): + """Dual Moving Average algorithm. + """ + def __init__(self, sid, amount=100, short_window=20, long_window=40): + self.sids = [sid] + self.amount = amount + self.done = False + self.order = None + self.frame_count = 0 + self.portfolio = None + self.orders = [] + self.market_entered = False + self.prices = [] + self.events = 0 + + self.add_transform(MovingAverage, 'short_mavg', ['price'], + market_aware=False, + delta=timedelta(days=int(short_window))) + + self.add_transform(MovingAverage, 'long_mavg', ['price'], + market_aware=False, + delta=timedelta(days=int(long_window))) + + def handle_data(self, data): + self.events += 1 + sid = self.sids[0] + # access transforms via their user-defined tag + if (data[sid].short_mavg > data[sid].long_mavg) and not self.market_entered: + self.order(sid, 100) + self.market_entered = True + elif (data[sid].short_mavg < data[sid].long_mavg) and self.market_entered: + self.order(sid, -100) + self.market_entered = False + + +def run((short_window, long_window)): + data = pd.DataFrame.from_csv('SP500.csv') + myalgo = DMA(sid=0, amount=100, short_window=short_window, long_window=long_window) + stats = myalgo.run(data, compute_risk_metrics=False) + stats['sw'] = short_window + stats['lw'] = long_window return stats +sws, lws = np.mgrid[50:80:5, 100:140:5] -#cProfile.run('run()') +stats_all = map(run, zip(sws.flatten(), lws.flatten())) -stats = run() -stats.returns.plot() +# for sw, lw in zip(sws.flatten(), lws.flatten()): +# stats = run(short_window=sw, long_window=lw) +# stats_all.append(stats) + +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() \ No newline at end of file From 7f1c5100b5858c03e377d964954cef86aa0aeb7d Mon Sep 17 00:00:00 2001 From: fawce Date: Fri, 10 Aug 2012 11:55:30 -0400 Subject: [PATCH 05/23] SimulatedTrading is iterable agian. --- zipline/lines.py | 27 +++++++++++++++------------ 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/zipline/lines.py b/zipline/lines.py index b1463e38..e861706c 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -68,15 +68,16 @@ from setproctitle import setproctitle from zipline.test_algorithms import TestAlgorithm from zipline.finance.trading import SIMULATION_STYLE -from zipline.utils.log_utils import ZeroMQLogHandler +from zipline.utils.log_utils import ZeroMQLogHandler, stdout_only_pipe from zipline.utils import factory -from zipline.gens.composites import ( - date_sorted_sources, - sequential_transforms -) +from zipline.test_algorithms import TestAlgorithm + +from zipline.gens.composites import \ + date_sorted_sources, merged_transforms, sequential_transforms +from zipline.gens.transform import Passthrough, StatefulTransform from zipline.gens.tradesimulation import TradeSimulationClient as tsc -from logbook import Logger +from logbook import Logger, NestedSetup, Processor import zipline.protocol as zp @@ -171,8 +172,6 @@ class SimulatedTrading(object): def close(self): log.info("Closing Simulation: {id}".format(id=self.sim_id)) - if self.results_socket: - self.results_socket.close() if self.proc and self.send_sighup: ppid = os.getppid() if self.success: @@ -245,6 +244,12 @@ class SimulatedTrading(object): else: return [] + def __iter__(self): + return self + + def next(self): + return self.gen.next() + @staticmethod def create_test_zipline(**config): """ @@ -319,13 +324,11 @@ class SimulatedTrading(object): trade_source = config['trade_source'] else: trade_source = factory.create_daily_trade_source( - sid_list, + sids, trade_count, - trading_environment, - concurrent=concurrent_trades + trading_environment ) - #------------------- # Transforms #------------------- From ace0b25d315ccc5513f58bafba56f9fe0d0a446e Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Thu, 23 Aug 2012 14:32:22 -0400 Subject: [PATCH 06/23] Dummy. --- tests/test_optimize.py | 4 ++-- zipline/optimize/algorithms.py | 19 ++++++++----------- 2 files changed, 10 insertions(+), 13 deletions(-) diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 0b132b8f..806f2dc3 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -57,7 +57,7 @@ class TestUpDown(TestCase): 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), @@ -73,7 +73,7 @@ class TestUpDown(TestCase): "Minimum price does not equal expected maximum price." ) - zipline.simulate(blocking=True) + zipline.run(config['trade_source']) algo = config['algorithm'] diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index 85dac494..b901bc4c 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -10,7 +10,7 @@ from zipline.protocol import DATASOURCE_TYPE 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.lines import SimulatedTradingLite class BuySellAlgorithm(object): """Algorithm that buys and sells alternatingly. The amount for @@ -46,6 +46,9 @@ class BuySellAlgorithm(object): def set_portfolio(self, portfolio): self.portfolio = portfolio + def set_logger(self, logger): + self.logger = logger + def handle_data(self, frame): order_size = self.buy_or_sell * (self.amount - (self.offset**2)) self.order(self.sid, order_size) @@ -83,21 +86,15 @@ class TradingAlgorithm(object): transforms.append(sf) - results_socket_uri = None - context = None - sim_id = None + style = SIMULATION_STYLE.FIXED_SLIPPAGE - self.simulated_trading = SimulatedTrading( + self.simulated_trading = SimulatedTradingLite( [self.source], transforms, self, environment, - style, - results_socket_uri, - context, - sim_id) - + style) #self.simulated_trading.trading_client.performance_tracker.compute_risk_metrics = compute_risk_metrics @@ -122,7 +119,7 @@ class TradingAlgorithm(object): self._setup(compute_risk_metrics=compute_risk_metrics) # drain simulated_trading - perfs = [perf for perf in self.simulated_trading] + perfs = list(self.simulated_trading) daily_stats = self._create_daily_stats(perfs) return daily_stats From 7491e1f88ea4f3c3cc585f5352b9363bf3a046e5 Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Thu, 23 Aug 2012 14:33:25 -0400 Subject: [PATCH 07/23] Resolved internal conflict due to stashing. --- tests/test_optimize.py | 1 + zipline/gens/tradegens.py | 4 +++ zipline/optimize/algorithms.py | 61 +++++++++++++++++++++++++++------- zipline/optimize/factory.py | 44 ++++++++++-------------- 4 files changed, 71 insertions(+), 39 deletions(-) diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 806f2dc3..52e27bb8 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -47,6 +47,7 @@ class TestUpDown(TestCase): UpDownSource and BuySellAlgorithm interact correctly." """ + zipline, config = create_predictable_zipline( self.zipline_test_config, offset=0, diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index 30d575aa..5716f374 100644 --- a/zipline/gens/tradegens.py +++ b/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), diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index b901bc4c..dc80ef5a 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -46,9 +46,6 @@ class BuySellAlgorithm(object): def set_portfolio(self, portfolio): self.portfolio = portfolio - def set_logger(self, logger): - self.logger = logger - def handle_data(self, frame): order_size = self.buy_or_sell * (self.amount - (self.offset**2)) self.order(self.sid, order_size) @@ -76,15 +73,16 @@ class TradingAlgorithm(object): # 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 + 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) + transforms.append(sf) style = SIMULATION_STYLE.FIXED_SLIPPAGE @@ -95,6 +93,7 @@ class TradingAlgorithm(object): self, environment, style) + #self.simulated_trading.trading_client.performance_tracker.compute_risk_metrics = compute_risk_metrics @@ -119,7 +118,7 @@ class TradingAlgorithm(object): self._setup(compute_risk_metrics=compute_risk_metrics) # drain simulated_trading - perfs = list(self.simulated_trading) + perfs = [perf for perf in self.simulated_trading] daily_stats = self._create_daily_stats(perfs) return daily_stats @@ -146,3 +145,41 @@ class TradingAlgorithm(object): 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.sid, order_size) + + #sell next time around. + self.buy_or_sell *= -1 + + self.orders.append(order_size) + + self.frame_count += 1 + self.incr += 1 diff --git a/zipline/optimize/factory.py b/zipline/optimize/factory.py index ff91dd35..85ee2c34 100644 --- a/zipline/optimize/factory.py +++ b/zipline/optimize/factory.py @@ -1,21 +1,22 @@ + """ Factory functions to prepare useful data for optimize tests. Author: Thomas V. Wiecki (thomas.wiecki@gmail.com), 2012 """ from datetime import timedelta +import pandas as pd +from copy import copy +from itertools import cycle import zipline.protocol as zp from zipline.utils.factory import get_next_trading_dt, create_trading_environment -from zipline.gens.tradegens import SpecificEquityTrades -from zipline.optimize.algorithms import BuySellAlgorithm +from zipline.gens.tradegens import SpecificEquityTrades, DataFrameSource +from zipline.optimize.algorithms import BuySellAlgorithmNew from zipline.lines import SimulatedTrading from zipline.finance.trading import SIMULATION_STYLE -from copy import copy -from itertools import cycle - def create_updown_trade_source(sid, trade_count, trading_environment, base_price, amplitude): """Create the updown trade source. This source emits events with the price going up and down by the same amount in each @@ -38,8 +39,6 @@ def create_updown_trade_source(sid, trade_count, trading_environment, base_price source : SpecificEquityTrades The trade source emitting up down events. """ - volume = 1000 - events = [] price = base_price-amplitude/2. cur = trading_environment.first_open @@ -47,27 +46,18 @@ def create_updown_trade_source(sid, trade_count, trading_environment, base_price #create iterator to cycle through up and down phases change = cycle([1,-1]) - + prices = [] + dts = [] for i in xrange(trade_count + 2): cur = get_next_trading_dt(cur, one_day, trading_environment) - - event = zp.ndict({ - "type" : zp.DATASOURCE_TYPE.TRADE, - "sid" : sid, - "price" : price, - "volume" : volume, - "dt" : cur, - }) - - events.append(event) + dts.append(cur) + prices.append(price) price += change.next()*amplitude - trading_environment.period_end = cur + df = pd.DataFrame(index=dts, data=prices, columns=[0]) - source = SpecificEquityTrades(event_list=events) - - return source + return df def create_predictable_zipline(config, offset=0, simulate=True): @@ -121,7 +111,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 @@ -130,9 +120,9 @@ def create_predictable_zipline(config, offset=0, simulate=True): config['simulation_style'] = SIMULATION_STYLE.FIXED_SLIPPAGE config['devel'] = True - zipline = SimulatedTrading.create_test_zipline(**config) - if simulate: - zipline.drain_zipline(blocking=True) + algorithm.run() + + return algorithm, config + - return zipline, config From 0b589ee39c372ab1e0a62b1381ef1a1b5b73f12d Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Thu, 23 Aug 2012 16:03:02 -0400 Subject: [PATCH 08/23] Fixes to the updown algorithm. First unittest passes. --- tests/test_optimize.py | 8 +++----- zipline/gens/tradegens.py | 5 ++--- zipline/lines.py | 13 ++++++++++--- zipline/optimize/algorithms.py | 10 +++++++--- 4 files changed, 22 insertions(+), 14 deletions(-) diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 52e27bb8..83dca6a9 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -27,7 +27,7 @@ class TestUpDown(TestCase): def setUp(self): self.zipline_test_config = { 'allocator' : allocator, - 'sid' : 133, + 'sid' : [0], 'trade_count' : 5, 'amplitude' : 30, 'base_price' : 50 @@ -48,7 +48,7 @@ class TestUpDown(TestCase): """ - zipline, config = create_predictable_zipline( + algo, config = create_predictable_zipline( self.zipline_test_config, offset=0, simulate=False @@ -74,9 +74,7 @@ class TestUpDown(TestCase): "Minimum price does not equal expected maximum price." ) - zipline.run(config['trade_source']) - - algo = config['algorithm'] + algo.run(config['trade_source']) orders = np.asarray(algo.orders) max_order_idx = np.where(orders==orders.max())[0] diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index 5716f374..09685cba 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -182,7 +182,7 @@ class DataFrameSource(SpecificEquityTrades): self.data = data # Unpack config dictionary with default values. self.count = kwargs.get('count', 500) - self.sids = kwargs.get('sids', [1, 2]) + 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)) @@ -198,8 +198,7 @@ class DataFrameSource(SpecificEquityTrades): def create_fresh_generator(self): def _generator(df=self.data): for dt, series in df.iterrows(): - dt = dt.tz_localize('UTC') - if (self.start > dt) or (dt < self.end): + if (dt < self.start) or (dt > self.end): continue event = {'dt': dt, 'source_id': self.get_hash(), diff --git a/zipline/lines.py b/zipline/lines.py index e4e4af8c..9f2a8153 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -179,7 +179,7 @@ class SimulatedTrading(object): else: log.warning("Sending SIGINT") os.kill(ppid, SIGINT) - + def handle_exception(self, exc): if isinstance(exc, CancelSignal): # signal from monitor of an orderly shutdown, @@ -206,7 +206,7 @@ class SimulatedTrading(object): exc_type.__name__, exc_value.message ) - + self.results_socket.send(msg) except: log.exception("Exception while reporting simulation exception.") @@ -380,6 +380,13 @@ class SimulatedTradingLite(object): self.with_tnfms = sequential_transforms(self.date_sorted, *self.transforms) self.trading_client = tsc(algorithm, environment, style) self.gen = self.trading_client.simulate(self.with_tnfms) - + + def get_results(self): return self.gen + + def __iter__(self): + return self + + def next(self): + return self.gen.next() diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index dc80ef5a..727e43db 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -10,8 +10,12 @@ from zipline.protocol import DATASOURCE_TYPE from zipline import ndict from zipline.utils.factory import create_trading_environment from zipline.gens.transform import StatefulTransform -from zipline.lines import SimulatedTradingLite +from zipline.lines import SimulatedTradingLite, SimulatedTrading +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 @@ -118,7 +122,7 @@ class TradingAlgorithm(object): self._setup(compute_risk_metrics=compute_risk_metrics) # drain simulated_trading - perfs = [perf for perf in self.simulated_trading] + perfs = list(self.simulated_trading) daily_stats = self._create_daily_stats(perfs) return daily_stats @@ -174,7 +178,7 @@ class BuySellAlgorithmNew(TradingAlgorithm): def handle_data(self, data): order_size = self.buy_or_sell * (self.amount - (self.offset**2)) - self.order(self.sid, order_size) + self.order(self.sids[0], order_size) #sell next time around. self.buy_or_sell *= -1 From 3f801ef387a0ebeb11ffdcf8ed375899acb1405f Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Thu, 23 Aug 2012 17:38:48 -0400 Subject: [PATCH 09/23] Debugging. --- tests/test_optimize.py | 18 ++++++++++++------ zipline/gens/tradesimulation.py | 10 +++++----- zipline/optimize/algorithms.py | 12 +++++++++--- zipline/optimize/factory.py | 8 ++------ zipline/utils/factory.py | 9 ++++++--- 5 files changed, 34 insertions(+), 23 deletions(-) diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 83dca6a9..0b237981 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -50,8 +50,7 @@ class TestUpDown(TestCase): algo, config = create_predictable_zipline( self.zipline_test_config, - offset=0, - simulate=False + offset=0 ) #extract arguments @@ -74,7 +73,9 @@ class TestUpDown(TestCase): "Minimum price does not equal expected maximum price." ) - algo.run(config['trade_source']) + stats = algo.run(config['trade_source']) + + self.assertTrue(len(stats) != 0) orders = np.asarray(algo.orders) max_order_idx = np.where(orders==orders.max())[0] @@ -93,6 +94,8 @@ class TestUpDown(TestCase): "Algorithm did not sell when price was going to increase." ) + from nose.tools import set_trace; set_trace() + # @skip def test_concavity_of_returns(self): """verify concave relationship between free parameter and @@ -110,12 +113,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." diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 98aa727f..8ad54053 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -109,12 +109,12 @@ class TradeSimulationClient(object): yield message class AlgorithmSimulator(object): - + def __init__(self, order_book, algo, algo_start): - + # ========== # Algo Setup # ========== @@ -205,7 +205,7 @@ class AlgorithmSimulator(object): # simulator so that it can fill the placed order when it # receives its next message. self.order_book.place_order(order) - + def transform(self, stream_in): """ Main generator work loop. @@ -266,7 +266,7 @@ class AlgorithmSimulator(object): del event['perf_message'] self.update_universe(event) - + # Send the current state of the universe to the user's algo. self.simulate_snapshot(date) @@ -289,7 +289,7 @@ class AlgorithmSimulator(object): # Needs to be set so that we inject the proper date into algo # log/print lines. self.snapshot_dt = date - + start_tic = datetime.now() with self.heartbeat_monitor: self.algo.handle_data(self.universe) diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index 727e43db..30feaf59 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -73,7 +73,7 @@ class TradingAlgorithm(object): assert hasattr(self, 'source'), 'source not set.' assert hasattr(self, 'sids'), "sids not set." - environment = create_trading_environment() + environment = create_trading_environment(start=self.data.index[0], end=self.data.index[-1]) # Create transforms by wrapping them into StatefulTransforms transforms = [] @@ -118,11 +118,16 @@ class TradingAlgorithm(object): def run(self, data, compute_risk_metrics=False): self.source = DataFrameSource(data, sids=self.sids) - + self.data = data self._setup(compute_risk_metrics=compute_risk_metrics) # drain simulated_trading - perfs = list(self.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 @@ -187,3 +192,4 @@ class BuySellAlgorithmNew(TradingAlgorithm): self.frame_count += 1 self.incr += 1 + from nose.tools import set_trace; set_trace() diff --git a/zipline/optimize/factory.py b/zipline/optimize/factory.py index 85ee2c34..319ee0dd 100644 --- a/zipline/optimize/factory.py +++ b/zipline/optimize/factory.py @@ -42,7 +42,7 @@ def create_updown_trade_source(sid, trade_count, trading_environment, base_price price = base_price-amplitude/2. cur = trading_environment.first_open - one_day = timedelta(minutes = 1)#days = 1) + one_day = timedelta(days = 1) #create iterator to cycle through up and down phases change = cycle([1,-1]) @@ -60,7 +60,7 @@ def create_updown_trade_source(sid, trade_count, trading_environment, base_price return df -def create_predictable_zipline(config, offset=0, simulate=True): +def create_predictable_zipline(config, offset=0): """Create a test zipline object as specified by config. The zipline will use the UpDown tradesource which is perfectly predictable. @@ -118,10 +118,6 @@ def create_predictable_zipline(config, offset=0, simulate=True): config['trade_source'] = source config['environment'] = trading_environment config['simulation_style'] = SIMULATION_STYLE.FIXED_SLIPPAGE - config['devel'] = True - - if simulate: - algorithm.run() return algorithm, config diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py index cf2168fb..8b2565c4 100644 --- a/zipline/utils/factory.py +++ b/zipline/utils/factory.py @@ -55,12 +55,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, From 5fcaed3644f4fe48e2681f52aa5602b35ba532c7 Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Wed, 29 Aug 2012 09:02:48 -0400 Subject: [PATCH 10/23] Added logging statements. --- zipline/gens/tradesimulation.py | 2 ++ zipline/optimize/algorithms.py | 5 +++-- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 8ad54053..2f60c7e0 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -190,6 +190,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( diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index 30feaf59..cc438fd1 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -14,8 +14,8 @@ from zipline.lines import SimulatedTradingLite, SimulatedTrading 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 @@ -184,6 +184,7 @@ class BuySellAlgorithmNew(TradingAlgorithm): 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 @@ -192,4 +193,4 @@ class BuySellAlgorithmNew(TradingAlgorithm): self.frame_count += 1 self.incr += 1 - from nose.tools import set_trace; set_trace() + From b74721421e5dde6dbd1f4608ed991614fe522a3e Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Fri, 7 Sep 2012 16:55:01 -0400 Subject: [PATCH 11/23] WIP: First naive implementation of BatchWindow. --- zipline/gens/transform.py | 91 +++++++++++++++++++++++++++++++++------ 1 file changed, 78 insertions(+), 13 deletions(-) diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index 03eb41ea..bc8b716a 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -9,6 +9,8 @@ from datetime import datetime from collections import deque from abc import ABCMeta, abstractmethod +import pandas as pd + from zipline import ndict from zipline.utils.tradingcalendar import non_trading_days from zipline.gens.utils import assert_sort_unframe_protocol, hash_args @@ -36,7 +38,7 @@ class TransformMeta(type): still recover an instance of a "raw" Foo by introspecting the resulting StatefulTransform's 'state' field. """ - + def __call__(cls, *args, **kwargs): return StatefulTransform(cls, *args, **kwargs) @@ -53,19 +55,19 @@ class StatefulTransform(object): def __init__(self, tnfm_class, *args, **kwargs): assert isinstance(tnfm_class, (types.ObjectType, types.ClassType)), \ "Stateful transform requires a class." - assert tnfm_class.__dict__.has_key('update'), \ + assert hasattr(tnfm_class, 'update'), \ "Stateful transform requires the class to have an update method" # Flag set inside the Passthrough transform class to signify special # behavior if we are being fed to merged_transforms. - self.passthrough = tnfm_class.__dict__.get('PASSTHROUGH', False) - + self.passthrough = hasattr(tnfm_class, 'PASSTHROUGH') + # Flags specifying how to append the calculated value. # Merged is the default for ease of testing, but we use sequential # in production. self.sequential = False self.merged = True - + # Create an instance of our transform class. if isinstance(tnfm_class, TransformMeta): # Classes derived TransformMeta have their __call__ @@ -104,12 +106,12 @@ class StatefulTransform(object): continue assert_sort_unframe_protocol(message) - + # This flag is set by by merged_transforms to ensure # isolation of messages. if self.merged: message = deepcopy(message) - + tnfm_value = self.state.update(message) # PASSTHROUGH flag means we want to keep all original @@ -133,7 +135,7 @@ class StatefulTransform(object): out_message.tnfm_value = tnfm_value out_message.dt = message.dt yield out_message - + # Sequential flag should be used to add a single new # key-value pair to the event. The new key is this # transform's namestring, and its value is the value @@ -147,9 +149,11 @@ class StatefulTransform(object): out_message = message out_message[self.namestring] = tnfm_value yield out_message - + log.info('Finished StatefulTransform [%s]' % self.get_hash()) + + class EventWindow(object): """ Abstract base class for transform classes that calculate iterative @@ -218,7 +222,7 @@ class EventWindow(object): # Subclasses should override handle_add to define behavior for # adding new ticks. self.handle_add(event) - + if self.market_aware: self.add_new_holidays(event.dt) @@ -229,14 +233,14 @@ class EventWindow(object): # | | # V V while self.drop_condition(self.ticks[0].dt, self.ticks[-1].dt): - + # popleft removes and returns the oldest tick in self.ticks popped = self.ticks.popleft() # Subclasses should override handle_remove to define # behavior for removing ticks. self.handle_remove(popped) - + def add_new_holidays(self, newest): # Add to our tracked window any untracked holidays that are # older than our newest event. (newest should always be @@ -256,7 +260,7 @@ class EventWindow(object): calendar_dates_between = (newest.date() - oldest.date()).days holidays_between = len(self.cur_holidays) trading_days_between = calendar_dates_between - holidays_between - + # "Put back" a day if oldest is earlier in its day than newest, # reflecting the fact that we haven't yet completed the last # day in the window. @@ -277,3 +281,64 @@ class EventWindow(object): # Something is wrong if new event is older than previous. assert event.dt >= self.ticks[-1].dt, \ "Events arrived out of order in EventWindow: %s -> %s" % (event, self.ticks[0]) + + +class BatchWindow(EventWindow): + def __init__(self, func, refresh_period=None, wind_length=None, sids=None): + super(BatchWindow, self).__init__(True, days=wind_length, delta=None) + self.func = func + self.sids = sids + self.refresh_period = refresh_period + self.wind_length = wind_length + + self.last_calc = False + self.full = False + self.last_refresh = None + + self.updated = False + + def handle_data(self, data): + """ + New method to handle a data frame as sent to the algorithm's handle_data + method. + """ + dts = [data[sid].datetime for sid in self.sids] + prices = [data[sid].price for sid in self.sids] + volumes = [data[sid].volume for sid in self.sids] + + price_df = pd.DataFrame(prices, columns=self.sids, index=dts) + volume_df = pd.DataFrame(volumes, columns=self.sids, index=dts) + + event = ndict({ + 'dt' : max(dts), + 'prices': price_df, + 'volumes': volume_df, + }) + + self.update(event) + + def handle_add(self, event): + if not self.last_calc: + self.last_calc = event.dt + return + + age = event.dt - self.last_refresh + if age.days >= self.refresh_period: + self.prices = pd.concat(self.ticks.prices) + self.volumes = pd.concat(self.ticks.volumes) + + self.updated = True + else: + self.updated = False + + self.last_refresh = event.dt + + def handle_remove(self, event): + # since an event is expiring, we know the window is full + self.full = True + + def __call__(self, *args, **kwargs): + if self.updated: + self.cached = self.func(self.prices, self.volumes, *args, **kwargs) + + return self.cached From 784fc72569778ae219e0bae99757f4e89477e94e Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Wed, 12 Sep 2012 22:06:19 -0400 Subject: [PATCH 12/23] WIP: Changed Batch event window to inheritance for now. Added example covariance. --- zipline/gens/cov.py | 5 +++++ zipline/gens/transform.py | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) create mode 100644 zipline/gens/cov.py diff --git a/zipline/gens/cov.py b/zipline/gens/cov.py new file mode 100644 index 00000000..eefb1c96 --- /dev/null +++ b/zipline/gens/cov.py @@ -0,0 +1,5 @@ +from zipline.gens.transform import EventWindowBatch + +class CovEventWindow(EventWindowBatch): + def get_value(self, prices, volumes): + return prices.cov() \ No newline at end of file diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index bc8b716a..87371ebb 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -339,6 +339,6 @@ class BatchWindow(EventWindow): def __call__(self, *args, **kwargs): if self.updated: - self.cached = self.func(self.prices, self.volumes, *args, **kwargs) + self.cached = self.get_value(self.prices, self.volumes, *args, **kwargs) return self.cached From ea5cbaba56b2b9fd620e954d56df91b1d8b8bcea Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Mon, 17 Sep 2012 10:30:22 -0400 Subject: [PATCH 13/23] Commented out handle_data(). --- zipline/gens/transform.py | 35 ++++++++++++++++++----------------- 1 file changed, 18 insertions(+), 17 deletions(-) diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index 87371ebb..87dad979 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -286,7 +286,6 @@ class EventWindow(object): class BatchWindow(EventWindow): def __init__(self, func, refresh_period=None, wind_length=None, sids=None): super(BatchWindow, self).__init__(True, days=wind_length, delta=None) - self.func = func self.sids = sids self.refresh_period = refresh_period self.wind_length = wind_length @@ -297,27 +296,29 @@ class BatchWindow(EventWindow): self.updated = False - def handle_data(self, data): - """ - New method to handle a data frame as sent to the algorithm's handle_data - method. - """ - dts = [data[sid].datetime for sid in self.sids] - prices = [data[sid].price for sid in self.sids] - volumes = [data[sid].volume for sid in self.sids] + # def handle_data(self, data): + # """ + # New method to handle a data frame as sent to the algorithm's handle_data + # method. + # """ + # dts = [data[sid].datetime for sid in self.sids] + # prices = [data[sid].price for sid in self.sids] + # volumes = [data[sid].volume for sid in self.sids] - price_df = pd.DataFrame(prices, columns=self.sids, index=dts) - volume_df = pd.DataFrame(volumes, columns=self.sids, index=dts) + # price_df = pd.DataFrame(prices, columns=self.sids, index=dts) + # volume_df = pd.DataFrame(volumes, columns=self.sids, index=dts) - event = ndict({ - 'dt' : max(dts), - 'prices': price_df, - 'volumes': volume_df, - }) + # event = ndict({ + # 'dt' : max(dts), + # 'prices': price_df, + # 'volumes': volume_df, + # }) - self.update(event) + # self.update(event) def handle_add(self, event): + import pdb; pdb.set_trace() + if not self.last_calc: self.last_calc = event.dt return From 35a7da6ee7e3a5efe4255aa670542ea30983f605 Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Mon, 17 Sep 2012 17:13:20 -0400 Subject: [PATCH 14/23] Cleaned up depracated merged_transforms. --- zipline/gens/composites.py | 46 ++------------------------------------ 1 file changed, 2 insertions(+), 44 deletions(-) diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index ef97bfd1..7effcf19 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -1,10 +1,7 @@ -from itertools import tee, chain +from itertools import chain from zipline.gens.utils import roundrobin, done_message from zipline.gens.sort import date_sort -from zipline.gens.merge import merge -from zipline.gens.transform import StatefulTransform - def date_sorted_sources(*sources): """ @@ -29,46 +26,6 @@ def date_sorted_sources(*sources): return date_sort(stream_in, names) - -def merged_transforms(sorted_stream, *transforms): - """ - A generator that takes the expected output of a date_sort, pipes - it through a given set of transforms, and runs the results - through a merge to output a unified stream. tnfms should be a - list of pointers to generator functions. tnfm_args should be a - list of tuples, representing the arguments to be passed to each - transform. tnfm_kwargs should be a list of dictionaries - representing keyword arguments to each transform. - """ - for transform in transforms: - assert isinstance(transform, StatefulTransform) - transform.merged = True - transform.sequential = False - - # Generate expected hashes for each transform - namestrings = [tnfm.get_hash() for tnfm in transforms] - - # Create a copy of the stream for each transform. - split = tee(sorted_stream, len(transforms)) - - # Package a stream copy with each StatefulTransform instance. - bundles = zip(transforms, split) - - # Convert the copies into transform streams. - tnfm_gens = [tnfm.transform(stream) for tnfm, stream in bundles] - - # Roundrobin the outputs of our transforms to create a single flat - # stream. - to_merge = roundrobin(tnfm_gens, namestrings) - - # Pipe the stream into merge. - merged = merge(to_merge, namestrings) - - dt_aliased = alias_dt(merged) - # Return the merged events. - return add_done(dt_aliased) - - def sequential_transforms(stream_in, *transforms): """ Apply each transform in transforms sequentially to each event in stream_in. @@ -87,6 +44,7 @@ def sequential_transforms(stream_in, *transforms): transforms, stream_in) + dt_aliased = alias_dt(stream_out) return add_done(dt_aliased) From 280f122353d5bef8164d44fd0ed3b26342bf5624 Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Mon, 17 Sep 2012 18:35:21 -0400 Subject: [PATCH 15/23] WIP: Lot of refactoring and bugfixing. --- zipline/gens/cov.py | 12 ++- zipline/gens/tradegens.py | 12 +-- zipline/gens/tradesimulation.py | 4 +- zipline/gens/transform.py | 87 +++++++++++------- zipline/optimize/algorithms.py | 152 +++++++++++++++++++++++--------- zipline/optimize/example.py | 28 +++--- zipline/test_algorithms.py | 2 +- 7 files changed, 199 insertions(+), 98 deletions(-) diff --git a/zipline/gens/cov.py b/zipline/gens/cov.py index eefb1c96..85985795 100644 --- a/zipline/gens/cov.py +++ b/zipline/gens/cov.py @@ -1,5 +1,9 @@ -from zipline.gens.transform import EventWindowBatch +from zipline.gens.transform import BatchWindow, batch_transform -class CovEventWindow(EventWindowBatch): - def get_value(self, prices, volumes): - return prices.cov() \ No newline at end of file +class CovEventWindow(BatchWindow): + def get_value(self, data): + return data.cov() + +@batch_transform +def cov(data): + return data.cov() diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index e7c4e375..652ea135 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -181,11 +181,11 @@ class DataFrameSource(SpecificEquityTrades): 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)) + self.count = kwargs.get('count', len(data)) + self.sids = kwargs.get('sids', data.columns) + self.start = kwargs.get('start', data.index[0]) + self.end = kwargs.get('end', data.index[-1]) + self.delta = kwargs.get('delta', data.index[1]-data.index[0]) # Default to None for event_list and filter. self.filter = kwargs.get('filter') @@ -207,7 +207,7 @@ class DataFrameSource(SpecificEquityTrades): for sid, price in series.iterkv(): event = copy(event) - event['sid'] = 0 + event['sid'] = sid event['price'] = price yield ndict(event) diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index cca56e18..61527c1a 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -2,6 +2,7 @@ from logbook import Logger, Processor from datetime import datetime from itertools import groupby +from operator import attrgetter from zipline import ndict from zipline.utils.timeout import Heartbeat, Timeout @@ -226,8 +227,7 @@ class AlgorithmSimulator(object): # Group together events with the same dt field. This depends on the # events already being sorted. - for date, snapshot in groupby(stream_in, lambda e: e.dt): - + 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: diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index 87dad979..e4171ebc 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -217,7 +217,7 @@ class EventWindow(object): self.assert_well_formed(event) # Add new event and increment totals. - self.ticks.append(event) + self.ticks.append(deepcopy(event)) # Subclasses should override handle_add to define behavior for # adding new ticks. @@ -266,6 +266,7 @@ class EventWindow(object): # day in the window. if oldest.time() > newest.time(): trading_days_between -= 1 + return trading_days_between >= self.days def out_of_delta(self, oldest, newest): @@ -284,62 +285,84 @@ class EventWindow(object): class BatchWindow(EventWindow): - def __init__(self, func, refresh_period=None, wind_length=None, sids=None): - super(BatchWindow, self).__init__(True, days=wind_length, delta=None) + def __init__(self, func=None, refresh_period=None, days=None, sids=None): + super(BatchWindow, self).__init__(True, days=days, delta=None) + self.func = func self.sids = sids self.refresh_period = refresh_period - self.wind_length = wind_length + self.days = days - self.last_calc = False self.full = False self.last_refresh = None self.updated = False + self.data = None - # def handle_data(self, data): - # """ - # New method to handle a data frame as sent to the algorithm's handle_data - # method. - # """ - # dts = [data[sid].datetime for sid in self.sids] - # prices = [data[sid].price for sid in self.sids] - # volumes = [data[sid].volume for sid in self.sids] + def handle_data(self, data): + """ + New method to handle a data frame as sent to the algorithm's handle_data + method. + """ + # extract dates + dts = [data[sid].datetime for sid in self.sids] + # we have to provide the event with a dt. This is only for + # checking if the event is outside the window or not so a + # couple of seconds shouldn't matter + data.dt = max(dts) - # price_df = pd.DataFrame(prices, columns=self.sids, index=dts) - # volume_df = pd.DataFrame(volumes, columns=self.sids, index=dts) + # append data frame to window + self.update(data) - # event = ndict({ - # 'dt' : max(dts), - # 'prices': price_df, - # 'volumes': volume_df, - # }) - - # self.update(event) + # return newly computed or cached value + return self.compute() def handle_add(self, event): - import pdb; pdb.set_trace() - - if not self.last_calc: - self.last_calc = event.dt + if not self.last_refresh: + self.last_refresh = event.dt return age = event.dt - self.last_refresh if age.days >= self.refresh_period: - self.prices = pd.concat(self.ticks.prices) - self.volumes = pd.concat(self.ticks.volumes) + # 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) + + # concatenate different sids into one df + self.data = pd.concat(data_sids, axis=1) self.updated = True + self.last_refresh = event.dt else: self.updated = False - self.last_refresh = event.dt - def handle_remove(self, event): # since an event is expiring, we know the window is full self.full = True - def __call__(self, *args, **kwargs): + def get_value(self, *args, **kwargs): + raise NotImplementedError("Either overwrite get_value or provide a func argument.") + + def compute(self, *args, **kwargs): + if self.data is None: + return False + if self.updated: - self.cached = self.get_value(self.prices, self.volumes, *args, **kwargs) + if self.func is not None: + # user supplied function + self.cached = self.func(self.data, *args, **kwargs) + else: + # assume inheritance + self.cached = self.get_value(self.data, *args, **kwargs) return self.cached + + +# decorator for BatchWindow +def batch_transform(func): + def create_transform(*args, **kwargs): + return BatchWindow(*args, func=func, **kwargs) + + return create_transform diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index 0a860bd2..218dc989 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -48,6 +48,7 @@ class BuySellAlgorithm(object): self.portfolio = portfolio def handle_data(self, frame): + print frame.sid order_size = self.buy_or_sell * (self.amount - (self.offset**2)) self.order(self.sid, order_size) @@ -62,40 +63,114 @@ 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]) +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(data). + """ + environment = create_trading_environment(start=source.data.index[0], end=source.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 + 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) + transforms.append(sf) - - self.simulated_trading = SimulatedTrading( - [self.source], + # 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.Timeseries) + + 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 = [] + for perf in simulated_trading: + #from nose.tools import set_trace; set_trace() + perfs.append(perf) + + #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 stats dataframe + # create daily and cumulative stats dataframe daily_perfs = [] cum_perfs = [] for perf in perfs: @@ -109,21 +184,23 @@ class TradingAlgorithm(object): return daily_stats - def run(self, data, compute_risk_metrics=False): - self.source = DataFrameSource(data, sids=self.sids) - self.data = data - self._setup() + def add_transform(self, transform_class, tag, *args, **kwargs): + """Add a single-sid, sequential transform to the model. - # drain simulated_trading - perfs = [] - for perf in self.simulated_trading: - #from nose.tools import set_trace; set_trace() - perfs.append(perf) + :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() - #perfs = list(self.simulated_trading) + Extra args and kwargs will be forwarded to the transform + instantiation. - daily_stats = self._create_daily_stats(perfs) - return daily_stats + """ + self.registered_transforms[tag] = {'class': transform_class, + 'args': args, + 'kwargs': kwargs} def set_portfolio(self, portfolio): self.portfolio = portfolio @@ -137,19 +214,12 @@ class TradingAlgorithm(object): def set_logger(self, logger): self.logger = logger - def initialize(self): + def initialize(self, *args, **kwargs): 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): diff --git a/zipline/optimize/example.py b/zipline/optimize/example.py index feca6306..c25dafc9 100644 --- a/zipline/optimize/example.py +++ b/zipline/optimize/example.py @@ -6,6 +6,7 @@ import numpy as np 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 @@ -15,15 +16,9 @@ from datetime import timedelta 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 + def initialize(self, amount=100, short_window=20, long_window=40): self.orders = [] - + self.amount = amount self.prices = [] self.events = 0 @@ -33,15 +28,22 @@ class DMA(TradingAlgorithm): self.add_transform(MovingAverage, 'short_mavg', ['price'], market_aware=True, - days=short_window) #timedelta(days=int(short_window))) + days=short_window) self.add_transform(MovingAverage, 'long_mavg', ['price'], market_aware=True, - days=long_window) #timedelta(days=int(long_window))) + 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) 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: # access transforms via their user-defined tag if (data[sid].short_mavg['price'] > data[sid].long_mavg['price']) and not self.invested[sid]: @@ -86,8 +88,8 @@ def load_close_px(indexes=None, stocks=None): 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) + data = pd.DataFrame.from_csv('aapl.csv') #load_close_px() + myalgo = DMA([0, 1], amount=100, short_window=short_window, long_window=long_window) stats = myalgo.run(data) stats['sw'] = short_window stats['lw'] = long_window @@ -153,3 +155,5 @@ def plot_returns(port_returns, bmk_returns): cum_bmk.plot(label='Benchmark') plt.title('Portfolio performance') plt.legend(loc='best') + +print run((10, 20)) \ No newline at end of file diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index 3b04b243..055c8c32 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -98,7 +98,7 @@ class TestAlgorithm(): def set_slippage_override(self, slippage_callable): pass - # + class HeavyBuyAlgorithm(): """ This algorithm will send a specified number of orders, to allow unit tests From 7242a4ee86a6fc818d4c7bd56823031a3e81285e Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Tue, 18 Sep 2012 13:56:58 -0400 Subject: [PATCH 16/23] BUG: asserting for correct index type. --- zipline/optimize/algorithms.py | 11 +++-------- zipline/optimize/example.py | 3 ++- 2 files changed, 5 insertions(+), 9 deletions(-) diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index 218dc989..098acad8 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -105,7 +105,7 @@ class TradingAlgorithm(object): """ Create trading environment, transforms and SimulatedTrading object. - Gets called by self.run(data). + Gets called by self.run(). """ environment = create_trading_environment(start=source.data.index[0], end=source.data.index[-1]) @@ -147,7 +147,7 @@ class TradingAlgorithm(object): """ assert isinstance(data, pd.DataFrame) - assert isinstance(data.index, pd.Timeseries) + assert isinstance(data.index, pd.tseries.index.DatetimeIndex) source = DataFrameSource(data, sids=self.sids) @@ -156,12 +156,7 @@ class TradingAlgorithm(object): # loop through simulated_trading, each iteration returns a # perf ndict - perfs = [] - for perf in simulated_trading: - #from nose.tools import set_trace; set_trace() - perfs.append(perf) - - #perfs = list(self.simulated_trading) + perfs = list(self.simulated_trading) # convert perf ndict to pandas dataframe daily_stats = self._create_daily_stats(perfs) diff --git a/zipline/optimize/example.py b/zipline/optimize/example.py index c25dafc9..02af8d7e 100644 --- a/zipline/optimize/example.py +++ b/zipline/optimize/example.py @@ -88,7 +88,8 @@ def load_close_px(indexes=None, stocks=None): def run((short_window, long_window)): #data = pd.DataFrame.from_csv('SP500.csv') - data = pd.DataFrame.from_csv('aapl.csv') #load_close_px() + #data = pd.DataFrame.from_csv('aapl.csv') #load_close_px() + data = load_close_px() myalgo = DMA([0, 1], amount=100, short_window=short_window, long_window=long_window) stats = myalgo.run(data) stats['sw'] = short_window From 729a4d9058079d3496362de5ed813eabb3d94b5f Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Wed, 19 Sep 2012 17:49:39 -0400 Subject: [PATCH 17/23] Large refactoring and documentation of new algorithm base class and batch transform. --- tests/test_optimize.py | 6 -- tests/test_transforms.py | 110 ++++++++++++++++----- zipline/__init__.py | 4 +- zipline/algorithm.py | 163 +++++++++++++++++++++++++++++++ zipline/gens/cov.py | 9 -- zipline/gens/tradegens.py | 1 - zipline/gens/tradesimulation.py | 96 +++++++++---------- zipline/gens/transform.py | 96 +++++++++++++++---- zipline/optimize/algorithms.py | 165 +------------------------------- zipline/optimize/example.py | 21 ++-- zipline/optimize/factory.py | 2 +- zipline/utils/factory.py | 19 +++- 12 files changed, 402 insertions(+), 290 deletions(-) create mode 100644 zipline/algorithm.py delete mode 100644 zipline/gens/cov.py diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 25528372..6d5ccce2 100644 --- a/tests/test_optimize.py +++ b/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 diff --git a/tests/test_transforms.py b/tests/test_transforms.py index d617ce20..0acdf004 100644 --- a/tests/test_transforms.py +++ b/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]) diff --git a/zipline/__init__.py b/zipline/__init__.py index 4151cd02..3916f8cf 100644 --- a/zipline/__init__.py +++ b/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 ] diff --git a/zipline/algorithm.py b/zipline/algorithm.py new file mode 100644 index 00000000..b102ea7e --- /dev/null +++ b/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 + + + diff --git a/zipline/gens/cov.py b/zipline/gens/cov.py deleted file mode 100644 index 85985795..00000000 --- a/zipline/gens/cov.py +++ /dev/null @@ -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() diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index 652ea135..aba3329b 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -212,6 +212,5 @@ class DataFrameSource(SpecificEquityTrades): yield ndict(event) - # Return the filtered event stream. return _generator() \ No newline at end of file diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 61527c1a..7975fb49 100644 --- a/zipline/gens/tradesimulation.py +++ b/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): """ diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index e4171ebc..3b253727 100644 --- a/zipline/gens/transform.py +++ b/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 diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index 098acad8..6625012a 100644 --- a/zipline/optimize/algorithms.py +++ b/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 diff --git a/zipline/optimize/example.py b/zipline/optimize/example.py index 02af8d7e..c8912558 100644 --- a/zipline/optimize/example.py +++ b/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: diff --git a/zipline/optimize/factory.py b/zipline/optimize/factory.py index ce16f20d..24a0783e 100644 --- a/zipline/optimize/factory.py +++ b/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 diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py index 58e58fd2..66804957 100644 --- a/zipline/utils/factory.py +++ b/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 From 2fe37fa34cc20a450befc728dc03de447ac4ac0d Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Wed, 19 Sep 2012 18:02:09 -0400 Subject: [PATCH 18/23] Fixed pandas depracation. --- zipline/optimize/factory.py | 3 +-- zipline/utils/factory.py | 4 ++-- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/zipline/optimize/factory.py b/zipline/optimize/factory.py index 24a0783e..48703a62 100644 --- a/zipline/optimize/factory.py +++ b/zipline/optimize/factory.py @@ -9,9 +9,8 @@ import pandas as pd 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.gens.tradegens import SpecificEquityTrades from zipline.optimize.algorithms import BuySellAlgorithmNew -from zipline.lines import SimulatedTrading from zipline.finance.slippage import FixedSlippage from copy import copy diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py index 66804957..c22b401b 100644 --- a/zipline/utils/factory.py +++ b/zipline/utils/factory.py @@ -238,8 +238,8 @@ def create_trade_source(sids, trade_count, trade_time_increment, trading_environ 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) + end = pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc) + index = pd.DatetimeIndex(start=start, end=end, freq=pd.datetools.day) x = np.arange(0, 16).reshape((8, 2)) df = pd.DataFrame(x, index=index, columns=[0, 1]) From 3be2f313cdf9caeb7117670dc272c8e614dad25f Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Wed, 19 Sep 2012 18:39:42 -0400 Subject: [PATCH 19/23] Shortened test window. Renamed functions in window transform. --- tests/test_transforms.py | 29 +++++++++++++---------------- zipline/gens/tradegens.py | 1 + zipline/gens/tradesimulation.py | 4 ++-- zipline/gens/transform.py | 17 ++++++++--------- zipline/optimize/factory.py | 1 - zipline/utils/factory.py | 4 ++-- 6 files changed, 26 insertions(+), 30 deletions(-) diff --git a/tests/test_transforms.py b/tests/test_transforms.py index 0acdf004..d33951b2 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -317,7 +317,7 @@ class BatchTransformAlgorithm(TradingAlgorithm): self.history_class.append(window_class) self.history_decorator.append(window_decorator) -class BatchTransformTestCase(TestCase): +class BatchTransformTestCase(): def setUp(self): setup_logger(self) self.source, self.df = factory.create_test_df_source() @@ -326,27 +326,24 @@ class BatchTransformTestCase(TestCase): algo = BatchTransformAlgorithm(sids=[0, 1]) algo.run(self.source) - assert algo.history_class[:2] == algo.history_decorator[:2] == [None, None] + assert algo.history_class[:2] == algo.history_decorator[:2] == [None, None], "First two iterations should return 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 + # every 2nd event should be identical because of refresh_period=2 + # not sure why actual length gets up to 4, bug in EventWindow? + assert np.all(algo.history_class[2][0].values == [2, 4, 6]) + assert np.all(algo.history_class[2][1].values == [3, 5, 7]) + assert np.all(algo.history_class[3][0].values == [2, 4, 6]) + assert np.all(algo.history_class[3][1].values == [3, 5, 7]) 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[2][0].values == [2, 4, 6]) + assert np.all(algo.history_decorator[2][1].values == [3, 5, 7]) + assert np.all(algo.history_decorator[3][0].values == [2, 4, 6]) + assert np.all(algo.history_decorator[3][1].values == [3, 5, 7]) 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]) + diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index aba3329b..c1b00c65 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -82,6 +82,7 @@ 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') diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 7975fb49..1c7f0efe 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -296,8 +296,8 @@ class AlgorithmSimulator(object): self.snapshot_dt = date start_tic = datetime.now() - with self.heartbeat_monitor: - self.algo.handle_data(self.universe) + #with self.heartbeat_monitor: + self.algo.handle_data(self.universe) stop_tic = datetime.now() # How long did you take? diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index 3b253727..8ec862eb 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -324,7 +324,11 @@ class BatchTransform(EventWindow): 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 + if func is not None: + self.compute_transform_value = func + else: + self.compute_transform_value = self.get_value + self.sids = sids self.refresh_period = refresh_period self.days = days @@ -352,7 +356,7 @@ class BatchTransform(EventWindow): self.update(data) # return newly computed or cached value - return self.compute() + return self.get_transform_value() def handle_add(self, event): if not self.last_refresh: @@ -398,17 +402,12 @@ class BatchTransform(EventWindow): def get_value(self, *args, **kwargs): raise NotImplementedError("Either overwrite get_value or provide a func argument.") - def compute(self, *args, **kwargs): + def get_transform_value(self, *args, **kwargs): if self.data is None: return None if self.updated: - if self.func is not None: - # user supplied function - self.cached = self.func(self.data, *args, **kwargs) - else: - # assume inheritance - self.cached = self.get_value(self.data, *args, **kwargs) + self.cached = self.compute_transform_value(self.data, *args, **kwargs) return self.cached diff --git a/zipline/optimize/factory.py b/zipline/optimize/factory.py index 48703a62..901798b3 100644 --- a/zipline/optimize/factory.py +++ b/zipline/optimize/factory.py @@ -4,7 +4,6 @@ Factory functions to prepare useful data for optimize tests. Author: Thomas V. Wiecki (thomas.wiecki@gmail.com), 2012 """ from datetime import timedelta -import pandas as pd import zipline.protocol as zp diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py index c22b401b..1f0026f5 100644 --- a/zipline/utils/factory.py +++ b/zipline/utils/factory.py @@ -237,10 +237,10 @@ def create_trade_source(sids, trade_count, trade_time_increment, trading_environ return source def create_test_df_source(): - start = pd.datetime(1990, 1, 1, 0, 0, 0, 0, pytz.utc) + start = pd.datetime(1990, 1, 3, 0, 0, 0, 0, pytz.utc) end = pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc) index = pd.DatetimeIndex(start=start, end=end, freq=pd.datetools.day) - x = np.arange(0, 16).reshape((8, 2)) + x = np.arange(0, 12).reshape((6, 2)) df = pd.DataFrame(x, index=index, columns=[0, 1]) return DataFrameSource(df), df From 4526cde4c511a279dadef70d697355ac03c7ca7a Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Wed, 19 Sep 2012 19:37:21 -0400 Subject: [PATCH 20/23] Fixed weird regression in TestAlgorithm. --- zipline/test_algorithms.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index 055c8c32..fbb9d5a9 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -59,7 +59,8 @@ class TestAlgorithm(): at the close of a simulation. """ - def __init__(self, sid, amount, sid_filter=None): + def __init__(self, sid, amount, order_count, sid_filter=None): + self.count = order_count self.sid = sid self.amount = amount self.incr = 0 From 2dbf905d43019e40943dd10ca8fdcf8e3648cb28 Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Wed, 19 Sep 2012 20:01:10 -0400 Subject: [PATCH 21/23] Added unittest for dataframe source. --- tests/test_sources.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 tests/test_sources.py diff --git a/tests/test_sources.py b/tests/test_sources.py new file mode 100644 index 00000000..0711f72f --- /dev/null +++ b/tests/test_sources.py @@ -0,0 +1,13 @@ +import zipline.utils.factory as factory + +def test_dataframe_source(): + source, df = factory.create_test_df_source() + + for expected_dt, expected_price in df.iterrows(): + sid0 = source.next() + sid1 = source.next() + + assert expected_dt == sid0.dt == sid1.dt + assert expected_price[0] == sid0.price + assert expected_price[1] == sid1.price + From 4324f95f36fa3d58fd60168c36cc5e3129994703 Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Fri, 21 Sep 2012 11:59:46 -0400 Subject: [PATCH 22/23] Better unittest coverage. DataFrameSource is now filtering sids. Fixed outstanding issues. --- tests/test_sources.py | 25 ++++++++---- tests/test_transforms.py | 57 ++++------------------------ zipline/__init__.py | 4 +- zipline/algorithm.py | 55 ++++++++++++++++++--------- zipline/gens/tradegens.py | 35 +++++++++++------ zipline/gens/tradesimulation.py | 6 +-- zipline/optimize/algorithms.py | 67 ++------------------------------- zipline/optimize/factory.py | 4 +- zipline/test_algorithms.py | 47 +++++++++++++++++++++++ zipline/utils/factory.py | 2 +- 10 files changed, 143 insertions(+), 159 deletions(-) diff --git a/tests/test_sources.py b/tests/test_sources.py index 0711f72f..548f8754 100644 --- a/tests/test_sources.py +++ b/tests/test_sources.py @@ -1,13 +1,22 @@ +from unittest2 import TestCase + import zipline.utils.factory as factory +from zipline.gens.tradegens import DataFrameSource -def test_dataframe_source(): - source, df = factory.create_test_df_source() +class TestDataFrameSource(TestCase): + def test_streaming_of_df(self): + source, df = factory.create_test_df_source() - for expected_dt, expected_price in df.iterrows(): - sid0 = source.next() - sid1 = source.next() + for expected_dt, expected_price in df.iterrows(): + sid0 = source.next() + sid1 = source.next() - assert expected_dt == sid0.dt == sid1.dt - assert expected_price[0] == sid0.price - assert expected_price[1] == sid1.price + assert expected_dt == sid0.dt == sid1.dt + assert expected_price[0] == sid0.price + assert expected_price[1] == sid1.price + def test_sid_filtering(self): + _, df = factory.create_test_df_source() + source = DataFrameSource(df, sids=[0]) + assert 1 not in [event.sid for event in source], \ + "DataFrameSource should only stream selected sid 0, not sid 1." \ No newline at end of file diff --git a/tests/test_transforms.py b/tests/test_transforms.py index d33951b2..ea1024f6 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -10,13 +10,14 @@ 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, BatchTransform, batch_transform +from zipline.gens.transform import StatefulTransform, EventWindow 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 + +from zipline.test_algorithms import BatchTransformAlgorithm def to_dt(msg): return ndict({'dt': msg}) @@ -288,36 +289,7 @@ class FinanceTransformsTestCase(TestCase): ############################################################ # 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(): +class BatchTransformTestCase(TestCase): def setUp(self): setup_logger(self) self.source, self.df = factory.create_test_df_source() @@ -329,21 +301,8 @@ class BatchTransformTestCase(): assert algo.history_class[:2] == algo.history_decorator[:2] == [None, None], "First two iterations should return None" # test overloaded class - # every 2nd event should be identical because of refresh_period=2 - # not sure why actual length gets up to 4, bug in EventWindow? - assert np.all(algo.history_class[2][0].values == [2, 4, 6]) - assert np.all(algo.history_class[2][1].values == [3, 5, 7]) - assert np.all(algo.history_class[3][0].values == [2, 4, 6]) - assert np.all(algo.history_class[3][1].values == [3, 5, 7]) - 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]) - - # test decorator - assert np.all(algo.history_decorator[2][0].values == [2, 4, 6]) - assert np.all(algo.history_decorator[2][1].values == [3, 5, 7]) - assert np.all(algo.history_decorator[3][0].values == [2, 4, 6]) - assert np.all(algo.history_decorator[3][1].values == [3, 5, 7]) - 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]) - + for test_history in [algo.history_class, algo.history_decorator]: + self.assertTrue(np.all(test_history[2].values.flatten() == range(4, 10))) + self.assertTrue(np.all(test_history[3].values.flatten() == range(4, 10))) + self.assertTrue(np.all(test_history[4].values.flatten() == range(6, 14))) diff --git a/zipline/__init__.py b/zipline/__init__.py index 3916f8cf..4151cd02 100644 --- a/zipline/__init__.py +++ b/zipline/__init__.py @@ -6,9 +6,7 @@ Zipline # it is a place to expose the public interfaces. from utils.protocol_utils import ndict -from algorithm import TradingAlgorithm __all__ = [ - ndict, - TradingAlgorithm + ndict ] diff --git a/zipline/algorithm.py b/zipline/algorithm.py index b102ea7e..08d388da 100644 --- a/zipline/algorithm.py +++ b/zipline/algorithm.py @@ -9,8 +9,8 @@ from zipline.finance.slippage import FixedSlippage class TradingAlgorithm(object): - """ - Base class for trading algorithms. Inherit and overload handle_data(data). + """Base class for trading algorithms. Inherit and overload + initialize() and handle_data(data). A new algorithm could look like this: ``` @@ -22,7 +22,7 @@ class TradingAlgorithm(object): sid = self.sids[0] self.order(sid, amount) ``` - To then run this algorithm: + To then to run this algorithm: >>> my_algo = MyAlgo(100, sids=[0]) >>> stats = my_algo.run(data) @@ -45,13 +45,13 @@ class TradingAlgorithm(object): # call to user-defined initialize method self.initialize(*args, **kwargs) - def _create_simulator(self, source): + def _create_simulator(self, start, end): """ 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]) + environment = create_trading_environment(start=start, end=end) # Create transforms by wrapping them into StatefulTransforms transforms = [] @@ -68,39 +68,59 @@ class TradingAlgorithm(object): # SimulatedTrading is the main class handling data streaming, # application of transforms and calling of the user algo. return SimulatedTrading( - [source], + self.sources, transforms, self, environment, FixedSlippage() ) - def run(self, source): - """ - Run the algorithm. + def run(self, source, start=None, end=None): + """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 + source : can be either: + - pandas.DataFrame + - zipline source + - list of zipline sources + + If pandas.DataFrame is provided, it must have the + following structure: + * column names must consist of ints representing the + different sids + * index must be DatetimeIndex + * array contents should be price info. :Returns: daily_stats : pandas.DataFrame Daily performance metrics such as returns, alpha etc. """ - if isinstance(source, pd.DataFrame): + if isinstance(source, (list, tuple)): + assert start is not None and end is not None, \ + "When providing a list of sources, start and end date have to be specified." + elif isinstance(source, pd.DataFrame): assert isinstance(source.index, pd.tseries.index.DatetimeIndex) + # if DataFrame provided, wrap in DataFrameSource source = DataFrameSource(source, sids=self.sids) + # If values not set, try to extract from source. + if start is None: + start = source.start + if end is None: + end = source.end + + if not isinstance(source, (list, tuple)): + self.sources = [source] + else: + self.sources = source + # create transforms and zipline - simulated_trading = self._create_simulator(source) + self.simulated_trading = self._create_simulator(start=start, end=end) # loop through simulated_trading, each iteration returns a # perf ndict - perfs = list(simulated_trading) + perfs = list(self.simulated_trading) # convert perf ndict to pandas dataframe daily_stats = self._create_daily_stats(perfs) @@ -123,6 +143,7 @@ class TradingAlgorithm(object): return daily_stats + def add_transform(self, transform_class, tag, *args, **kwargs): """Add a single-sid, sequential transform to the model. diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index c1b00c65..f088ae8b 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -9,6 +9,7 @@ from itertools import chain, cycle, ifilter, izip, repeat from datetime import datetime, timedelta import pandas as pd from copy import copy +import numpy as np from zipline.protocol import DATASOURCE_TYPE from zipline.utils import ndict @@ -77,17 +78,31 @@ class SpecificEquityTrades(object): # We shouldn't get any positional arguments. assert len(args) == 0 - # Unpack config dictionary with default values. - self.count = kwargs.get('count', 500) - 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') self.filter = kwargs.get('filter') + if self.event_list is not None: + # If event_list is provided, extract parameters from there + # This isn't really clean and ultimately I think this + # class should serve a single purpose (either take an + # event_list or autocreate events). + self.count = kwargs.get('count', len(self.event_list)) + self.sids = kwargs.get('sids', np.unique([event.sid for event in self.event_list]).tolist()) + self.start = kwargs.get('start', self.event_list[0].dt) + self.end = kwargs.get('start', self.event_list[-1].dt) + self.delta = kwargs.get('delta', self.event_list[1].dt - self.event_list[0].dt) + self.concurrent = kwargs.get('concurrent', False) + + else: + # Unpack config dictionary with default values. + self.count = kwargs.get('count', 500) + 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) + + # Hash_value for downstream sorting. self.arg_string = hash_args(*args, **kwargs) @@ -188,9 +203,6 @@ class DataFrameSource(SpecificEquityTrades): self.end = kwargs.get('end', data.index[-1]) self.delta = kwargs.get('delta', data.index[1]-data.index[0]) - # Default to None for event_list and filter. - self.filter = kwargs.get('filter') - # Hash_value for downstream sorting. self.arg_string = hash_args(data, **kwargs) @@ -214,4 +226,5 @@ class DataFrameSource(SpecificEquityTrades): yield ndict(event) # Return the filtered event stream. - return _generator() \ No newline at end of file + drop_sids = lambda x: x.sid in self.sids + return ifilter(drop_sids, _generator()) diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 1c7f0efe..5efb1505 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -194,8 +194,6 @@ 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( @@ -296,8 +294,8 @@ class AlgorithmSimulator(object): self.snapshot_dt = date start_tic = datetime.now() - #with self.heartbeat_monitor: - self.algo.handle_data(self.universe) + with self.heartbeat_monitor: + self.algo.handle_data(self.universe) stop_tic = datetime.now() # How long did you take? diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index 6625012a..976979d2 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -1,9 +1,9 @@ from logbook import Logger -from zipline import TradingAlgorithm +from zipline.algorithm import TradingAlgorithm logger = Logger('Algo') -class BuySellAlgorithm(object): +class BuySellAlgorithm(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 @@ -15,69 +15,11 @@ class BuySellAlgorithm(object): """ - def __init__(self, sid, amount, offset): - self.sid = sid + def initialize(self, amount=100, offset=0): 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 initialize(self): - pass - - def set_order(self, order_callable): - self.order = order_callable - - def set_portfolio(self, portfolio): - self.portfolio = portfolio - - def handle_data(self, frame): - print frame.sid - order_size = self.buy_or_sell * (self.amount - (self.offset**2)) - self.order(self.sid, order_size) - - #sell next time around. - self.buy_or_sell *= -1 - - self.orders.append(order_size) - - self.frame_count += 1 - self.incr += 1 - - def get_sid_filter(self): - return [self.sid] - - -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)) @@ -89,6 +31,3 @@ class BuySellAlgorithmNew(TradingAlgorithm): self.orders.append(order_size) - self.frame_count += 1 - self.incr += 1 - diff --git a/zipline/optimize/factory.py b/zipline/optimize/factory.py index 901798b3..14fea2ad 100644 --- a/zipline/optimize/factory.py +++ b/zipline/optimize/factory.py @@ -9,7 +9,7 @@ import zipline.protocol as zp from zipline.utils.factory import get_next_trading_dt, create_trading_environment from zipline.gens.tradegens import SpecificEquityTrades -from zipline.optimize.algorithms import BuySellAlgorithmNew +from zipline.optimize.algorithms import BuySellAlgorithm from zipline.finance.slippage import FixedSlippage from copy import copy @@ -120,7 +120,7 @@ def create_predictable_zipline(config, offset=0, simulate=True): amplitude) if 'algorithm' not in config: - algorithm = BuySellAlgorithmNew(sid, 100, offset) + algorithm = BuySellAlgorithm(sids=[sid], amount=100, offset=offset) config['order_count'] = trade_count - 1 config['trade_count'] = trade_count diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index fbb9d5a9..10795e33 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -52,6 +52,7 @@ The algorithm must expose methods: """ + class TestAlgorithm(): """ This algorithm will send a specified number of orders, to allow unit tests @@ -382,3 +383,49 @@ class TestLoggingAlgorithm(): def set_slippage_override(self, slippage_callable): pass + + +from datetime import timedelta +from zipline.algorithm import TradingAlgorithm +from zipline.gens.transform import BatchTransform, batch_transform +from zipline.gens.mavg import MovingAverage + +class TestRegisterTransformAlgorithm(TradingAlgorithm): + def initialize(self): + self.add_transform(MovingAverage, 'mavg', ['price'], + market_aware=True, + days=2) + + def handle_data(self, data): + pass + +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) + + diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py index 1f0026f5..7517fff5 100644 --- a/zipline/utils/factory.py +++ b/zipline/utils/factory.py @@ -240,7 +240,7 @@ def create_test_df_source(): start = pd.datetime(1990, 1, 3, 0, 0, 0, 0, pytz.utc) end = pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc) index = pd.DatetimeIndex(start=start, end=end, freq=pd.datetools.day) - x = np.arange(0, 12).reshape((6, 2)) + x = np.arange(2., 14.).reshape((6, 2)) df = pd.DataFrame(x, index=index, columns=[0, 1]) return DataFrameSource(df), df From 32c7dfbef9b245e8074b244f97eaace08d55d6d4 Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Fri, 21 Sep 2012 12:00:09 -0400 Subject: [PATCH 23/23] Added test_algorithm.py. --- tests/test_algorithm.py | 57 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) create mode 100644 tests/test_algorithm.py diff --git a/tests/test_algorithm.py b/tests/test_algorithm.py new file mode 100644 index 00000000..84f94008 --- /dev/null +++ b/tests/test_algorithm.py @@ -0,0 +1,57 @@ +from unittest2 import TestCase +from datetime import timedelta + +from zipline.utils.test_utils import setup_logger +import zipline.utils.factory as factory +from zipline.test_algorithms import TestRegisterTransformAlgorithm +from zipline.gens.tradegens import SpecificEquityTrades, DataFrameSource +from zipline.gens.mavg import MovingAverage + +class TestTransformAlgorithm(TestCase): + def setUp(self): + setup_logger(self) + self.trading_environment = factory.create_trading_environment() + setup_logger(self) + + trade_history = factory.create_trade_history( + 133, + [10.0, 10.0, 11.0, 11.0], + [100, 100, 100, 300], + timedelta(days=1), + self.trading_environment + ) + self.source = SpecificEquityTrades(event_list=trade_history) + + self.df_source, self.df = factory.create_test_df_source() + + def test_source_as_input(self): + algo = TestRegisterTransformAlgorithm(sids=[133]) + algo.run(self.source) + self.assertEqual(len(algo.sources), 1) + assert isinstance(algo.sources[0], SpecificEquityTrades) + + def test_multi_source_as_input_no_start_end(self): + algo = TestRegisterTransformAlgorithm(sids=[133]) + with self.assertRaises(AssertionError): + algo.run([self.source, self.df_source]) + + def test_multi_source_as_input(self): + algo = TestRegisterTransformAlgorithm(sids=[0, 1, 133]) + algo.run([self.source, self.df_source], start=self.df.index[0], end=self.df.index[-1]) + self.assertEqual(len(algo.sources), 2) + + def test_df_as_input(self): + algo = TestRegisterTransformAlgorithm(sids=[0, 1]) + algo.run(self.df) + assert isinstance(algo.sources[0], DataFrameSource) + + def test_transform_registered(self): + algo = TestRegisterTransformAlgorithm(sids=[133]) + algo.run(self.source) + assert algo.get_sid_filter() == algo.sids == [133] + assert 'mavg' in algo.registered_transforms + assert algo.registered_transforms['mavg']['args'] == (['price'],) + assert algo.registered_transforms['mavg']['kwargs'] == {'days': 2, 'market_aware': True} + assert algo.registered_transforms['mavg']['class'] is MovingAverage + +