diff --git a/tests/test_optimize.py b/tests/test_optimize.py index 365a2a23..25528372 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 @@ -24,7 +24,7 @@ class TestUpDown(TestCase): def setUp(self): self.zipline_test_config = { - 'sid' : 133, + 'sid' : [0], 'trade_count' : 5, 'amplitude' : 30, 'base_price' : 50 @@ -44,17 +44,17 @@ class TestUpDown(TestCase): UpDownSource and BuySellAlgorithm interact correctly." """ - zipline, config = create_predictable_zipline( + + algo, config = create_predictable_zipline( self.zipline_test_config, - offset=0, - simulate=False + offset=0 ) #extract arguments base_price = self.zipline_test_config['base_price'] amplitude = self.zipline_test_config['amplitude'] - prices = np.array([event.price for event in config['trade_source'].event_list]) + prices = config['trade_source'][0].values max_price_idx = np.where(prices==prices.max())[0] min_price_idx = np.where(prices==prices.min())[0] self.assertTrue(np.all(max_price_idx % 2 == 1), @@ -70,9 +70,9 @@ class TestUpDown(TestCase): "Minimum price does not equal expected maximum price." ) - zipline.simulate(blocking=True) + stats = algo.run(config['trade_source']) - algo = config['algorithm'] + self.assertTrue(len(stats) != 0) orders = np.asarray(algo.orders) max_order_idx = np.where(orders==orders.max())[0] @@ -108,12 +108,15 @@ class TestUpDown(TestCase): compound_returns = np.empty(len(test_offsets)) ziplines = [] for i, offset in enumerate(test_offsets): - zipline, config = create_predictable_zipline( + algo, config = create_predictable_zipline( self.zipline_test_config, offset=offset, ) - ziplines.append(zipline) - compound_returns[i] = zipline.get_cumulative_performance()['returns'] + results = algo.run(config['trade_source']) + ziplines.append(algo) + + compound_returns[i] = results.returns.sum() + self.assertTrue(np.all(compound_returns[supposed_max] > compound_returns[np.logical_not(supposed_max)]), "Maximum compound returns are not where they are supposed to be." diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index 05839a28..ef97bfd1 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -14,7 +14,7 @@ def date_sorted_sources(*sources): for source in sources: assert iter(source), "Source %s not iterable" % source - assert 'get_hash' in source.__class__.__dict__, "No get_hash" + assert hasattr(source, 'get_hash'), "No get_hash" # Get name hashes to pass to date_sort. names = [source.get_hash() for source in sources] diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index 0842b3bd..e7c4e375 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), @@ -78,7 +82,6 @@ class SpecificEquityTrades(object): self.sids = kwargs.get('sids', [1, 2]) self.start = kwargs.get('start', datetime(2008, 6, 6, 15, tzinfo = pytz.utc)) self.delta = kwargs.get('delta', timedelta(minutes = 1)) - self.concurrent = kwargs.get('concurrent', False) # Default to None for event_list and filter. self.event_list = kwargs.get('event_list') @@ -137,6 +140,7 @@ class SpecificEquityTrades(object): volumes = mock_volumes(self.count) sids = cycle(self.sids) + # Combine the iterators into a single iterator of arguments arg_gen = izip(sids, prices, volumes, dates) @@ -157,33 +161,57 @@ class SpecificEquityTrades(object): return filtered -# !!!!!!! Deprecated for now !!!!!!!!! +class DataFrameSource(SpecificEquityTrades): + """ + Yields all events in event_list that match the given sid_filter. + If no event_list is specified, generates an internal stream of events + to filter. Returns all events if filter is None. -def RandomEquityTrades(object): + Configuration options: - def __init__(self): - # We shouldn't get any positional args. - assert args == () + count : integer representing number of trades + sids : list of values representing simulated internal sids + start : start date + delta : timedelta between internal events + filter : filter to remove the sids + """ - self.count = config.get('count', 500) - self.sids = config.get('sids', [1,2]) - self.filter = config.get('filter') + def __init__(self, data, **kwargs): + assert isinstance(data.index, pd.tseries.index.DatetimeIndex) - dates = fuzzy_dates(count) - prices = mock_prices(count, rand = True) - volumes = mock_volumes(count, rand = True) - sids = cycle(sids) + self.data = data + # Unpack config dictionary with default values. + self.count = kwargs.get('count', 500) + self.sids = kwargs.get('sids', [0]) + self.start = kwargs.get('start', datetime(1957, 1, 1, 0, tzinfo = pytz.utc)) + self.end = kwargs.get('end', datetime(2010, 1, 1, tzinfo=pytz.utc)) + self.delta = kwargs.get('delta', timedelta(days = 1)) - arg_gen = izip(sids, prices, volumes, dates) + # Default to None for event_list and filter. + self.filter = kwargs.get('filter') - unfiltered = (create_trade(*args) for args in arg_gen) + # Hash_value for downstream sorting. + self.arg_string = hash_args(data, **kwargs) - if filter: - filtered = ifilter(lambda event: event.sid in filter, unfiltered) - else: - filtered = unfiltered - return filtered + self.generator = self.create_fresh_generator() -# if __name__ == "__main__": -# import nose.tools; nose.tools.set_trace() -# trades = SpecificEquityTrades(filter = [1]) + def create_fresh_generator(self): + def _generator(df=self.data): + for dt, series in df.iterrows(): + if (dt < self.start) or (dt > self.end): + continue + event = {'dt': dt, + 'source_id': self.get_hash(), + 'type': DATASOURCE_TYPE.TRADE + } + + for sid, price in series.iterkv(): + event = copy(event) + event['sid'] = 0 + event['price'] = price + + yield ndict(event) + + + # Return the filtered event stream. + return _generator() \ No newline at end of file diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 9c2df532..cca56e18 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -193,6 +193,8 @@ class AlgorithmSimulator(object): 'filled' : 0 }) + log.debug(order) + # Tell the user if they try to buy 0 shares of something. if order.amount == 0: zero_message = "Requested to trade zero shares of {sid}".format( diff --git a/zipline/optimize/algorithms.py b/zipline/optimize/algorithms.py index c4e3fb9d..0a860bd2 100644 --- a/zipline/optimize/algorithms.py +++ b/zipline/optimize/algorithms.py @@ -1,3 +1,18 @@ +import pandas as pd +import numpy as np + +from datetime import datetime +from zipline.gens.tradegens import DataFrameSource +from zipline import ndict +from zipline.utils.factory import create_trading_environment +from zipline.gens.transform import StatefulTransform +from zipline.lines import SimulatedTrading +from zipline.finance.slippage import FixedSlippage + +from logbook import Logger + +logger = Logger('Algo') + class BuySellAlgorithm(object): """Algorithm that buys and sells alternatingly. The amount for each order can be specified. In addition, an offset that will @@ -46,3 +61,132 @@ class BuySellAlgorithm(object): def get_sid_filter(self): return [self.sid] + +# Algorithm base class, user algorithms inherit from this as they +# don't want to have to copy and know about set_order and +# set_portfolio +class TradingAlgorithm(object): + def _setup(self): + assert hasattr(self, 'source'), 'source not set.' + assert hasattr(self, 'sids'), "sids not set." + + environment = create_trading_environment(start=self.data.index[0], end=self.data.index[-1]) + + # Create transforms by wrapping them into StatefulTransforms + transforms = [] + if hasattr(self, 'registered_transforms'): + for namestring, trans_descr in self.registered_transforms.iteritems(): + sf = StatefulTransform( + trans_descr['class'], + *trans_descr['args'], + **trans_descr['kwargs'] + ) + sf.namestring = namestring + + transforms.append(sf) + + + self.simulated_trading = SimulatedTrading( + [self.source], + transforms, + self, + environment, + FixedSlippage() + ) + + def _create_daily_stats(self, perfs): + # create daily stats dataframe + daily_perfs = [] + cum_perfs = [] + for perf in perfs: + if 'daily_perf' in perf: + daily_perfs.append(perf['daily_perf']) + else: + cum_perfs.append(perf) + + daily_dts = [np.datetime64(perf['period_close'], utc=True) for perf in daily_perfs] + daily_stats = pd.DataFrame(daily_perfs, index=daily_dts) + + return daily_stats + + def run(self, data, compute_risk_metrics=False): + self.source = DataFrameSource(data, sids=self.sids) + self.data = data + self._setup() + + # drain simulated_trading + perfs = [] + for perf in self.simulated_trading: + #from nose.tools import set_trace; set_trace() + perfs.append(perf) + + #perfs = list(self.simulated_trading) + + daily_stats = self._create_daily_stats(perfs) + return daily_stats + + def set_portfolio(self, portfolio): + self.portfolio = portfolio + + def set_order(self, order_callable): + self.order = order_callable + + def get_sid_filter(self): + return self.sids + + def set_logger(self, logger): + self.logger = logger + + def initialize(self): + pass + + def set_slippage_override(self, slippage_callable): + pass + + def add_transform(self, transform_class, tag, *args, **kwargs): + if not hasattr(self, 'registered_transforms'): + self.registered_transforms = {} + + self.registered_transforms[tag] = {'class': transform_class, + 'args': args, + 'kwargs': kwargs} + + +class BuySellAlgorithmNew(TradingAlgorithm): + """Algorithm that buys and sells alternatingly. The amount for + each order can be specified. In addition, an offset that will + quadratically reduce the amount that will be bought can be + specified. + + This algorithm is used to test the parameter optimization + framework. If combined with the UpDown trade source, an offset of + 0 will produce maximum returns. + + """ + + def __init__(self, sids, amount, offset): + self.sids = sids + self.amount = amount + self.incr = 0 + self.done = False + self.order = None + self.frame_count = 0 + self.portfolio = None + self.buy_or_sell = -1 + self.offset = offset + self.orders = [] + self.prices = [] + + def handle_data(self, data): + order_size = self.buy_or_sell * (self.amount - (self.offset**2)) + self.order(self.sids[0], order_size) + logger.debug("ordering" + str(order_size)) + + #sell next time around. + self.buy_or_sell *= -1 + + self.orders.append(order_size) + + self.frame_count += 1 + self.incr += 1 + diff --git a/zipline/optimize/example.py b/zipline/optimize/example.py new file mode 100644 index 00000000..feca6306 --- /dev/null +++ b/zipline/optimize/example.py @@ -0,0 +1,155 @@ +import pandas as pd + +import numpy as np +#from mpl_toolkits.mplot3d import Axes3D + +import matplotlib.pyplot as plt +import cProfile +from zipline.gens.mavg import MovingAverage +from zipline.optimize.algorithms import TradingAlgorithm +from datetime import timedelta + +#from mpi4py_map import map + +# Inherits from Algorithm base class +class DMA(TradingAlgorithm): + """Dual Moving Average algorithm. + """ + def __init__(self, sids, amount=100, short_window=20, long_window=40): + self.sids = sids + self.amount = amount + self.done = False + self.order = None + self.frame_count = 0 + self.portfolio = None + self.orders = [] + + self.prices = [] + self.events = 0 + + self.invested = {} + for sid in self.sids: + self.invested[sid] = False + + self.add_transform(MovingAverage, 'short_mavg', ['price'], + market_aware=True, + days=short_window) #timedelta(days=int(short_window))) + + self.add_transform(MovingAverage, 'long_mavg', ['price'], + market_aware=True, + days=long_window) #timedelta(days=int(long_window))) + + def handle_data(self, data): + self.events += 1 + + for sid in self.sids: + # access transforms via their user-defined tag + if (data[sid].short_mavg['price'] > data[sid].long_mavg['price']) and not self.invested[sid]: + self.order(sid, self.amount) + self.invested[sid] = True + elif (data[sid].short_mavg['price'] < data[sid].long_mavg['price']) and self.invested[sid]: + self.order(sid, -self.amount) + self.invested[sid] = False + + +def load_close_px(indexes=None, stocks=None): + from pandas.io.data import DataReader + import pytz + + if indexes is None: + indexes = {'SPX' : '^GSPC'} + if stocks is None: + stocks = ['AAPL'] #, 'GE', 'IBM', 'MSFT', 'XOM', 'AA', 'JNJ', 'PEP'] + + #start = pd.datetime(1990, 1, 1) + start = pd.datetime(1990, 1, 1, 0, 0, 0, 0, pytz.utc) + end = pd.datetime(1992, 1, 1, 0, 0, 0, 0, pytz.utc) #pd.datetime.today() + + data = {} + + for stock in stocks: + print stock + stkd = DataReader(stock, 'yahoo', start, end).sort_index() + data[stock] = stkd + + for name, ticker in indexes.iteritems(): + print name + stkd = DataReader(ticker, 'yahoo', start, end).sort_index() + data[name] = stkd + + #df = pd.DataFrame({key: d['Close'] for key, d in data.iteritems()}) + df = pd.DataFrame({i: d['Close'] for i, d in enumerate(data.itervalues())}) + df.index = df.index.tz_localize(pytz.utc) + + return df + + +def run((short_window, long_window)): + #data = pd.DataFrame.from_csv('SP500.csv') + data = load_close_px() + myalgo = DMA([0], amount=100, short_window=short_window, long_window=long_window) + stats = myalgo.run(data) + stats['sw'] = short_window + stats['lw'] = long_window + return stats + +def explore_params(): + sws, lws = np.mgrid[10:20:5, 10:20:5] + + stats_all = map(run, zip(sws.flatten(), lws.flatten())) + stats = pd.concat(stats_all) + returns = stats.groupby(['sw', 'lw']).sum() + + plt.contourf(sws, lws, returns.returns.reshape(sws.shape)) + plt.xlabel('Short window length') + plt.ylabel('Long window length') + plt.savefig('DMA_contour.png') + plt.show() + +#stats = run((10, 50)) + +def get_opt_holdings_qp(univ_rets, track_rets): + from cvxopt import matrix + from cvxopt.solvers import qp + # set up the QP for CVXOPT + # .5 x' P x + q'x + # P = 2 * R'R + # q = - 2 * bmk'R + R = univ_rets.values + b = track_rets.values + P = matrix(2 * np.dot(R.T, R)) + q = matrix(-2 * np.dot(R.T, b)) + result = qp(P, q) + if result['status'] != 'optimal': + raise Exception('optimum not reached by QP') + return pd.Series(np.array(result['x']).ravel(), index=univ_rets.columns) + +def opt_portfolio(cov, budget, min_return): + from cvxopt import matrix + from cvxopt.solvers import qp + n = len(cov) + cov = matrix(2 * cov) + q = matrix(np.zeros(n)) + + h = matrix(budget) # G*x < h + # coneqp + result = qp(cov, q, h=h) + if result['status'] != 'optimal': + raise Exception('optimum not reached by QP') + + return pd.Series(np.array(result['x']).ravel()) + +def calc_te(weights, univ_rets, track_rets): + port_rets = (univ_rets * weights).sum(1) + return (port_rets - track_rets).std() + +def plot_returns(port_returns, bmk_returns): + plt.figure() + cum_port = ((1 + port_returns).cumprod() - 1) + cum_bmk = ((1 + bmk_returns).cumprod() - 1) + # cum_port = port_returns.cumsum() + # cum_bmk = bmk_returns.cumsum() + cum_port.plot(label='Portfolio returns') + cum_bmk.plot(label='Benchmark') + plt.title('Portfolio performance') + plt.legend(loc='best') diff --git a/zipline/optimize/factory.py b/zipline/optimize/factory.py index 8967d28a..ce16f20d 100644 --- a/zipline/optimize/factory.py +++ b/zipline/optimize/factory.py @@ -122,7 +122,7 @@ def create_predictable_zipline(config, offset=0, simulate=True): amplitude) if 'algorithm' not in config: - config['algorithm'] = BuySellAlgorithm(sid, 100, offset) + algorithm = BuySellAlgorithmNew(sid, 100, offset) config['order_count'] = trade_count - 1 config['trade_count'] = trade_count @@ -131,9 +131,4 @@ def create_predictable_zipline(config, offset=0, simulate=True): config['slippage'] = FixedSlippage() config['devel'] = True - zipline = SimulatedTrading.create_test_zipline(**config) - - if simulate: - zipline.simulate(blocking=True) - - return zipline, config + return algorithm, config diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index bcff7f81..3b04b243 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -52,7 +52,6 @@ The algorithm must expose methods: """ - class TestAlgorithm(): """ This algorithm will send a specified number of orders, to allow unit tests @@ -60,8 +59,7 @@ class TestAlgorithm(): at the close of a simulation. """ - def __init__(self, sid, amount, order_count, sid_filter=None): - self.count = order_count + def __init__(self, sid, amount, sid_filter=None): self.sid = sid self.amount = amount self.incr = 0 @@ -69,6 +67,7 @@ class TestAlgorithm(): self.order = None self.frame_count = 0 self.portfolio = None + if sid_filter: self.sid_filter = sid_filter else: diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py index 7146c0de..58e58fd2 100644 --- a/zipline/utils/factory.py +++ b/zipline/utils/factory.py @@ -14,7 +14,6 @@ from zipline.utils.protocol_utils import ndict import zipline.finance.risk as risk -from zipline.gens.tradegens import RandomEquityTrades from zipline.gens.tradegens import SpecificEquityTrades from zipline.gens.utils import create_trade from zipline.finance.trading import TradingEnvironment @@ -57,12 +56,15 @@ def load_market_data(): return bm_returns, tr_curves -def create_trading_environment(year=2006): +def create_trading_environment(year=2006, start=None, end=None): """Construct a complete environment with reasonable defaults""" benchmark_returns, treasury_curves = load_market_data() - start = datetime(year, 1, 1, tzinfo=pytz.utc) - end = datetime(year, 12, 31, tzinfo=pytz.utc) + if start is None: + start = datetime(year, 1, 1, tzinfo=pytz.utc) + if end is None: + end = datetime(year, 12, 31, tzinfo=pytz.utc) + trading_environment = TradingEnvironment( benchmark_returns, treasury_curves,