diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index d5c7d115..50326368 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -1,4 +1,4 @@ -import logbook +from logbook import Logger, Processor from datetime import datetime, timedelta from numbers import Integral @@ -9,6 +9,9 @@ from zipline.gens.transform import StatefulTransform from zipline.finance.trading import TransactionSimulator from zipline.finance.performance import PerformanceTracker from zipline.utils.log_utils import stdout_only_pipe +from zipline.gens.utils import hash_args + +log = Logger('Trade Simulation') class TradeSimulationClient(object): """ @@ -43,14 +46,15 @@ class TradeSimulationClient(object): overwritten so that only the most recent snapshot of the universe is sent to the algo. """ - + def __init__(self, algo, environment, sim_style): self.algo = algo self.sids = algo.get_sid_filter() self.environment = environment self.style = sim_style - + self.algo_sim = None + def get_hash(self): """ There should only ever be one TSC in the system. @@ -83,36 +87,36 @@ class TradeSimulationClient(object): 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( + self.algo_sim = AlgorithmSimulator( with_portfolio, ordering_client.state, - self.algo, + 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: + for message in self.algo_sim: yield message class AlgorithmSimulator(object): - - def __init__(self, stream_in, order_book, algo): - + + def __init__(self, stream_in, order_book, algo): + self.stream_in = stream_in - + # ========== # Algo Setup # ========== - # We extract the order book from the txn client so that + # We extract the order book from the txn client so that # the algo can place new orders. self.order_book = order_book @@ -122,10 +126,8 @@ class AlgorithmSimulator(object): # 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")) + self.algo.set_logger(Logger("AlgoLog")) - # Call the user's initialize method. - self.algo.initialize() # ============== # Snapshot Setup @@ -133,11 +135,11 @@ 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 - + # We don't have a datetime for the current snapshot until we # receive a message. self.simulation_dt = None @@ -146,22 +148,22 @@ class AlgorithmSimulator(object): # ============= # Logging Setup # ============= - + # Processor function for injecting the algo_dt into # user prints/logs. def inject_algo_dt(record): record.extra['algo_dt'] = self.this_snapshot_dt - self.processor = logbook.Processor(inject_algo_dt) + self.processor = Processor(inject_algo_dt) # This is a class, which is instantiated later # in run_algorithm. The class provides a generator. self.stdout_capture = stdout_only_pipe - + self.__generator = None def __iter__(self): return self - + def next(self): if self.__generator: return self.__generator.next() @@ -181,17 +183,17 @@ class AlgorithmSimulator(object): 'amount' : int(amount), 'filled' : 0 }) - + # Tell the user if they try to buy 0 shares of something. if order.amount == 0: zero_message = "Requested to trade zero shares of {sid}".format( - sid=event.sid + sid=order.sid ) log.debug(zero_message) # Don't bother placing orders for 0 shares. - return + return - # Add non-zero orders to the order book. + # 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 @@ -207,7 +209,9 @@ class AlgorithmSimulator(object): # snapshot time to any log record generated. with self.processor.threadbound(), self.stdout_capture(Logger('Print'),''): - + # Call the user's initialize method. + self.algo.initialize() + for event in self.stream_in: # Yield any perf messages received to be relayed back to # the browser. @@ -215,6 +219,12 @@ class AlgorithmSimulator(object): yield event.perf_message del event['perf_message'] if event.dt == "DONE": + if self.this_snapshot_dt: + # stop iteration happened + # mid-snapshot, so we have a universe + # snapshot that is not yet processed + # by the algorithm. + self.simulate_current_snapshot() break # This should only happen for the first event we run. @@ -232,67 +242,65 @@ class AlgorithmSimulator(object): # 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: + 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 + 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. + + # The new event does not match our snapshot. else: 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(self, event): """ Update the universe with new event information. """ # Update our portfolio. self.universe.portfolio = event.portfolio - + # Update our knowledge of this event's sid for field in event.keys(): self.universe[event.sid][field] = event[field] - - - - diff --git a/zipline/lines.py b/zipline/lines.py index 66e3d70f..9a0b49ef 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -103,7 +103,7 @@ class SimulatedTrading(object): self.date_sorted = date_sorted_sources(*sources) self.transforms = transforms # Formerly merged_transforms. - self.with_tnfms = sequential_transforms(self.date_sorted, *self.transforms) + self.with_tnfms = sequential_transforms(self.date_sorted, *self.transforms) self.trading_client = tsc(algorithm, environment, style) self.gen = self.trading_client.simulate(self.with_tnfms) self.results_uri = results_socket_uri @@ -137,8 +137,7 @@ class SimulatedTrading(object): setproctitle(self.sim_id) self.open() if self.zmq_out: - log_pipeline = NestedSetup([self.zmq_out,data_injector]) - with log_pipeline.threadbound(), self.stdout_capture(self.print_logger, ''): + with self.zmq_out.threadbound(): self.stream_results() # if no log socket, just run the algo normally else: @@ -148,13 +147,13 @@ class SimulatedTrading(object): assert self.results_socket, \ "Results socket must exist to stream results" try: - for event in self.gen: + for event in self.gen: if event.has_key('daily_perf'): msg = zp.PERF_FRAME(event) else: msg = zp.RISK_FRAME(event) self.results_socket.send(msg) - + self.signal_done() except Exception as exc: self.handle_exception(exc) @@ -207,7 +206,7 @@ class SimulatedTrading(object): ) self.results_socket.send(msg) - + except: log.exception("Exception while reporting simulation exception.") @@ -222,12 +221,16 @@ class SimulatedTrading(object): def setup_logging(self): assert self.results_socket + # The filter behavior is: matches are logged, mismatches + # are bubbled. If bubble is True, matches are also + # bubbled. Since we do not want user logs in our system + # logs, we set bubble to False. self.zmq_out = ZeroMQLogHandler( socket = self.results_socket, filter = lambda r, h: r.channel in ['Print', 'AlgoLog'], - bubble = True + bubble=False ) - + def join(self): if self.proc: self.proc.join() diff --git a/zipline/utils/log_utils.py b/zipline/utils/log_utils.py index 6bcf80f8..f9fbc57c 100644 --- a/zipline/utils/log_utils.py +++ b/zipline/utils/log_utils.py @@ -120,12 +120,6 @@ class ZeroMQLogHandler(Handler): #can't be serialized by JSON, so we need to convert to #unix epoch representation. - if record.time: - assert isinstance(record.time, datetime.datetime) - - time = record.time.replace(tzinfo = pytz.utc) - #logbook measures time in utc already, no need to convert. - record.time = EPOCH(time) #Do the same if algo_dt is a datetime object. if record.extra.has_key('algo_dt'): @@ -151,6 +145,14 @@ class ZeroMQLogHandler(Handler): data[field] = record.extra[field] else: data[field] = None + + if data['time']: + assert isinstance(data['time'], datetime.datetime) + + time = data['time'].replace(tzinfo = pytz.utc) + #logbook measures time in utc already, no need to convert. + data['time'] = EPOCH(time) + return data def emit(self, record): diff --git a/zipline/utils/test_utils.py b/zipline/utils/test_utils.py index 036ebe02..02ac4c69 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -61,7 +61,7 @@ def check(test, a, b, label=None): test.assertEqual(a, b, "mismatch on path: " + label) -def drain_zipline(test, zipline): +def drain_zipline(test, zipline, p_blocking=False): assert test.ctx, "method expects a valid zmq context" assert test.zipline_test_config, "method expects a valid test config" assert isinstance(test.zipline_test_config, dict) @@ -76,7 +76,7 @@ def drain_zipline(test, zipline): time.sleep(1) # start the simulation - zipline.simulate(blocking=False) + zipline.simulate(blocking=p_blocking) output, transaction_count = drain_receiver(test.receiver) # some processes will exit after the message stream is # finished. We block here to avoid collisions with subsequent