diff --git a/zipline/core/component.py b/zipline/core/component.py index 78ee69c8..313e76e5 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -114,7 +114,6 @@ class Component(object): # Core Methods # ------------ - def loop_send(self): """ The main component loop. This is wrapped inside a diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 92c96f3a..4c96ecce 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -203,10 +203,15 @@ class PerformanceTracker(object): self.todays_performance.positions[sid] = Position(sid) def update(self, event): - event.perf_message = self.process_event(event) - event.portfolio = self.get_portfolio() - del event['TRANSACTION'] - return event + if event.dt == "DONE": + event.perf_message = self.handle_simulation_end() + del event['TRANSACTION'] + return event + else: + event.perf_message = self.process_event(event) + event.portfolio = self.get_portfolio() + del event['TRANSACTION'] + return event def get_portfolio(self): return self.cumulative_performance.as_portfolio() @@ -270,6 +275,7 @@ class PerformanceTracker(object): #calculate performance as of last trade self.cumulative_performance.calculate_performance() self.todays_performance.calculate_performance() + return message @@ -296,7 +302,8 @@ class PerformanceTracker(object): # calculate progress of test self.progress = self.day_count / self.total_days - #TODO TODO TODO!! + # Take a snapshot of our current peformance to return to the + # browser. daily_update = self.to_dict() if self.trading_environment.max_drawdown: @@ -356,12 +363,8 @@ class PerformanceTracker(object): exceeded_max_loss = self.exceeded_max_loss ) - if self.results_socket: - log.info("about to stream the risk report...") - risk_dict = self.risk_report.to_dict() - - msg = zp.RISK_FRAME(risk_dict) - self.results_socket.send(msg) + risk_dict = self.risk_report.to_dict() + return risk_dict class Position(object): diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index dd6345d4..7bd8c7c3 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -11,8 +11,8 @@ log = logbook.Logger('Transaction Simulator') class TransactionSimulator(object): UPDATER = True - def __init__(self, open_orders, style=SIMULATION_STYLE.PARTIAL_VOLUME): - self.open_orders = open_orders + def __init__(self, sid_filter, style=SIMULATION_STYLE.PARTIAL_VOLUME): + self.open_orders = {} self.txn_count = 0 self.trade_window = datetime.timedelta(seconds=30) self.orderTTL = datetime.timedelta(days=1) @@ -27,8 +27,15 @@ class TransactionSimulator(object): elif style == SIMULATION_STYLE.NOOP: self.apply_trade_to_open_orders = self.simulate_noop + for sid in sid_filter: + self.open_orders[sid] = [] + + def place_order(self, order): + self.open_orders[order.sid].append(order) + def update(self, event): event.TRANSACTION = None + # We only fill transactions on trade events. if event.type == zp.DATASOURCE_TYPE.TRADE: event.TRANSACTION = self.apply_trade_to_open_orders(event) return event diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index f7875816..2716ee32 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -3,7 +3,7 @@ from itertools import tee, starmap from collections import namedtuple from zipline.gens.tradegens import SpecificEquityTrades -from zipline.gens.utils import roundrobin, hash_args +from zipline.gens.utils import roundrobin, hash_args, done_message from zipline.gens.sort import date_sort from zipline.gens.merge import merge from zipline.gens.transform import StatefulTransform @@ -15,7 +15,7 @@ def date_sorted_sources(*sources): """ Takes an iterable of SortBundles, generating namestrings and initialized datasources for each before piping them into a date_sort. -n """ + """ for source in sources: assert iter(source), "Source %s not iterable" % source @@ -34,8 +34,7 @@ n """ return date_sort(stream_in, names) - -def merged_transforms(sorted_stream, bundles): +def merged_transforms(sorted_stream, *transforms): """ A generator that takes the expected output of a date_sort, pipes it through a given set of transforms, and runs the results throught a @@ -45,32 +44,34 @@ def merged_transforms(sorted_stream, bundles): tnfm_kwargs should be a list of dictionaries representing keyword arguments to each transform. """ + for transform in transforms: + assert isinstance(transform, StatefulTransform) + # Generate expected hashes for each transform - namestrings = [bundle.tnfm.__name__ + hash_args(*bundle.args, **bundle.kwargs) - for bundle in bundles] + namestrings = [tnfm.get_hash() for tnfm in transforms] # Create a copy of the stream for each transform. - split = tee(sorted_stream, len(bundles)) - # Package a stream copy with each bundle - tnfms_with_streams = zip(split, bundles) + split = tee(sorted_stream, len(transforms)) + + # Package a stream copy with each StatefulTransform instance. + bundles = zip(transforms, split) # Convert the copies into transform streams. - tnfms = [ - StatefulTransform( - stream_copy, - bundle.tnfm, - *bundle.args, - **bundle.kwargs - ) - for stream_copy, bundle in tnfms_with_streams - ] - tnfm_gens = [tnfm.gen() for tnfm in tnfms] + tnfm_gens = [tnfm.transform(stream) for tnfm, stream in bundles] - - # Roundrobin the outputs of our transforms to create a single flat stream. + # Roundrobin the outputs of our transforms to create a single flat + # stream. to_merge = roundrobin(tnfm_gens, namestrings) # Pipe the stream into merge. merged = merge(to_merge, namestrings) # Return the merged events. return merged + +def zipline(sources, transforms, endpoint): + assert isinstance(sources, (list, tuple)) + assert isinstance(transforms, (list, tuple)) + + + + diff --git a/zipline/gens/examples.py b/zipline/gens/examples.py index 3027eab7..a6a95f59 100644 --- a/zipline/gens/examples.py +++ b/zipline/gens/examples.py @@ -11,7 +11,7 @@ from zipline.gens.composites import SourceBundle, TransformBundle, \ date_sorted_sources, merged_transforms from zipline.gens.tradegens import SpecificEquityTrades from zipline.gens.transform import MovingAverage, Passthrough, StatefulTransform -from zipline.gens.tradesimulation import trade_simulation_client as tsc +from zipline.gens.tradesimulation import TradeSimulationClient as tsc import zipline.protocol as zp @@ -21,9 +21,10 @@ if __name__ == "__main__": #Set up source a. One minute between events. args_a = tuple() kwargs_a = { + 'count' : 325, 'sids' : [1,2,3], 'start' : datetime(2012,1,3,15, tzinfo = pytz.utc), - 'delta' : timedelta(minutes = 10), + 'delta' : timedelta(hours = 6), 'filter' : filter } source_a = SpecificEquityTrades(*args_a, **kwargs_a) @@ -31,29 +32,29 @@ if __name__ == "__main__": #Set up source b. Two minutes between events. args_b = tuple() kwargs_b = { + 'count' : 7500, 'sids' : [2,3,4], 'start' : datetime(2012,1,3,14, tzinfo = pytz.utc), - 'delta' : timedelta(minutes = 10), + 'delta' : timedelta(minutes = 5), 'filter' : filter } source_b = SpecificEquityTrades(*args_b, **kwargs_b) #Set up source c. Three minutes between events. - sort_out = date_sorted_sources(source_a, source_b) + sorted = date_sorted_sources(source_a, source_b) - passthrough = TransformBundle(Passthrough, (), {}) - mavg_price = TransformBundle(MovingAverage, (timedelta(minutes = 20), ['price']), {}) - tnfm_bundles = (passthrough, mavg_price) + passthrough = StatefulTransform(Passthrough) + mavg_price = StatefulTransform(MovingAverage, timedelta(minutes = 20), ['price']) - merge_out = merged_transforms(sort_out, tnfm_bundles) + merged = merged_transforms(sorted, passthrough, mavg_price) algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3]) environment = create_trading_environment(year = 2012) style = zp.SIMULATION_STYLE.FIXED_SLIPPAGE - client_out = tsc(merge_out, algo, environment, style) - for message in client_out: - pp(message) - sleep(1) + trading_client = tsc(algo, environment, style) + + for message in trading_client.simulate(merged): + pp(message) diff --git a/zipline/gens/merge.py b/zipline/gens/merge.py index dfd904d2..32035492 100644 --- a/zipline/gens/merge.py +++ b/zipline/gens/merge.py @@ -6,7 +6,7 @@ from collections import deque from zipline import ndict from zipline.gens.utils import hash_args, \ - assert_merge_protocol + assert_merge_protocol, done_message from itertools import repeat def merge(stream_in, tnfm_ids): @@ -51,6 +51,7 @@ def merge(stream_in, tnfm_ids): assert len(queue) == 1, "Bad queue in merge on exit: %s" % queue assert queue[0].dt == "DONE", \ "Bad last message in merge on exit: %s" % queue + yield done_message('Merge') def merge_one(sources): dict_primer = zip(sources.keys(), repeat(None)) diff --git a/zipline/gens/tradegens.py b/zipline/gens/tradegens.py index a8a691b5..b1a0ed96 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -84,7 +84,7 @@ class SpecificEquityTrades(object): self.generator = self.create_fresh_generator() def __iter__(self): - return self.generator + return self def next(self): return self.generator.next() diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index a4a576fe..972f44e8 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -9,7 +9,7 @@ from zipline.gens.transform import StatefulTransform from zipline.finance.trading import TransactionSimulator from zipline.finance.performance import PerformanceTracker -def trade_simulation_client(stream_in, algo, environment, sim_style): +class TradeSimulationClient(object): """ Generator that takes the expected output of a merge, a user algorithm, a trading environment, and a simulator style as @@ -42,61 +42,118 @@ def trade_simulation_client(stream_in, algo, environment, sim_style): overwritten so that only the most recent snapshot of the universe is sent to the algo. """ - - #============ - # Algo Setup - #============ - - # Initialize txn_sim's dictionary of orders here so that we can - # reference it from within the user's algorithm. - sids = algo.get_sid_filter() - open_orders = {} + def __init__(self, algo, environment, sim_style): - for sid in sids: - open_orders[sid] = [] + self.algo = algo + self.sids = algo.get_sid_filter() + self.environment = environment + self.style = sim_style + + def get_hash(self): + """ + There should only ever be one TSC in the system. + """ + return self.__class__.__name__ + hash_args() + + def simulate(self, stream_in): + """ + Main generator work loop. + """ + + # Simulate filling any open orders made by the previous run of + # the user's algorithm. Sets the txn 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) + + # 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 + # 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) + + # Pass the messages from perf along with the trading client's + # state into the algorithm for simulation. We provide the + # trading client so that the algorithm can place new orders + # into the client's order book. + algo_results = AlgorithmSimulator( + with_portfolio, + ordering_client.state, + self.algo, + ) + + # 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 algo_results: + yield message + + +class AlgorithmSimulator(object): - # Pipe the in stream into the transaction simulator. - # Creates a txn field on the event containing transaction - # information if we filled any pending orders on the event's sid. - # TRANSACTION is None if we didn't fill any orders. - with_txns = StatefulTransform( - stream_in, - TransactionSimulator, - open_orders, - style = sim_style - ) - - # 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 algorithm. Also adds - # a PERF_MESSAGE field which is usually none, but contains an update - # message once per day. - with_portfolio_and_perf_msg = StatefulTransform( - with_txns, - PerformanceTracker, - environment, - sids - ) - - # Batch the event stream by dt to be processed by the user's algo. - # Yields perf messages whenever it encounters them. - perf_messages = algo_simulator(with_portfolio_and_perf_msg, sids, algo, open_orders) - - for message in perf_messages: - yield message - - -def algo_simulator(stream_in, sids, algo, order_book): + def __init__(self, stream_in, order_book, algo): - simulation_dt = None + self.stream_in = stream_in - # Closure to pass into the user's algo to allow placing orders - # into the txn_sim's dict of open orders. - def order(sid, amount): - assert sid in sids, "Order on invalid sid: %i" % sid + # We extract the order book from the txn client so that + # the algo can place new orders. + self.order_book = order_book + + self.algo = algo + self.sids = algo.get_sid_filter() + + # Monkey patch the user algorithm to place orders in the + # TransactionSimulator's order book. + self.algo.set_order(self.order) + self.algo.set_logger(logbook.Logger("Algolog")) + + # Call the user's initialize method. + self.algo.initialize() + + # 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 + + # We don't have a datetime for the current snapshot until we + # receive a message. + self.simulation_dt = None + self.this_snapshot_dt = None + + 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 + into the txn_sim's dict of open orders. + """ + assert sid in self.sids, "Order on invalid sid: %i" % sid order = ndict({ - 'dt' : simulation_dt, + 'dt' : self.simulation_dt, 'sid' : sid, 'amount' : int(amount), 'filled' : 0 @@ -104,91 +161,107 @@ def algo_simulator(stream_in, sids, algo, order_book): # Tell the user if they try to buy 0 shares of something. if order.amount == 0: - log = "requested to trade zero shares of {sid}".format( + zero_message = "Requested to trade zero shares of {sid}".format( sid=event.sid ) - log.debug(log) + log.debug(zero_message) + # Don't bother placing orders for 0 shares. return - - order_book[sid].append(order) - - # Set the algo's order method. - algo.set_order(order) - # Provide a logbook logging interface to user code. - algo.set_logger(logbook.Logger("Algolog")) + # Add non-zero orders to the order book. + # !!!IMPORTANT SIDE-EFFECT!!! + # This modifies the internal state of the transaction + # simulator so that it can fill the placed order when it + # receives its next message. + self.order_book.place_order(order) - # Call user-defined initialize method before we process any - # events. - algo.initialize() - - universe = ndict() - for sid in sids: - universe[sid] = ndict() - universe.portfolio = None - this_snapshot_dt = None - - for event in stream_in: - # Yield any perf messages received to be relayed back to the browser. - if event.perf_message: - yield event.perf_message - del event['perf_message'] - - # This should only happen for the first event we run. - if simulation_dt == None: - simulation_dt = event.dt - - # If we are currently creating a new message and this update - # matches the message dt, update the state of the universe. - - if this_snapshot_dt != None: - - if event.dt == this_snapshot_dt: - update_universe(event, universe) - - # If we are constructing a snapshot and we hit a new dt, call - # handle_data and record how long it takes. - else: - start_tic = datetime.now() - algo.handle_data(universe) - stop_tic = datetime.now() - - # How long did you take? - delta = stop_tic - start_tic - - # Update the simulation time. - simulation_dt = this_snapshot_dt + delta + def _gen(self): + """ + Internal generator work loop. + """ + for event in self.stream_in: + # Yield any perf messages received to be relayed back to the browser. + if event.perf_message: + yield event.perf_message + del event['perf_message'] + if event.dt == "DONE": + break - # Update the universe with the new event. - update_universe(event, universe) + # This should only happen for the first event we run. + if self.simulation_dt == None: + self.simulation_dt = event.dt + + # ====================== + # Time Compression Logic + # ====================== + + if self.this_snapshot_dt != None: + self.update_current_snapshot(event) - # If the current event is later than the simulation - # time, update the universe and start constructing - # another snapshot. - if event.dt >= simulation_dt: - this_snapshot_dt = event.dt - else: - this_snapshot_dt = None - # We have been fastforwarding. Update the universe - # and check if we can start a new snapshot. + # The algorithm has been missing events because it took + # too long processing. Update the universe with data from + # this event, then check if enough time has passed that we + # can start a new snapshot. + else: + self.update_universe(event) + if event.dt >= self.simulation_dt: + self.this_snapshot_dt = event.dt + + def update_current_snapshot(self, event): + """ + Update our current snapshot of the universe. Call handle_data if + """ + # The new event matches our snapshot dt. Just update the + # universe and move on. + if event.dt == self.this_snapshot_dt: + self.update_universe(event) + + # The new event does not match our snapshot. else: - update_universe(event, universe) - if event.dt >= simulation_dt: - this_snapshot_dt = event.dt - + self.simulate_current_snapshot() + + # Once we've finished simulating the old snapshot, + # we can update the universe with the new event. + self.update_universe(event) + + # The current event is later than the simulation time, + # which means the algorithm finished quickly enough to + # receive the new event. Start a new snapshot with this + # event's dt. + if event.dt >= self.simulation_dt: + self.this_snapshot_dt = event.dt - + # The algorithm spent enough time processing that it + # missed the new event. Wait to start a new snapshot until + # the events catch up to the algo's simulated dt. + else: + self.this_snapshot_dt = None + + def simulate_current_snapshot(self): + """ + Run the user's algo against our current snapshot and update the algo's + simulated time. + """ + start_tic = datetime.now() + self.algo.handle_data(self.universe) + stop_tic = datetime.now() + + # How long did you take? + delta = stop_tic - start_tic + + # Update the simulation time. + self.simulation_dt = self.this_snapshot_dt + delta -def update_universe(event, universe): - - universe.portfolio = event.portfolio - del event['portfolio'] + def update_universe(self, event): + """ + Update the universe with new event information. + """ + # Update our portfolio. + self.universe.portfolio = event.portfolio - event_sid = event.sid - del event['sid'] - - for field in event.keys(): - universe[event_sid][field] = event[field] + # Update our knowledge of this event's sid + for field in event.keys(): + self.universe[event.sid][field] = event[field] diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index 4d453927..564284b5 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -42,13 +42,13 @@ def functional_transform(stream_in, func, *args, **kwargs): class StatefulTransform(object): """ Generic transform generator that takes each message from an - in-stream and passes it to a state class. For each call to + in-stream and passes it to a state object. For each call to update, the state class must produce a message to be fed downstream. Any transform class with the FORWARDER class variable set to true will forward all fields in the original message. Otherwise only dt, tnfm_id, and tnfm_value are forwarded. """ - def __init__(self, stream_in, tnfm_class, *args, **kwargs): + def __init__(self, tnfm_class, *args, **kwargs): assert isinstance(tnfm_class, (types.ObjectType, types.ClassType)), \ "Stateful transform requires a class." assert tnfm_class.__dict__.has_key('update'), \ @@ -56,26 +56,26 @@ class StatefulTransform(object): self.forward_all = tnfm_class.__dict__.get('FORWARDER', False) self.update_in_place = tnfm_class.__dict__.get('UPDATER', False) + + # You can't be both a forwarded and an updater. assert not all([self.forward_all, self.update_in_place]) - self.stream_in = stream_in - # Create an instance of our transform class. self.state = tnfm_class(*args, **kwargs) - # Generate the string associated with this generator's output. + # Create the string associated with this generator's output. self.namestring = tnfm_class.__name__ + hash_args(*args, **kwargs) - + def get_hash(self): return self.namestring - def __iter__(self): - return self.gen() - - def gen(self): + def transform(self, stream_in): + return self._gen(stream_in) + + def _gen(self, stream_in): # IMPORTANT: Messages may contain pointers that are shared with # other streams, so we only manipulate copies. - for message in self.stream_in: + for message in stream_in: assert_sort_unframe_protocol(message) message_copy = deepcopy(message) diff --git a/zipline/gens/utils.py b/zipline/gens/utils.py index bacd07bd..f979e063 100644 --- a/zipline/gens/utils.py +++ b/zipline/gens/utils.py @@ -23,7 +23,7 @@ def mock_done(id): "source_id" : id, 'tnfm_id' : id, 'tnfm_value': None, - 'type' : 0 + 'type' : DATASOURCE_TYPE.DONE }) done_message = mock_done