From dba0a99a1603dad87dccff67c78d364c6d914ff3 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 6 May 2015 23:06:20 -0400 Subject: [PATCH] PERF: Use specific methods for processing events. By having both the trade simulation main loop route events to "process" methods based on event type and the process methods also checking event type, there was some duplicated effort in doing that comparison many times. A particular case where this was noted in profiling was for the `process_event` function which was checking if the type was not a trade and returning early, when in a larger universe of stocks the value returned False 99% of the time. Instead provide separate process functions specific to each type, e.g. e.g. `process_trade` and `process_transaction` and route traffic to those functions in tradesimulation. For a universe of 160 stocks on both no-op algo and an algo that rebuys its universe every day, saw about a 10% increase locally. Also: - Add process_benchmark to blotter since internal subclass relies on logic on benchmark, this allows the internal process_trade to be a `pass`. - Add warning on unrecoginzed event types. --- tests/test_finance.py | 8 ++- tests/test_perf_tracking.py | 32 +++++++-- zipline/finance/blotter.py | 6 +- zipline/finance/performance/tracker.py | 89 ++++++++++++-------------- zipline/gens/tradesimulation.py | 62 ++++++++++++------ 5 files changed, 118 insertions(+), 79 deletions(-) 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)