wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-15 11:22:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor tradesimulation client to not use StatefulTransform unnecessarily

Browse Source
This commit is contained in:
scottsanderson
2012-08-21 19:55:40 -04:00
parent ccc80525bf
commit 1f78a07d30
5 changed files with 95 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/finance/performance.py
+26 -28
View File
@@ -133,7 +133,7 @@ import zipline.finance.risk as risk
log = logbook.Logger('Performance')
class PerformanceTracker(object):
UPDATER = True
"""
Tracks the performance of the zipline as it is running in
the simulator, relays this out to the Deluge broker and then
@@ -166,7 +166,6 @@ class PerformanceTracker(object):
self.event_count = 0
self.last_dict = None
self.exceeded_max_loss = False
self.no_more_updates = False
self.results_socket = None
self.results_addr = None
@@ -202,28 +201,30 @@ class PerformanceTracker(object):
for sid in sid_list:
self.cumulative_performance.positions[sid] = Position(sid)
self.todays_performance.positions[sid] = Position(sid)
def update(self, event):
if self.no_more_updates:
return zp.ndict({'dt':0})
elif event.dt == "DONE":
event.perf_message = self.handle_simulation_end()
del event['TRANSACTION']
self.no_more_updates = True
return event
elif self.exceeded_max_loss:
# in case of max_loss, signal to downstream
# generators that we are done.
event.dt = "DONE"
event.perf_message = self.handle_simulation_end()
del event['TRANSACTION']
self.no_more_updates = True
return event
else:
event.perf_message = self.process_event(event)
event.portfolio = self.get_portfolio()
del event['TRANSACTION']
return event
def transform(self, stream_in):
"""
Main generator work loop.
"""
for event in stream_in:
if event.dt == "DONE":
event.perf_message = self.handle_simulation_end()
del event['TRANSACTION']
yield event
elif self.exceeded_max_loss:
# in case of max_loss, signal to downstream
# generators that we are done.
event.dt = "DONE"
event.perf_message = self.handle_simulation_end()
del event['TRANSACTION']
yield event
# Cut off the rest of the stream.
yield StopIteration()
else:
event.perf_message = self.process_event(event)
event.portfolio = self.get_portfolio()
del event['TRANSACTION']
yield event
def get_portfolio(self):
return self.cumulative_performance.as_portfolio()
@@ -241,8 +242,6 @@ class PerformanceTracker(object):
Publish the performance results asynchronously to a
socket.
"""
#assert isinstance(results_addr, basestring), type(results_addr)
#self.results_addr = results_addr
self.results_socket = results_addr
def to_dict(self):
@@ -267,7 +266,7 @@ class PerformanceTracker(object):
message = None
if self.exceeded_max_loss:
return
return message
assert isinstance(event, zp.ndict)
self.event_count += 1
@@ -288,7 +287,6 @@ class PerformanceTracker(object):
self.cumulative_performance.calculate_performance()
self.todays_performance.calculate_performance()
return message
def handle_market_close(self):
zipline/finance/trading.py
+7 -1
View File
@@ -9,7 +9,6 @@ from zipline.protocol import SIMULATION_STYLE
log = logbook.Logger('Transaction Simulator')
class TransactionSimulator(object):
UPDATER = True
def __init__(self, sid_filter, style=SIMULATION_STYLE.PARTIAL_VOLUME):
self.open_orders = {}
@@ -35,6 +34,13 @@ class TransactionSimulator(object):
order.filled = 0
self.open_orders[order.sid].append(order)
def transform(self, stream_in):
"""
Main generator work loop.
"""
for event in stream_in:
yield self.update(event)
def update(self, event):
event.TRANSACTION = None
# We only fill transactions on trade events.
zipline/gens/composites.py
+3 -1
View File
@@ -43,7 +43,9 @@ def merged_transforms(sorted_stream, *transforms):
"""
for transform in transforms:
assert isinstance(transform, StatefulTransform)
transform.set_copying()
transform.merged = True
transform.sequential = False
# Generate expected hashes for each transform
namestrings = [tnfm.get_hash() for tnfm in transforms]
zipline/gens/tradesimulation.py
+27 -54
View File
@@ -61,10 +61,16 @@ class TradeSimulationClient(object):
self.sids = algo.get_sid_filter()
self.environment = environment
self.style = sim_style
self.algo_sim = None
self.warmup_start = self.environment.prior_day_open
self.ordering_client = TransactionSimulator(self.sids, style=self.style)
self.perf_tracker = PerformanceTracker(self.environment, self.sids)
self.algo_start = self.environment.first_open
self.algo_sim = AlgorithmSimulator(
self.ordering_client,
self.algo,
self.algo_start
)
def get_hash(self):
"""
@@ -79,56 +85,36 @@ class TradeSimulationClient(object):
"""
# Simulate filling any open orders made by the previous run of
# the user's algorithm. Sets the txn field to true on any
# the user's algorithm. Sets the TRANSACTION field to true on any
# event that results in a filled order.
ordering_client = StatefulTransform(
TransactionSimulator,
self.sids,
style = self.style
)
with_filled_orders = ordering_client.transform(stream_in)
with_filled_orders = self.ordering_client.transform(stream_in)
# Pipe the events with transactions to perf. This will remove
# the txn field added by TransactionSimulator and replace it
# with a portfolio object to be passed to the user's
# the TRANSACTION field added by TransactionSimulator and replace it
# with a portfolio field to be passed to the user's
# algorithm. Also adds a perf_message field which is usually
# none, but contains an update message once per day.
perf_tracker = StatefulTransform(
PerformanceTracker,
self.environment,
self.sids
)
with_portfolio = perf_tracker.transform(with_filled_orders)
with_portfolio = self.perf_tracker.transform(with_filled_orders)
# Pass the messages from perf along with the trading client's
# state into the algorithm for simulation. We provide a
# pointer to the ordering client's internal state so that the
# algorithm can place new orders into the client's order book.
self.algo_sim = AlgorithmSimulator(
with_portfolio,
ordering_client.state,
self.algo,
self.algo_start
)
# Pass the messages from perf to the user's algorithm for simulation.
# Events are batched by dt so that the algo handles all events for a
# given timestamp at one one go.
performance_messages = self.algo_sim.transform(with_portfolio)
# The algorithm will yield a daily_results message (as
# calculated by the performance tracker) at the end of each
# day. It will also yield a risk report at the end of the
# simulation.
for message in self.algo_sim:
for message in performance_messages:
yield message
class AlgorithmSimulator(object):
def __init__(self,
stream_in,
order_book,
algo,
algo_start):
self.stream_in = stream_in
# ==========
# Algo Setup
# ==========
@@ -168,7 +154,6 @@ class AlgorithmSimulator(object):
# 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
@@ -188,22 +173,10 @@ class AlgorithmSimulator(object):
record.extra['algo_dt'] = self.snapshot_dt
self.processor = Processor(inject_algo_dt)
# This is a class, which is instantiated later
# in run_algorithm. The class provides a generator.
# Single_use generator that uses the @contextmanager decorator
# to monkey patch sys.stdout with a logbook interface.
self.stdout_capture = stdout_only_pipe
self.__generator = None
def __iter__(self):
return self
def next(self):
if self.__generator:
return self.__generator.next()
else:
self.__generator = self._gen()
return self.__generator.next()
def order(self, sid, amount):
"""
Closure to pass into the user's algo to allow placing orders
@@ -232,10 +205,10 @@ class AlgorithmSimulator(object):
# simulator so that it can fill the placed order when it
# receives its next message.
self.order_book.place_order(order)
def _gen(self):
def transform(self, stream_in):
"""
Internal generator work loop.
Main generator work loop.
"""
# Capture any output of this generator to stdout and pipe it
# to a logbook interface. Also inject the current algo
@@ -248,7 +221,7 @@ class AlgorithmSimulator(object):
# Group together events with the same dt field. This depends on the
# events already being sorted.
for date, snapshot in groupby(self.stream_in, lambda e: e.dt):
for date, snapshot in groupby(stream_in, lambda e: e.dt):
# Set the simulation date to be the first event we see.
# This should only occur once, at the start of the test.
@@ -259,7 +232,7 @@ class AlgorithmSimulator(object):
if date == 'DONE':
for event in snapshot:
yield event.perf_message
break
raise StopIteration()
# We're still in the warmup period. Use the event to
# update our universe, but don't yield any perf messages,
zipline/gens/transform.py
+32 -66
View File
@@ -19,7 +19,7 @@ from zipline.gens.utils import assert_sort_unframe_protocol, \
log = logbook.Logger('Transform')
class Passthrough(object):
FORWARDER = True
PASSTHROUGH = True
"""
Trivial class for forwarding events.
"""
@@ -29,23 +29,6 @@ class Passthrough(object):
def update(self, event):
pass
# Deprecated
def functional_transform(stream_in, func, *args, **kwargs):
"""
Generic transform generator that takes each message from an in-stream
and yields the output of a function on that message. Not sure how
useful this will be in reality, but good for testing.
"""
assert isinstance(func, types.FunctionType), \
"Functional"
namestring = func.__name__ + hash_args(*args, **kwargs)
for message in stream_in:
assert_sort_unframe_protocol(message)
out_value = func(message, *args, **kwargs)
assert_transform_protocol(out_value)
yield(namestring, out_value)
class StatefulTransform(object):
"""
Generic transform generator that takes each message from an
@@ -61,18 +44,15 @@ class StatefulTransform(object):
assert tnfm_class.__dict__.has_key('update'), \
"Stateful transform requires the class to have an update method"
self.forward_all = tnfm_class.__dict__.get('FORWARDER', False)
self.update_in_place = tnfm_class.__dict__.get('UPDATER', False)
self.append_value = tnfm_class.__dict__.get('APPENDER', False)
# You only one special behavior mode can be set.
assert sum(map(int, [self.forward_all,
self.update_in_place,
self.append_value])) <= 1
# Flag set inside the Passthrough transform class to signify special
# behavior if we are being fed to merged_transforms.
self.passthrough = tnfm_class.__dict__.get('PASSTHROUGH', False)
self.sequential = True
self.merged = False
# Create an instance of our transform class.
self.state = tnfm_class(*args, **kwargs)
self._copying = False
# Create the string associated with this generator's output.
self.namestring = tnfm_class.__name__ + hash_args(*args, **kwargs)
@@ -81,9 +61,6 @@ class StatefulTransform(object):
def get_hash(self):
return self.namestring
def set_copyting(self):
self._copying = True
def transform(self, stream_in):
return self._gen(stream_in)
@@ -101,59 +78,48 @@ class StatefulTransform(object):
assert_sort_unframe_protocol(message)
# Copying flag is used by merged_transforms to ensure
# This flag is set by by merged_transforms to ensure
# isolation of messages.
if self._copying:
if self.merged:
message = deepcopy(message)
# Same shared pointer issue here as above.
tnfm_value = self.state.update(message)
# FORWARDER flag means we want to keep all original
# PASSTHROUGH flag means we want to keep all original
# values, plus append tnfm_id and tnfm_value. Used for
# preserving the original event fields when our output
# will be fed into a merge. Currently only Passthrough
# uses this flag.
if self.forward_all:
if self.passthrough and self.merged:
out_message = message
out_message.tnfm_id = self.namestring
out_message.tnfm_value = tnfm_value
yield out_message
# UPDATER flag should be used for transforms that
# side-effectfully modify the event they are passed.
# Updated messages are passed along exactly as they are
# returned to use by our state class. Useful for chaining
# specific transforms that won't be fed to a merge. (See
# the implementation of TradeSimulationClient for example
# usage of this flag with PerformanceTracker and
# TransactionSimulator.
elif self.update_in_place:
yield tnfm_value
# APPENDER flag should be used to add a single new
# key-value pair to the event. The new key is this
# transform's namestring, and it's value is the value
# returned by state.update(event). This is almost
# identical to the behavior of FORWARDER, except we
# compress the two calculated values (tnfm_id, and
# tnfm_value) into a single field. This mode is used by
# the sequential_transforms composite.
elif self.append_value:
out_message = message
out_message[self.namestring] = tnfm_value
yield out_message
# If no flags are set, we create a new message containing
# just the tnfm_id, the event's datetime, and the
# calculated tnfm_value. This is the default behavior for
# a transform being fed into a merge.
else:
# If the merged flag is set, we create a new message
# containing just the tnfm_id, the event's datetime, and
# the calculated tnfm_value. This is the default behavior
# for a non-passthrough transform being fed into a merge.
elif self.merged:
out_message = ndict()
out_message.tnfm_id = self.namestring
out_message.tnfm_value = tnfm_value
out_message.dt = message.dt
yield out_message
# Sequential flag should be used to add a single new
# key-value pair to the event. The new key is this
# transform's namestring, and its value is the value
# returned by state.update(event). This is almost
# identical to the behavior of FORWARDER, except we
# compress the two calculated values (tnfm_id, and
# tnfm_value) into a single field. This mode is used by
# the sequential_transforms composite and is the default
# if no behavior is specified by the internal state class.
elif self.sequential:
out_message = message
out_message[self.namestring] = tnfm_value
yield out_message
log.info('Finished StatefulTransform [%s]' % self.get_hash())
class EventWindow: