diff --git a/tests/test_finance.py b/tests/test_finance.py index 87ac9858..4f796521 100644 --- a/tests/test_finance.py +++ b/tests/test_finance.py @@ -286,9 +286,11 @@ class FinanceTestCase(TestCase): for txn, order in blotter.process_trade(event): transactions.append(txn) - tracker.process_event(txn) - - tracker.process_event(event) + tracker.process_transaction(txn) + elif event.type == DATASOURCE_TYPE.BENCHMARK: + tracker.process_benchmark(event) + elif event.type == DATASOURCE_TYPE.TRADE: + tracker.process_trade(event) if complete_fill: self.assertEqual(len(transactions), len(order_list)) diff --git a/tests/test_perf_tracking.py b/tests/test_perf_tracking.py index 889eaa7e..9082b27a 100644 --- a/tests/test_perf_tracking.py +++ b/tests/test_perf_tracking.py @@ -201,17 +201,23 @@ def calculate_results(host, for txn in filter(lambda txn: txn.dt == date, txns): # Process txns for this date. - perf_tracker.process_event(txn) + perf_tracker.process_transaction(txn) for event in group: - perf_tracker.process_event(event) - if event.type == zp.DATASOURCE_TYPE.BENCHMARK: + if event.type == zp.DATASOURCE_TYPE.TRADE: + perf_tracker.process_trade(event) + elif event.type == zp.DATASOURCE_TYPE.DIVIDEND: + perf_tracker.process_dividend(event) + elif event.type == zp.DATASOURCE_TYPE.BENCHMARK: + perf_tracker.process_benchmark(event) bm_updated = True + elif event.type == zp.DATASOURCE_TYPE.COMMISSION: + perf_tracker.process_commission(event) for split in filter(lambda split: split.dt == date, splits): # Process splits for this date. - perf_tracker.process_event(split) + perf_tracker.process_split(split) if bm_updated: msg = perf_tracker.handle_market_close_daily() @@ -1770,7 +1776,14 @@ class TestPerformanceTracker(unittest.TestCase): for date, group in grouped_events: for event in group: - perf_tracker.process_event(event) + if event.type == zp.DATASOURCE_TYPE.TRADE: + perf_tracker.process_trade(event) + elif event.type == zp.DATASOURCE_TYPE.ORDER: + perf_tracker.process_order(event) + elif event.type == zp.DATASOURCE_TYPE.BENCHMARK: + perf_tracker.process_benchmark(event) + elif event.type == zp.DATASOURCE_TYPE.TRANSACTION: + perf_tracker.process_transaction(event) msg = perf_tracker.handle_market_close_daily() perf_messages.append(msg) @@ -1877,7 +1890,14 @@ class TestPerformanceTracker(unittest.TestCase): for date, group in grouped_events: tracker.set_date(date) for event in group: - tracker.process_event(event) + if event.type == zp.DATASOURCE_TYPE.TRADE: + tracker.process_trade(event) + elif event.type == zp.DATASOURCE_TYPE.BENCHMARK: + tracker.process_benchmark(event) + elif event.type == zp.DATASOURCE_TYPE.ORDER: + tracker.process_order(event) + elif event.type == zp.DATASOURCE_TYPE.TRANSACTION: + tracker.process_transaction(event) tracker.handle_minute_close(date) msg = tracker.to_dict() messages[date] = msg diff --git a/zipline/finance/blotter.py b/zipline/finance/blotter.py index 319c0949..18bdd173 100644 --- a/zipline/finance/blotter.py +++ b/zipline/finance/blotter.py @@ -190,9 +190,11 @@ class Blotter(object): for order in orders_to_modify: order.handle_split(split_event) + def process_benchmark(self, benchmark_event): + return + yield + def process_trade(self, trade_event): - if trade_event.type != zp.DATASOURCE_TYPE.TRADE: - return if trade_event.sid not in self.open_orders: return diff --git a/zipline/finance/performance/tracker.py b/zipline/finance/performance/tracker.py index 3c2c0c8b..58f2a696 100644 --- a/zipline/finance/performance/tracker.py +++ b/zipline/finance/performance/tracker.py @@ -66,7 +66,6 @@ import numpy as np import pandas as pd from pandas.tseries.tools import normalize_date -import zipline.protocol as zp import zipline.finance.risk as risk from zipline.finance import trading from . period import PerformancePeriod @@ -281,62 +280,56 @@ class PerformanceTracker(object): return _dict - def process_event(self, event): + def process_trade(self, event): + self.position_tracker.update_last_sale(event) - if event.type == zp.DATASOURCE_TYPE.TRADE: - # update last sale - self.position_tracker.update_last_sale(event) + def process_transaction(self, event): - elif event.type == zp.DATASOURCE_TYPE.TRANSACTION: - # Trade simulation always follows a transaction with the - # TRADE event that was used to simulate it, so we don't - # check for end of day rollover messages here. - self.txn_count += 1 - self.position_tracker.execute_transaction(event) + self.txn_count += 1 + self.position_tracker.execute_transaction(event) + for perf_period in self.perf_periods: + perf_period.handle_execution(event) + + def process_dividend(self, dividend): + + log.info("Ignoring DIVIDEND event.") + + def process_split(self, event): + leftover_cash = self.position_tracker.handle_split(event) + if leftover_cash > 0: for perf_period in self.perf_periods: - perf_period.handle_execution(event) + perf_period.handle_cash_payment(leftover_cash) - elif event.type == zp.DATASOURCE_TYPE.DIVIDEND: - log.info("Ignoring DIVIDEND event.") + def process_order(self, event): + for perf_period in self.perf_periods: + perf_period.record_order(event) - elif event.type == zp.DATASOURCE_TYPE.SPLIT: - leftover_cash = self.position_tracker.handle_split(event) - if leftover_cash > 0: - for perf_period in self.perf_periods: - perf_period.handle_cash_payment(leftover_cash) + def process_commission(self, event): - elif event.type == zp.DATASOURCE_TYPE.ORDER: - for perf_period in self.perf_periods: - perf_period.record_order(event) + self.position_tracker.handle_commission(event) + for perf_period in self.perf_periods: + perf_period.handle_commission(event) - elif event.type == zp.DATASOURCE_TYPE.COMMISSION: - self.position_tracker.handle_commission(event) - for perf_period in self.perf_periods: - perf_period.handle_commission(event) + def process_benchmark(self, event): + if self.sim_params.data_frequency == 'minute' and \ + self.sim_params.emission_rate == 'daily': + # Minute data benchmarks should have a timestamp of market + # close, so that calculations are triggered at the right time. + # However, risk module uses midnight as the 'day' + # marker for returns, so adjust back to midnight. + midnight = pd.tseries.tools.normalize_date(event.dt) + else: + midnight = event.dt - elif event.type == zp.DATASOURCE_TYPE.CUSTOM: - pass + if midnight not in self.all_benchmark_returns.index: + raise AssertionError( + ("Date %s not allocated in all_benchmark_returns. " + "Calendar seems to mismatch with benchmark. " + "Benchmark container is=%s" % + (midnight, + self.all_benchmark_returns.index))) - elif event.type == zp.DATASOURCE_TYPE.BENCHMARK: - if self.sim_params.data_frequency == 'minute' and \ - self.sim_params.emission_rate == 'daily': - # Minute data benchmarks should have a timestamp of market - # close, so that calculations are triggered at the right time. - # However, risk module uses midnight as the 'day' - # marker for returns, so adjust back to midnight. - midnight = pd.tseries.tools.normalize_date(event.dt) - else: - midnight = event.dt - - if midnight not in self.all_benchmark_returns.index: - raise AssertionError( - ("Date %s not allocated in all_benchmark_returns. " - "Calendar seems to mismatch with benchmark. " - "Benchmark container is=%s" % - (midnight, - self.all_benchmark_returns.index))) - - self.all_benchmark_returns[midnight] = event.returns + self.all_benchmark_returns[midnight] = event.returns def check_upcoming_dividends(self, midnight_of_date_that_just_ended): """ diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index a43dad48..29e78639 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -70,12 +70,6 @@ class AlgorithmSimulator(object): record.extra['algo_dt'] = self.simulation_dt self.processor = Processor(inject_algo_dt) - def _process_event(self, blotter_process_trade, perf_process_event, event): - for txn, order in blotter_process_trade(event): - perf_process_event(txn) - perf_process_event(order) - perf_process_event(event) - def transform(self, stream_in): """ Main generator work loop. @@ -104,10 +98,12 @@ class AlgorithmSimulator(object): if event.type == DATASOURCE_TYPE.SPLIT: self.algo.blotter.process_split(event) - elif event.type in (DATASOURCE_TYPE.TRADE, - DATASOURCE_TYPE.CUSTOM): + elif event.type == DATASOURCE_TYPE.TRADE: self.update_universe(event) - self.algo.perf_tracker.process_event(event) + self.algo.perf_tracker.process_trade(event) + elif event.type == DATASOURCE_TYPE.CUSTOM: + self.update_universe(event) + else: message = self._process_snapshot( date, @@ -200,18 +196,41 @@ class AlgorithmSimulator(object): # # Done here, to allow for perf_tracker or blotter to be swapped out # or changed in between snapshots. - perf_process_event = self.algo.perf_tracker.process_event + perf_process_trade = self.algo.perf_tracker.process_trade + perf_process_transaction = self.algo.perf_tracker.process_transaction + perf_process_order = self.algo.perf_tracker.process_order + perf_process_benchmark = self.algo.perf_tracker.process_benchmark + perf_process_split = self.algo.perf_tracker.process_split + perf_process_dividend = self.algo.perf_tracker.process_dividend + perf_process_commission = self.algo.perf_tracker.process_commission blotter_process_trade = self.algo.blotter.process_trade - process_event = self._process_event + blotter_process_benchmark = self.algo.blotter.process_benchmark for event in snapshot: if event.type == DATASOURCE_TYPE.TRADE: self.update_universe(event) any_trade_occurred = True + if instant_fill: + events_to_be_processed.append(event) + else: + for txn, order in blotter_process_trade(event): + if txn.type == DATASOURCE_TYPE.TRANSACTION: + perf_process_transaction(txn) + elif txn.type == DATASOURCE_TYPE.COMMISSION: + perf_process_commission(txn) + perf_process_order(order) + perf_process_trade(event) elif event.type == DATASOURCE_TYPE.BENCHMARK: benchmark_event_occurred = True + perf_process_benchmark(event) + for txn, order in blotter_process_benchmark(event): + if txn.type == DATASOURCE_TYPE.TRANSACTION: + perf_process_transaction(txn) + elif txn.type == DATASOURCE_TYPE.COMMISSION: + perf_process_commission(txn) + perf_process_order(order) elif event.type == DATASOURCE_TYPE.CUSTOM: self.update_universe(event) @@ -220,27 +239,30 @@ class AlgorithmSimulator(object): # process_split is not assigned to a variable since it is # called rarely compared to the other event processors. self.algo.blotter.process_split(event) + perf_process_split(event) + + elif event.type == DATASOURCE_TYPE.DIVIDEND: + perf_process_dividend(event) - if not instant_fill: - process_event(blotter_process_trade, - perf_process_event, - event) else: - events_to_be_processed.append(event) + raise log.warn("Unrecognized event=%s".format(event)) if any_trade_occurred: new_orders = self._call_handle_data() for order in new_orders: - perf_process_event(order) + perf_process_order(order) if instant_fill: # Now that handle_data has been called and orders have been placed, # process the event stream to fill user orders based on the events # from this snapshot. for event in events_to_be_processed: - process_event(blotter_process_trade, - perf_process_event, - event) + for txn, order in blotter_process_trade(event): + if txn is not None: + perf_process_transaction(txn) + if order is not None: + perf_process_order(order) + perf_process_trade(event) if benchmark_event_occurred: return self.get_message(dt)