diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 98ac8daf..05a3a875 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -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): diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 2fa99d9b..4b971394 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -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. diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index 78bea152..cec8b448 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -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] diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 5d036495..83f077fe 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -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, diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index 612cd098..e10380b6 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -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: