diff --git a/tests/finance/__init__.py b/tests/finance/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/finance/test_slippage.py b/tests/finance/test_slippage.py new file mode 100644 index 00000000..2d12329f --- /dev/null +++ b/tests/finance/test_slippage.py @@ -0,0 +1,77 @@ +# +# Copyright 2012 Quantopian, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Unit tests for finance.slippage +""" +import datetime + +import pytz + +from unittest2 import TestCase + +from zipline.finance.slippage import VolumeShareSlippage + +from zipline import ndict + + +class SlippageTestCase(TestCase): + + def test_volume_share_slippage(self): + + event = ndict( + {'volume': 200, + 'TRANSACTION': None, + 'type': 4, + 'price': 3.0, + 'datetime': datetime.datetime( + 2006, 1, 5, 14, 31, tzinfo=pytz.utc), + 'high': 3.15, + 'low': 2.85, + 'sid': 133, + 'source_id': 'test_source', + 'close': 3.0, + 'dt': + datetime.datetime(2006, 1, 5, 14, 31, tzinfo=pytz.utc), + 'open': 3.0} + ) + + slippage_model = VolumeShareSlippage() + + open_orders = {133: [ + ndict( + {'dt': datetime.datetime(2006, 1, 5, 14, 30, tzinfo=pytz.utc), + 'amount': 100, + 'filled': 0, 'sid': 133}) + ] + } + + txn = slippage_model.simulate( + event, + open_orders + ) + + expected_txn = { + 'price': float(3.01875), + 'dt': datetime.datetime( + 2006, 1, 5, 14, 31, tzinfo=pytz.utc), + 'amount': int(50), + 'sid': int(133) + } + + self.assertIsNotNone(txn) + + for key, value in expected_txn.items(): + self.assertEquals(value, txn[key]) diff --git a/tests/test_algorithm.py b/tests/test_algorithm.py index f908ed7d..52666921 100644 --- a/tests/test_algorithm.py +++ b/tests/test_algorithm.py @@ -65,7 +65,6 @@ class TestTransformAlgorithm(TestCase): 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'] == \ diff --git a/tests/test_exception_handling.py b/tests/test_exception_handling.py index 269522b8..fefd2bfe 100644 --- a/tests/test_exception_handling.py +++ b/tests/test_exception_handling.py @@ -17,11 +17,12 @@ from unittest2 import TestCase from collections import defaultdict import zipline.utils.simfactory as simfactory -from zipline.test_algorithms import ExceptionAlgorithm, DivByZeroAlgorithm, \ - InitializeTimeoutAlgorithm, TooMuchProcessingAlgorithm +from zipline.test_algorithms import ( + ExceptionAlgorithm, + DivByZeroAlgorithm, +) from zipline.finance.slippage import FixedSlippage from zipline.gens.transform import StatefulTransform -from zipline.utils.timeout import TimeoutException from zipline.utils.test_utils import ( @@ -78,25 +79,6 @@ class ExceptionTestCase(TestCase): self.assertEqual(ctx.exception.message, 'An assertion message') - def test_exception_in_init(self): - # Simulation - # ---------- - self.zipline_test_config['algorithm'] = \ - ExceptionAlgorithm( - 'initialize', - self.zipline_test_config['sid'] - ) - - zipline = simfactory.create_test_zipline( - **self.zipline_test_config - ) - - with self.assertRaises(Exception) as ctx: - output, _ = drain_zipline(self, zipline) - - self.assertEqual(ctx.exception.message, - 'Algo exception in initialize') - def test_exception_in_handle_data(self): # Simulation # ---------- @@ -134,37 +116,3 @@ class ExceptionTestCase(TestCase): self.assertEqual(ctx.exception.message, 'integer division or modulo by zero') - - def test_initialize_timeout(self): - - self.zipline_test_config['algorithm'] = \ - InitializeTimeoutAlgorithm( - self.zipline_test_config['sid'] - ) - - zipline = simfactory.create_test_zipline( - **self.zipline_test_config - ) - - with self.assertRaises(TimeoutException) as ctx: - output, _ = drain_zipline(self, zipline) - - self.assertEqual(ctx.exception.message, 'Call to initialize timed out') - - def test_heartbeat(self): - - self.zipline_test_config['algorithm'] = \ - TooMuchProcessingAlgorithm( - self.zipline_test_config['sid'] - ) - zipline = simfactory.create_test_zipline( - **self.zipline_test_config - ) - - with self.assertRaises(TimeoutException) as ctx: - output, _ = drain_zipline(self, zipline) - - self.assertEqual( - ctx.exception.message, - 'Too much time spent in handle_data call' - ) diff --git a/tests/test_finance.py b/tests/test_finance.py index 224b5db2..28718a6a 100644 --- a/tests/test_finance.py +++ b/tests/test_finance.py @@ -197,40 +197,19 @@ class FinanceTestCase(TestCase): } self.transaction_sim(**params2) - @timed(DEFAULT_TIMEOUT) - def test_partial_expiration_orders(self): - # create a scenario where orders expire without being filled - # entirely - params1 = { - 'trade_count': 100, + # Runs the collapsed trades over daily trade intervals. + # Ensuring that our delay works for daily intervals as well. + params3 = { + 'trade_count': 6, 'trade_amount': 100, - 'trade_delay': timedelta(minutes=5), 'trade_interval': timedelta(days=1), - 'order_count': 3, - 'order_amount': 1000, - 'order_interval': timedelta(minutes=30), - # because we placed an orders totaling less than 25% of one trade - # the simulator should produce just one transaction. + 'order_count': 24, + 'order_amount': 1, + 'order_interval': timedelta(minutes=1), 'expected_txn_count': 1, - 'expected_txn_volume': 25 + 'expected_txn_volume': 24 * 1 } - self.transaction_sim(**params1) - - # same scenario, but short sales. - params2 = { - 'trade_count': 100, - 'trade_amount': 100, - 'trade_delay': timedelta(minutes=5), - 'trade_interval': timedelta(days=1), - 'order_count': 3, - 'order_amount': -1000, - 'order_interval': timedelta(minutes=30), - # because we placed an orders totaling less than 25% of one trade - # the simulator should produce just one transaction. - 'expected_txn_count': 1, - 'expected_txn_volume': -25 - } - self.transaction_sim(**params2) + self.transaction_sim(**params3) @timed(DEFAULT_TIMEOUT) def test_alternating_long_short(self): @@ -320,7 +299,7 @@ class FinanceTestCase(TestCase): self.assertEqual(order.sid, sid) self.assertEqual(order.amount, order_amount * alternator ** i) - tracker = PerformanceTracker(trading_environment, [sid]) + tracker = PerformanceTracker(trading_environment) # this approximates the loop inside TradingSimulationClient transactions = [] diff --git a/tests/test_perf_tracking.py b/tests/test_perf_tracking.py index 1fcf8baa..57f62d78 100644 --- a/tests/test_perf_tracking.py +++ b/tests/test_perf_tracking.py @@ -567,8 +567,7 @@ shares in position" 'price', 'changed'] perf_tracker = perf.PerformanceTracker( - self.trading_environment, - [sid, sid2] + self.trading_environment ) for event in trade_history: diff --git a/tests/test_transforms.py b/tests/test_transforms.py index e87551c8..85daa29c 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -32,6 +32,7 @@ from zipline.gens.stddev import MovingStandardDev from zipline.gens.returns import Returns import zipline.utils.factory as factory + from zipline.test_algorithms import BatchTransformAlgorithm @@ -315,7 +316,7 @@ class BatchTransformTestCase(TestCase): self.source, self.df = factory.create_test_df_source() def test_event_window(self): - algo = BatchTransformAlgorithm(sids=[0, 1]) + algo = BatchTransformAlgorithm() algo.run(self.source) self.assertEqual(algo.history_return_price_class[:2], @@ -344,7 +345,7 @@ class BatchTransformTestCase(TestCase): ) def test_passing_of_args(self): - algo = BatchTransformAlgorithm([0, 1], 1, kwarg='str') + algo = BatchTransformAlgorithm(1, kwarg='str') self.assertEqual(algo.args, (1,)) self.assertEqual(algo.kwargs, {'kwarg': 'str'}) diff --git a/zipline/MESSAGES.py b/zipline/MESSAGES.py new file mode 100644 index 00000000..058a5707 --- /dev/null +++ b/zipline/MESSAGES.py @@ -0,0 +1,38 @@ +from zipline import ndict +# --------------------- +# Error Messages. +# User facing. +# --------------------- + +ERRORS = ndict({ + + # Raised if a user script calls the override_slippage magic + # with a slipage object that isn't a VolumeShareSlippage or + # FixedSlipapge + 'UNSUPPORTED_SLIPPAGE_MODEL': +"You attempted to override slippage with an unsupported class. \ +Please use VolumeShareSlippage or FixedSlippage.", + + # Raised if a users script calls override_slippage magic + # after the initialize method has returned. + 'OVERRIDE_SLIPPAGE_POST_INIT': +"You attempted to override slippage after the simulation has \ +started. You may only call override_slippage in your initialize \ +method.", + + # Raised if a user script calls the override_commission magic + # with a commission object that isn't a PerShare or + # PerTrade commission + 'UNSUPPORTED_COMMISSION_MODEL': +"You attempted to override commission with an unsupported class. \ +Please use PerShare or PerTrade.", + + # Raised if a users script calls override_commission magic + # after the initialize method has returned. + 'OVERRIDE_COMMISSION_POST_INIT': +"You attempted to override commission after the simulation has \ +started. You may only call override_commission in your initialize \ +method.", + + +}) diff --git a/zipline/algorithm.py b/zipline/algorithm.py index 4379c5b6..8780763e 100644 --- a/zipline/algorithm.py +++ b/zipline/algorithm.py @@ -19,9 +19,20 @@ 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, transact_partial -from zipline.finance.commission import PerShare +from zipline.finance.slippage import ( + VolumeShareSlippage, + FixedSlippage, + transact_partial +) +from zipline.finance.commission import PerShare, PerTrade + + +from zipline.gens.composites import ( + date_sorted_sources, + sequential_transforms +) +from zipline.gens.tradesimulation import TradeSimulationClient as tsc +from zipline import MESSAGES class TradingAlgorithm(object): @@ -44,55 +55,59 @@ class TradingAlgorithm(object): >>> stats = my_algo.run(data) """ - def __init__(self, sids, *args, **kwargs): + def __init__(self, *args, **kwargs): """ Initialize sids and other state variables. - - Calls user-defined initialize() forwarding *args and **kwargs. """ - self.sids = sids self.done = False self.order = None self.frame_count = 0 self.portfolio = None self.registered_transforms = {} + self.transforms = [] + self.sources = [] - # call to user-defined initialize method + self.logger = None + + # default components for transact + self.slippage = VolumeShareSlippage() + self.commission = PerShare() + + # an algorithm subclass needs to set initialized to True + # when it is fully initialized. + self.initialized = False + + # call to user-defined constructor method self.initialize(*args, **kwargs) - self.initialized = True - - def _create_simulator(self, start, end): + def _create_generator(self, environment): """ - Create trading environment, transforms and SimulatedTrading object. - - Gets called by self.run(). + Create a basic generator setup using the sources and + transforms attached to this algorithm. """ - environment = create_trading_environment(start=start, end=end) - # 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 + self.date_sorted = date_sorted_sources(*self.sources) + self.with_tnfms = sequential_transforms(self.date_sorted, + *self.transforms) + self.trading_client = tsc(self, environment) - transforms.append(sf) + transact_method = transact_partial(self.slippage, self.commission) + self.set_transact(transact_method) - # SimulatedTrading is the main class handling data streaming, - # application of transforms and calling of the user algo. - return SimulatedTrading( - self.sources, - transforms, - self, - environment, - transact_partial(FixedSlippage(), PerShare(0.0)) - ) + return self.trading_client.simulate(self.with_tnfms) + def get_generator(self, environment): + """ + Override this method to add new logic to the construction + of the generator. Overrides can use the _create_generator + method to get a standard construction generator. + """ + return self._create_generator(environment) + + # TODO: make a new subclass, e.g. BatchAlgorithm, and move + # the run method to the subclass, and refactor to put the + # generator creation logic into get_generator. def run(self, source, start=None, end=None): """Run the algorithm. @@ -121,7 +136,7 @@ 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) + source = DataFrameSource(source) # If values not set, try to extract from source. if start is None: @@ -134,12 +149,25 @@ start and end date have to be specified.""" else: self.sources = source + # Create transforms by wrapping them into StatefulTransforms + for namestring, trans_descr in self.registered_transforms.iteritems(): + sf = StatefulTransform( + trans_descr['class'], + *trans_descr['args'], + **trans_descr['kwargs'] + ) + sf.namestring = namestring + + self.transforms.append(sf) + + environment = create_trading_environment(start=start, end=end) + # create transforms and zipline - self.simulated_trading = self._create_simulator(start=start, end=end) + self.gen = self._create_generator(environment) # loop through simulated_trading, each iteration returns a # perf ndict - perfs = list(self.simulated_trading) + perfs = list(self.gen) # convert perf ndict to pandas dataframe daily_stats = self._create_daily_stats(perfs) @@ -186,14 +214,39 @@ start and end date have to be specified.""" 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): + def init(self, *args, **kwargs): + """Called from constructor.""" pass - def set_transact_setter(self, transact_setter): - pass + def set_transact(self, transact): + """ + Set the method that will be called to create a + transaction from open orders and trade events. + """ + self.trading_client.ordering_client.transact = transact + + def set_slippage(self, slippage): + assert isinstance(slippage, (VolumeShareSlippage, FixedSlippage)), \ + MESSAGES.ERRORS.UNSUPPORTED_SLIPPAGE_MODEL + if self.initialized: + raise Exception(MESSAGES.ERRORS.OVERRIDE_SLIPPAGE_POST_INIT) + self.slippage = slippage + + def set_commission(self, commission): + assert isinstance(commission, (PerShare, PerTrade)), \ + MESSAGES.ERRORS.UNSUPPORTED_COMMISSION_MODEL + + if self.initialized: + raise Exception(MESSAGES.ERRORS.OVERRIDE_COMMISSION_POST_INIT) + self.commission = commission + + def set_sources(self, sources): + assert isinstance(sources, list) + self.sources = sources + + def set_transforms(self, transforms): + assert isinstance(transforms, list) + self.transforms = transforms diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 228a98b8..a24c6f12 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -154,7 +154,7 @@ class PerformanceTracker(object): """ - def __init__(self, trading_environment, sid_list): + def __init__(self, trading_environment): self.trading_environment = trading_environment self.trading_day = datetime.timedelta(hours=6, minutes=30) @@ -178,7 +178,7 @@ class PerformanceTracker(object): # this performance period will span the entire simulation. self.cumulative_performance = PerformancePeriod( # initial positions are empty - {}, + positiondict(), # initial portfolio positions have zero value 0, # initial cash is your capital base. @@ -191,7 +191,7 @@ class PerformanceTracker(object): # this performance period will span just the current market day self.todays_performance = PerformancePeriod( # initial positions are empty - {}, + positiondict(), # initial portfolio positions have zero value 0, # initial cash is your capital base. @@ -203,10 +203,6 @@ class PerformanceTracker(object): keep_transactions=True ) - for sid in sid_list: - self.cumulative_performance.positions[sid] = Position(sid) - self.todays_performance.positions[sid] = Position(sid) - def transform(self, stream_in): """ Main generator work loop. @@ -409,7 +405,11 @@ class PerformancePeriod(object): self.period_capital_used = 0.0 self.pnl = 0.0 #sid => position object - self.positions = initial_positions + if not isinstance(initial_positions, positiondict): + self.positions = positiondict() + self.positions.update(initial_positions) + else: + self.positions = initial_positions self.starting_value = starting_value #cash balance at start of period self.starting_cash = starting_cash @@ -436,11 +436,8 @@ class PerformancePeriod(object): self.returns = 0.0 def execute_transaction(self, txn): - # Update Position # ---------------- - if txn.sid not in self.positions: - self.positions[txn.sid] = Position(txn.sid) self.positions[txn.sid].update(txn) self.period_capital_used += -1 * txn.price * txn.amount @@ -547,21 +544,16 @@ class PerformancePeriod(object): del(portfolio['max_leverage']) del(portfolio['max_capital_used']) - portfolio['positions'] = self.get_positions(ndicted=True) + portfolio['positions'] = self.get_positions() return zp.ndict(portfolio) - def get_positions(self, ndicted=False): - if ndicted: - positions = zp.ndict({}) - else: - positions = {} + def get_positions(self): + + positions = zp.ndict(internal=position_ndict()) for sid, pos in self.positions.iteritems(): cur = pos.to_dict() - if ndicted: - positions[sid] = zp.ndict(cur) - else: - positions[sid] = cur + positions[sid] = zp.ndict(cur) return positions @@ -571,3 +563,19 @@ class PerformancePeriod(object): cur = pos.to_dict() positions.append(cur) return positions + + +class positiondict(dict): + + def __missing__(self, key): + pos = Position(key) + self[key] = pos + return pos + + +class position_ndict(dict): + + def __missing__(self, key): + pos = Position(key) + self[key] = zp.ndict(pos.to_dict()) + return pos diff --git a/zipline/finance/slippage.py b/zipline/finance/slippage.py index fc53fa1d..49d539ab 100644 --- a/zipline/finance/slippage.py +++ b/zipline/finance/slippage.py @@ -12,6 +12,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +from datetime import timedelta import pytz import math @@ -21,13 +22,12 @@ from functools import partial import zipline.protocol as zp -def transact_stub(slippage, commission, open_orders, events): +def transact_stub(slippage, commission, event, open_orders): """ This is intended to be wrapped in a partial, so that the slippage and commission models can be enclosed. """ - - transaction = slippage.simulate(open_orders, events) + transaction = slippage.simulate(event, open_orders) if transaction and transaction.amount != 0: direction = abs(transaction.amount) / transaction.amount per_share, total_commission = commission.calculate(transaction) @@ -55,11 +55,13 @@ def create_transaction(sid, amount, price, dt): class VolumeShareSlippage(object): def __init__(self, - volume_limit=.25, - price_impact=0.1): + volume_limit=.25, + price_impact=0.1, + delay=timedelta(minutes=1)): self.volume_limit = volume_limit self.price_impact = price_impact + self.delay = delay def simulate(self, event, open_orders): @@ -71,6 +73,10 @@ class VolumeShareSlippage(object): if event.sid in open_orders: orders = open_orders[event.sid] orders = sorted(orders, key=lambda o: o.dt) + # Only use orders for the current day or before + current_orders = filter( + lambda o: o.dt + self.delay <= event.dt, + orders) else: return None @@ -79,42 +85,36 @@ class VolumeShareSlippage(object): simulated_amount = 0 simulated_impact = 0.0 direction = 1.0 - for order in orders: - if(order.dt < event.dt): + for order in current_orders: - # orders are only good on the day they are issued - if order.dt.day < event.dt.day: - continue + open_amount = order.amount - order.filled - open_amount = order.amount - order.filled + if(open_amount != 0): + direction = open_amount / math.fabs(open_amount) + else: + direction = 1 - if(open_amount != 0): - direction = open_amount / math.fabs(open_amount) - else: - direction = 1 + desired_order = total_order + open_amount - desired_order = total_order + open_amount + volume_share = min(direction * (desired_order) / event.volume, + self.volume_limit) + simulated_amount = int(volume_share * event.volume * direction) + simulated_impact = (volume_share) ** 2 \ + * self.price_impact * direction * event.price - volume_share = direction * (desired_order) / event.volume - if volume_share > self.volume_limit: - volume_share = self.volume_limit - simulated_amount = int(volume_share * event.volume * direction) - simulated_impact = (volume_share) ** 2 \ - * self.price_impact * direction * event.price + order.filled += (simulated_amount - total_order) + total_order = simulated_amount - order.filled += (simulated_amount - total_order) - total_order = simulated_amount + # we cap the volume share at configured % of a trade + if volume_share == self.volume_limit: + break - # we cap the volume share at configured % of a trade - if volume_share == self.volume_limit: - break + filled_orders = [x for x in orders + if abs(x.amount - x.filled) > 0 + and x.dt.day >= event.dt.day] - orders = [x for x in orders - if abs(x.amount - x.filled) > 0 - and x.dt.day >= event.dt.day] - - open_orders[event.sid] = orders + open_orders[event.sid] = filled_orders if simulated_amount != 0: return create_transaction( diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 133b45ba..502d0637 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -21,9 +21,9 @@ from collections import defaultdict import zipline.protocol as zp from zipline.finance.slippage import ( - VolumeShareSlippage, - transact_partial - ) + VolumeShareSlippage, + transact_partial +) from zipline.finance.commission import PerShare log = logbook.Logger('Transaction Simulator') @@ -31,16 +31,8 @@ log = logbook.Logger('Transaction Simulator') class TransactionSimulator(object): - def __init__(self, transact=None): - - if transact is not None: - self.transact = transact - else: - self.transact = transact_partial( - VolumeShareSlippage(), - PerShare() - ) - + def __init__(self): + self.transact = transact_partial(VolumeShareSlippage(), PerShare()) self.open_orders = defaultdict(list) def place_order(self, order): diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 819b21d6..8a702fbf 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -15,13 +15,13 @@ from logbook import Logger, Processor +from collections import defaultdict from datetime import datetime from itertools import groupby from operator import attrgetter from zipline import ndict -from zipline.utils.timeout import Heartbeat, Timeout from zipline.finance.trading import TransactionSimulator from zipline.finance.performance import PerformanceTracker @@ -29,11 +29,6 @@ from zipline.gens.utils import hash_args log = Logger('Trade Simulation') -# TODO: make these arguments rather than global constants -INIT_TIMEOUT = 5 -HEARTBEAT_INTERVAL = 1 # seconds -MAX_HEARTBEAT_INTERVALS = 15 # count - class TradeSimulationClient(object): """ @@ -69,15 +64,13 @@ class TradeSimulationClient(object): is sent to the algo. """ - def __init__(self, algo, environment, transact): + def __init__(self, algo, environment): self.algo = algo - self.sids = algo.get_sid_filter() self.environment = environment - self.transact = transact - self.ordering_client = TransactionSimulator(self.transact) - self.perf_tracker = PerformanceTracker(self.environment, self.sids) + self.ordering_client = TransactionSimulator() + self.perf_tracker = PerformanceTracker(self.environment) self.algo_start = self.environment.first_open self.algo_sim = AlgorithmSimulator( @@ -139,44 +132,20 @@ class AlgorithmSimulator(object): self.order_book = order_book self.algo = algo - self.sids = algo.get_sid_filter() self.algo_start = algo_start # Monkey patch the user algorithm to place orders in the # TransactionSimulator's order book and use our logger. self.algo.set_order(self.order) - self.algolog = Logger("AlgoLog") - self.algo.set_logger(self.algolog) - - # Provide user algorithm with a setter for the transact - # method (method that constructs transactions based on - # open orders and trade events). - self.algo.set_transact_setter(self.set_transact) - - # Handler for heartbeats during calls to handle_data. - def log_heartbeats(beat_count, stackframe): - t = beat_count * HEARTBEAT_INTERVAL - warning = "handle_data has been processing for %i seconds" % t - self.algolog.warn(warning) - - # Context manager that calls log_heartbeats every HEARTBEAT_INTERVAL - # seconds, raising an exception after MAX_HEARTBEATS - self.heartbeat_monitor = Heartbeat( - HEARTBEAT_INTERVAL, - MAX_HEARTBEAT_INTERVALS, - frame_handler=log_heartbeats, - timeout_message="Too much time spent in handle_data call" - ) # ============== # Snapshot Setup # ============== # The algorithm's universe as of our most recent event. - self.universe = ndict() - for sid in self.sids: - self.universe[sid] = ndict() - self.universe.portfolio = None + # We want an ndict that will have empty ndicts as default + # values on missing keys. + self.universe = ndict(internal=defaultdict(ndict)) # We don't have a datetime for the current snapshot until we # receive a message. @@ -193,19 +162,11 @@ class AlgorithmSimulator(object): record.extra['algo_dt'] = self.snapshot_dt self.processor = Processor(inject_algo_dt) - def set_transact(self, transact): - """ - Set the method that will be called to create a - transaction from open orders and trade events. - """ - self.order_book.transact = transact - def order(self, sid, amount): """ Closure to pass into the user's algo to allow placing orders into the transaction simulator's dict of open orders. """ - assert sid in self.sids, "Order on invalid sid: %i" % sid order = ndict({ 'dt': self.simulation_dt, 'sid': sid, @@ -233,12 +194,6 @@ class AlgorithmSimulator(object): """ Main generator work loop. """ - # Call user's initialize method with a timeout (only if - # initialize wasn't called already). - if not getattr(self.algo, 'initialized', False): - with Timeout(INIT_TIMEOUT, message="Call to initialize timed out"): - self.algo.initialize() - # inject the current algo # snapshot time to any log record generated. with self.processor.threadbound(): @@ -298,7 +253,7 @@ class AlgorithmSimulator(object): Update the universe with new event information. """ # Update our portfolio. - self.universe.portfolio = event.portfolio + self.algo.set_portfolio(event.portfolio) # Update our knowledge of this event's sid for field in event.keys(): @@ -312,10 +267,8 @@ 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) + 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 c9e455a5..f1c99293 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -348,16 +348,16 @@ class BatchTransform(EventWindow): refresh_period=None, market_aware=True, delta=None, - days=None, - sids=None): - super(BatchTransform, self).__init__( - market_aware, days=days, delta=delta) + days=None): + + super(BatchTransform, self).__init__(market_aware, + days=days, delta=delta) + 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 @@ -373,15 +373,20 @@ class BatchTransform(EventWindow): handle_data method. """ # extract dates - dts = [data[sid].datetime for sid in self.sids] + #dts = [data[sid].datetime for sid in self.sids] + dts = [event.datetime for event in data.itervalues()] # 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) + # couple of seconds shouldn't matter. We don't add it to + # the data parameter, because it would mix dt with the + # sid keys. + event = ndict() + event.dt = max(dts) + event.data = data # append data frame to window. update() will call handle_add() and # handle_remove() appropriately - self.update(data) + self.update(event) # return newly computed or cached value return self.get_transform_value(*args, **kwargs) @@ -403,17 +408,23 @@ class BatchTransform(EventWindow): # # This Panel data structure ultimately gets passed to the # user-overloaded get_value() method. + # + # self.ticks contains ndicts with data, dt keys. + # event parameter is an ndict with data, dt keys. fields = {} for field_name in ['price', 'volume']: + sids = self.ticks[0].data.keys() # Skip non-existant fields - if field_name not in self.ticks[0][self.sids[0]]: + if field_name not in self.ticks[0].data[sids[0]]: continue values_per_sid = {} - for sid in self.sids: + + for sid in sids: values_per_sid[sid] = pd.Series( - {tick[sid].dt: tick[sid][field_name] - for tick in self.ticks}) + {tick.data[sid].dt: tick.data[sid][field_name] + for tick in self.ticks} + ) # concatenate different sids into one df fields[field_name] = pd.DataFrame.from_dict(values_per_sid) diff --git a/zipline/lines.py b/zipline/lines.py deleted file mode 100644 index f7b2baf4..00000000 --- a/zipline/lines.py +++ /dev/null @@ -1,114 +0,0 @@ -""" -Ziplines are composed of multiple components connected by asynchronous -messaging. All ziplines follow a general topology of parallel sources, -datetimestamp serialization, parallel transformations, and finally sinks. -Furthermore, many ziplines have common needs. For example, all trade -simulations require a -:py:class:`~zipline.finance.trading.TradeSimulationClient`. - -To establish best practices and minimize code replication, the lines module -provides complete zipline topologies. You can extend any zipline without -the need to extend the class. Simply instantiate any additional components -that you would like included in the zipline, and add them to the zipline -before invoking simulate. - - - Here is a diagram of the SimulatedTrading zipline: - - - +----------------------+ +------------------------+ - | Trade History | | (DataSource added | - | | | via add_source) | - | | | | - +--------------------+-+ +-+----------------------+ - | | - | | - v v - +---------+ - | Feed | (ensures events are serialized - +-+------++ in chronological order) - | | - | | - v v - +----------------------+ +----------------------+ - | (Transforms added | | (Transforms added | - | via add_transform) | | via add_transform) | - +-------------------+--+ +-+--------------------+ - | | - | | - v v - +------------+ - | Merge | (combines original event and - +------+-----+ transforms into one vector) - | - | - V - +---------------+ +--------------------------------+ - | Risk and Perf | | | - | Tracker | | TradingSimulationClient | - +---------------+ | tracks performance and | - ^ Trades and | provides API to algorithm. | - | simulated | | - | transactions +--+------------------+----------+ - | | ^ | - +---------------------+ | orders | frames - | | - | v - +---------------------------------+ - | Algorithm added via | - | __init__. | - +---------------------------------+ -""" -from zipline.gens.composites import ( - date_sorted_sources, - sequential_transforms -) -from zipline.gens.tradesimulation import TradeSimulationClient as tsc - -from logbook import Logger - -log = Logger('Lines') - - -class SimulatedTrading(object): - - def __init__(self, - sources, - transforms, - algorithm, - environment, - sim_method=None): - """ - @sources - an iterable of iterables - These iterables must yield ndicts that contain: - - type :: a ziplines.protocol.DATASOURCE_TYPE - - dt :: a milliseconds since epoch timestamp in UTC - - @transforms - An iterable of instances of StatefulTransform. - - @algorithm - An object that implements: - `def initialize(self)` - `def handle_data(self, data)` - `def get_sid_filter(self)` - `def set_logger(self, logger)` - `def set_order(self, order_callable)` - - @environment - An instance of finance.trading.TradingEnvironment - - @slippage - an object with a simulate method that takes a - trade event and returns a transaction - """ - - self.date_sorted = date_sorted_sources(*sources) - self.transforms = transforms - # Formerly merged_transforms. - self.with_tnfms = sequential_transforms(self.date_sorted, - *self.transforms) - self.trading_client = tsc(algorithm, environment, sim_method) - self.gen = self.trading_client.simulate(self.with_tnfms) - - def __iter__(self): - return self - - def next(self): - return self.gen.next() diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index 5d79499b..f855a2c5 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -67,42 +67,28 @@ The algorithm must expose methods: and trade events. """ +from zipline.algorithm import TradingAlgorithm +from zipline.finance.slippage import FixedSlippage -class TestAlgorithm(): +class TestAlgorithm(TradingAlgorithm): """ This algorithm will send a specified number of orders, to allow unit tests to verify the orders sent/received, transactions created, and positions at the close of a simulation. """ - def __init__(self, sid, amount, order_count, sid_filter=None): + def initialize(self, sid, amount, order_count, sid_filter=None): self.count = order_count self.sid = sid self.amount = amount self.incr = 0 - self.done = False - self.order = None - self.frame_count = 0 - self.portfolio = None if sid_filter: self.sid_filter = sid_filter else: self.sid_filter = [self.sid] - def initialize(self): - pass - - def set_order(self, order_callable): - self.order = order_callable - - def set_logger(self, logger): - pass - - def set_portfolio(self, portfolio): - self.portfolio = portfolio - def handle_data(self, data): self.frame_count += 1 #place an order for 100 shares of sid @@ -110,40 +96,18 @@ class TestAlgorithm(): self.order(self.sid, self.amount) self.incr += 1 - def get_sid_filter(self): - return self.sid_filter - def set_transact_setter(self, txn_sim_callable): - pass - - -class HeavyBuyAlgorithm(): +class HeavyBuyAlgorithm(TradingAlgorithm): """ This algorithm will send a specified number of orders, to allow unit tests to verify the orders sent/received, transactions created, and positions at the close of a simulation. """ - def __init__(self, sid, amount): + def initialize(self, sid, amount): self.sid = sid self.amount = amount self.incr = 0 - self.done = False - self.order = None - self.frame_count = 0 - self.portfolio = None - - def initialize(self): - pass - - def set_order(self, order_callable): - self.order = order_callable - - def set_logger(self, logger): - pass - - def set_portfolio(self, portfolio): - self.portfolio = portfolio def handle_data(self, data): self.frame_count += 1 @@ -151,33 +115,11 @@ class HeavyBuyAlgorithm(): self.order(self.sid, self.amount) self.incr += 1 - def get_sid_filter(self): - return [self.sid] - def set_transact_setter(self, txn_sim_callable): - pass - - -class NoopAlgorithm(object): +class NoopAlgorithm(TradingAlgorithm): """ Dolce fa niente. """ - - def initialize(self): - pass - - def set_order(self, order_callable): - pass - - def set_logger(self, logger): - pass - - def set_portfolio(self, portfolio): - pass - - def handle_data(self, data): - pass - def get_sid_filter(self): return [] @@ -185,17 +127,17 @@ class NoopAlgorithm(object): pass -class ExceptionAlgorithm(object): +class ExceptionAlgorithm(TradingAlgorithm): """ Throw an exception from the method name specified in the constructor. """ - def __init__(self, throw_from, sid): + def initialize(self, throw_from, sid): + self.throw_from = throw_from self.sid = sid - def initialize(self): if self.throw_from == "initialize": raise Exception("Algo exception in initialize") else: @@ -207,9 +149,6 @@ class ExceptionAlgorithm(object): else: pass - def set_logger(self, logger): - pass - def set_portfolio(self, portfolio): if self.throw_from == "set_portfolio": raise Exception("Algo exception in set_portfolio") @@ -232,81 +171,23 @@ class ExceptionAlgorithm(object): pass -class DivByZeroAlgorithm(): +class DivByZeroAlgorithm(TradingAlgorithm): - def __init__(self, sid): + def initialize(self, sid): self.sid = sid self.incr = 0 - def initialize(self): - pass - - def set_order(self, order_callable): - pass - - def set_logger(self, logger): - pass - - def set_portfolio(self, portfolio): - pass - def handle_data(self, data): self.incr += 1 if self.incr > 4: 5 / 0 pass - def get_sid_filter(self): - return [self.sid] - def set_transact_setter(self, txn_sim_callable): - pass +class TooMuchProcessingAlgorithm(TradingAlgorithm): - -class InitializeTimeoutAlgorithm(): - def __init__(self, sid): + def initialize(self, sid): self.sid = sid - self.incr = 0 - - def initialize(self): - import time - from zipline.gens.tradesimulation import INIT_TIMEOUT - time.sleep(INIT_TIMEOUT + 1000) - - def set_order(self, order_callable): - pass - - def set_logger(self, logger): - pass - - def set_portfolio(self, portfolio): - pass - - def handle_data(self, data): - pass - - def get_sid_filter(self): - return [self.sid] - - def set_transact_setter(self, txn_sim_callable): - pass - - -class TooMuchProcessingAlgorithm(): - def __init__(self, sid): - self.sid = sid - - def initialize(self): - pass - - def set_order(self, order_callable): - pass - - def set_logger(self, logger): - pass - - def set_portfolio(self, portfolio): - pass def handle_data(self, data): # Unless we're running on some sort of @@ -314,100 +195,19 @@ class TooMuchProcessingAlgorithm(): for i in xrange(1000000000): self.foo = i - def get_sid_filter(self): - return [self.sid] - def set_transact_setter(self, txn_sim_callable): - pass +class TimeoutAlgorithm(TradingAlgorithm): - -class TimeoutAlgorithm(): - - def __init__(self, sid): + def initialize(self, sid): self.sid = sid self.incr = 0 - def initialize(self): - pass - - def set_order(self, order_callable): - pass - - def set_logger(self, logger): - pass - - def set_portfolio(self, portfolio): - pass - def handle_data(self, data): if self.incr > 4: import time time.sleep(100) pass - def get_sid_filter(self): - return [self.sid] - - def set_transact_setter(self, txn_sim_callable): - pass - - -class TestPrintAlgorithm(): - - def __init__(self, sid): - self.sid = sid - - def initialize(self): - print "Initializing..." - - def set_order(self, order_callable): - pass - - def set_logger(self, logger): - pass - - def set_portfolio(self, portfolio): - pass - - def handle_data(self, data): - print "Handling Data..." - pass - - def get_sid_filter(self): - return [self.sid] - - def set_transact_setter(self, txn_sim_callable): - pass - - -class TestLoggingAlgorithm(): - - def __init__(self, sid): - self.log = None - self.sid = sid - - def initialize(self): - self.log.info("Initializing...") - - def set_order(self, order_callable): - pass - - def set_logger(self, logger): - self.log = logger - - def set_portfolio(self, portfolio): - pass - - def handle_data(self, data): - self.log.info("Handling Data...") - - def get_sid_filter(self): - return [self.sid] - - def set_transact_setter(self, txn_sim_callable): - pass - - from datetime import timedelta from zipline.algorithm import TradingAlgorithm from zipline.gens.transform import BatchTransform, batch_transform @@ -415,11 +215,13 @@ from zipline.gens.mavg import MovingAverage class TestRegisterTransformAlgorithm(TradingAlgorithm): - def initialize(self): + def initialize(self, *args, **kwargs): self.add_transform(MovingAverage, 'mavg', ['price'], market_aware=True, days=2) + self.set_slippage(FixedSlippage()) + def handle_data(self, data): pass @@ -454,26 +256,25 @@ class BatchTransformAlgorithm(TradingAlgorithm): self.kwargs = kwargs self.return_price_class = ReturnPriceBatchTransform( - sids=self.sids, market_aware=False, refresh_period=2, delta=timedelta(days=self.days) ) self.return_price_decorator = return_price_batch_decorator( - sids=self.sids, market_aware=False, refresh_period=2, delta=timedelta(days=self.days) ) self.return_args_batch = return_args_batch_decorator( - sids=self.sids, market_aware=False, refresh_period=2, delta=timedelta(days=self.days) ) + self.set_slippage(FixedSlippage()) + def handle_data(self, data): self.history_return_price_class.append( self.return_price_class.handle_data(data)) diff --git a/zipline/utils/date_utils.py b/zipline/utils/date_utils.py index 50545b0c..71ba01d0 100644 --- a/zipline/utils/date_utils.py +++ b/zipline/utils/date_utils.py @@ -41,6 +41,53 @@ def parse_iso8061(date_string): dt = dt.replace(tzinfo=pytz.utc) return dt + +# quarter utilities +# --------------------- +def get_quarter(dt): + """ + convert the given datetime to an integer representing + the number of calendar quarters since 0. + """ + quarters = dt.year * 4 + month = dt.month + if month <= 3: + return quarters + 1 + elif month <= 6: + return quarters + 2 + elif month <= 9: + return quarters + 3 + else: + return quarters + 4 + + +def dates_of_quarter(quarter_num): + year = quarter_num / 4 + quarter = quarter_num % 4 + if quarter == 0: + quarter = 4 + + if quarter == 1: + start = datetime(year, 1, 1, 0, 0, tzinfo=pytz.utc) + end = datetime(year, 3, 31, 23, 59, tzinfo=pytz.utc) + return start, end + + elif quarter == 2: + start = datetime(year, 4, 1, 0, 0, tzinfo=pytz.utc) + end = datetime(year, 6, 30, 23, 59, tzinfo=pytz.utc) + return start, end + + elif quarter == 3: + start = datetime(year, 7, 1, 0, 0, tzinfo=pytz.utc) + end = datetime(year, 9, 30, 23, 59, tzinfo=pytz.utc) + return start, end + + elif quarter == 4: + start = datetime(year, 10, 1, 0, 0, tzinfo=pytz.utc) + end = datetime(year, 12, 31, 23, 59, tzinfo=pytz.utc) + return start, end + + # Epoch utilities # --------------------- UNIX_EPOCH = datetime(1970, 1, 1, 0, 0, tzinfo=pytz.utc) diff --git a/zipline/utils/protocol_utils.py b/zipline/utils/protocol_utils.py index f1c77102..bb836b24 100644 --- a/zipline/utils/protocol_utils.py +++ b/zipline/utils/protocol_utils.py @@ -45,8 +45,11 @@ class ndict(MutableMapping): cls = None __slots__ = ['cls', '__internal'] - def __init__(self, dct=None): - self.__internal = dict() + def __init__(self, dct=None, internal=None): + if internal is not None: + self.__internal = internal + else: + self.__internal = dict() if not ndict.cls: ndict.cls = frozenset(dir(self)) diff --git a/zipline/utils/simfactory.py b/zipline/utils/simfactory.py index 7cf18130..025628eb 100644 --- a/zipline/utils/simfactory.py +++ b/zipline/utils/simfactory.py @@ -1,36 +1,33 @@ import zipline.utils.factory as factory from zipline.test_algorithms import TestAlgorithm -from zipline.lines import SimulatedTrading -from zipline.finance.slippage import FixedSlippage, transact_partial -from zipline.finance.commission import PerShare 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: - - environment - a \ - :py:class:`zipline.finance.trading.TradingEnvironment` - - sid - an integer, which will be used as the security ID. - - order_count - the number of orders the test algo will place, - defaults to 100 - - order_amount - the number of shares per order, defaults to 100 - - trade_count - the number of trades to simulate, defaults to 101 - to ensure all orders are processed. - - algorithm - optional parameter providing an algorithm. defaults - to :py:class:`zipline.test.algorithms.TestAlgorithm` - - trade_source - optional parameter to specify trades, if present. - If not present :py:class:`zipline.sources.SpecificEquityTrades` - is the source, with daily frequency in trades. - - slippage: optional parameter that configures the - :py:class:`zipline.gens.tradingsimulation.TransactionSimulator`. Expects - an object with a simulate mehod, such as - :py:class:`zipline.gens.tradingsimulation.FixedSlippage`. - :py:mod:`zipline.finance.trading` - - transforms: optional parameter that provides a list - of StatefulTransform objects. - """ + - environment - a \ + :py:class:`zipline.finance.trading.TradingEnvironment` + - sid - an integer, which will be used as the security ID. + - order_count - the number of orders the test algo will place, + defaults to 100 + - order_amount - the number of shares per order, defaults to 100 + - trade_count - the number of trades to simulate, defaults to 101 + to ensure all orders are processed. + - algorithm - optional parameter providing an algorithm. defaults + to :py:class:`zipline.test.algorithms.TestAlgorithm` + - trade_source - optional parameter to specify trades, if present. + If not present :py:class:`zipline.sources.SpecificEquityTrades` + is the source, with daily frequency in trades. + - slippage: optional parameter that configures the + :py:class:`zipline.gens.tradingsimulation.TransactionSimulator`. + Expects an object with a simulate mehod, such as + :py:class:`zipline.gens.tradingsimulation.FixedSlippage`. + :py:mod:`zipline.finance.trading` + - transforms: optional parameter that provides a list + of StatefulTransform objects. + """ assert isinstance(config, dict) sid_list = config.get('sid_list') if not sid_list: @@ -64,12 +61,20 @@ def create_test_zipline(**config): # trade than order trade_count = 101 - slippage = config.get('slippage', FixedSlippage()) - commission = PerShare() - transact_method = transact_partial(slippage, commission) + #------------------- + # Create the Algo + #------------------- + if 'algorithm' in config: + test_algo = config['algorithm'] + else: + test_algo = TestAlgorithm( + sid, + order_amount, + order_count + ) #------------------- - # Trade Source + # Trade Source #------------------- if 'trade_source' in config: trade_source = config['trade_source'] @@ -81,34 +86,25 @@ def create_test_zipline(**config): concurrent=concurrent_trades ) + test_algo.set_sources([trade_source]) + #------------------- # Transforms #------------------- - transforms = config.get('transforms', []) + + transforms = config.get('transforms', None) + if transforms is not None: + test_algo.set_transforms(transforms) #------------------- - # Create the Algo - #------------------- - if 'algorithm' in config: - test_algo = config['algorithm'] - else: - test_algo = TestAlgorithm( - sid, - order_amount, - order_count - ) + # Slippage + # ------------------ + slippage = config.get('slippage', None) + if slippage is not None: + test_algo.set_slippage(slippage) - #------------------- - # Simulation - #------------------- - - sim = SimulatedTrading( - [trade_source], - transforms, - test_algo, - trading_environment, - transact_method - ) - #------------------- + # ------------------ + # generator/simulator + sim = test_algo.get_generator(trading_environment) return sim