wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-06 05:14:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

PERF: Use specific methods for processing events.

Browse Source 
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.
This commit is contained in:
Eddie Hebert
2015-05-06 23:06:20 -04:00
parent 72ab9e74dd
commit dba0a99a16
5 changed files with 118 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_finance.py
+5 -3
View File
@@ -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))
tests/test_perf_tracking.py
+26 -6
View File
@@ -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
zipline/finance/blotter.py
+4 -2
View File
@@ -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
zipline/finance/performance/tracker.py
+41 -48
View File
@@ -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):
"""
zipline/gens/tradesimulation.py
+42 -20
View File
@@ -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)