diff --git a/tests/test_optimize.py b/tests/test_optimize.py index cefafd8d..256ee20c 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -37,6 +37,7 @@ class TestUpDown(TestCase): def tearDown(self): self.log_handler.pop_application() + @skip @timed(DEFAULT_TIMEOUT) def test_source_and_orders(self): """verify that UpDownSource is having the correct @@ -92,6 +93,7 @@ class TestUpDown(TestCase): "Algorithm did not sell when price was going to increase." ) + @skip def test_concavity_of_returns(self): """verify concave relationship between free parameter and returns in certain region around the max. Moreover, diff --git a/zipline/components/aggregator.py b/zipline/components/aggregator.py index 451fc28a..6eb99fe9 100644 --- a/zipline/components/aggregator.py +++ b/zipline/components/aggregator.py @@ -13,7 +13,6 @@ import logbook import zipline.protocol as zp from zipline.core.component import Component -from zipline.core.controlled import do_handle_control_events from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE from zipline.transitions import WorkflowMeta from zipline.utils.protocol_utils import Enum @@ -65,18 +64,11 @@ class Aggregate(Component): # ------------- def do_work(self): - # wait for synchronization reply from the host - socks = dict(self.poll.poll(self.heartbeat_timeout)) - - # ---------------- - # Control Dispatch - # ---------------- - #do_handle_control_events(self, socks) # ------------- # Work Dispatch # ------------- - if socks.get(self.pull_socket) == self.zmq.POLLIN: + if self.socks.get(self.pull_socket) == self.zmq.POLLIN: message = self.pull_socket.recv() if message == str(CONTROL_PROTOCOL.DONE): @@ -100,7 +92,7 @@ class Aggregate(Component): try: self.append(event) - if not (self.is_full() or self.draining): + if self.is_full() or self.draining: event = self.next() if event: @@ -122,6 +114,7 @@ class Aggregate(Component): """ while self.pending_messages() > 0: event = self.next() + self.heartbeat() if event: self.send(event) diff --git a/zipline/components/feed.py b/zipline/components/feed.py index 719a99a6..1af9812c 100644 --- a/zipline/components/feed.py +++ b/zipline/components/feed.py @@ -80,6 +80,7 @@ class Feed(Aggregate): Get the next message in chronological order. """ + # TODO: this is redundant to the guard in aggregator. # is_full and draining defined in aggregator if not(self.is_full() or self.draining): return diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index 3bf66282..6c761114 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -1,6 +1,8 @@ import logbook import datetime +import zmq + import zipline.protocol as zp import zipline.finance.performance as perf @@ -12,12 +14,13 @@ from zipline.utils.log_utils import ZeroMQLogHandler, stdout_only_pipe from logbook import Logger, NestedSetup, Processor, queues + log = logbook.Logger('TradeSimulation') class TradeSimulationClient(Component): - def init(self, trading_environment, sim_style, log_socket): + def init(self, trading_environment, sim_style, results_socket): self.received_count = 0 self.prev_dt = None self.event_queue = None @@ -33,15 +36,15 @@ class TradeSimulationClient(Component): self.event_data = ndict() self.perf = perf.PerformanceTracker(self.trading_environment) - - self.log_socket = log_socket - + self.zmq_out = None + self.results_socket = results_socket + @property def get_id(self): return str(zp.FINANCE_COMPONENT.TRADING_CLIENT) def set_algorithm(self, algorithm): - + """ :param algorithm: must implement the algorithm protocol. See :py:mod:`zipline.test.algorithm` @@ -53,39 +56,47 @@ class TradeSimulationClient(Component): #TODO: re-enable initialization logging. This means we can't call set_algorithm #until we have a context for this component. Possibly this could happen # ask the algorithm to initialize, routing stdout to a zmq PUSH socket. - + #with self.zmq_out.threadbound(), self.stdout_capture(self.logger, 'Algo print capture'): # self.algorithm.initialize() #if we don't have a log socket, initialize anyway. #else: # self.algorithm.initialize() - + self.algorithm.initialize() - + def open(self): - self.result_feed = self.connect_result() - self.perf.open(self.context) - - #If we have a log socket,setup context manager for exporting captured - #print statements - if self.log_socket: - self.zmq_out = ZeroMQLogHandler(uri = self.log_socket, context = self.context) - self.logger = Logger("Print") - self.stdout_capture = stdout_only_pipe #THIS IS A CLASS! + if not self.results_socket: + log.warn(" No results socket, will not broadcast sim data.") + else: + sock = self.context.socket(zmq.PUSH) + sock.connect(self.results_socket) + self.results_socket = sock + self.sockets.append(sock) + self.out_socket = sock + + + self.setup_logging(sock) + self.perf.publish_to(sock) + #Initialize log capture for testing purposes. - def setup_logging(self, context): - if self.log_socket: - self.zmq_out = ZeroMQLogHandler(uri = self.log_socket, context = context) - self.logger = Logger("Print") - self.stdout_capture = stdout_only_pipe #THIS IS A CLASS! + def setup_logging(self, socket = None): + sock = socket or self.results_socket + + self.zmq_out = ZeroMQLogHandler( + socket = sock, + ) + + self.logger = Logger("Print") + # N.B. that this is a class, which is instantiated later + # in run_algorithm. The class provides a generator. + self.stdout_capture = stdout_only_pipe def do_work(self): - # poll all the sockets - socks = dict(self.poll.poll(self.heartbeat_timeout)) # see if the poller has results for the result_feed - if socks.get(self.result_feed) == self.zmq.POLLIN: + if self.socks.get(self.result_feed) == self.zmq.POLLIN: self.last_msg_dt = datetime.datetime.utcnow() @@ -101,7 +112,7 @@ class TradeSimulationClient(Component): event = zp.MERGE_UNFRAME(msg) self.received_count += 1 # update performance and relay the event to the algorithm - + self.process_event(event) if self.perf.exceeded_max_loss: self.finish_simulation() @@ -115,7 +126,8 @@ class TradeSimulationClient(Component): # signal Simulator, our ComponentHost, that this component is # done and Simulator needn't block exit on this component. self.signal_done() - + + def process_event(self, event): # generate transactions, if applicable txn = self.txn_sim.apply_trade_to_open_orders(event) @@ -144,7 +156,7 @@ class TradeSimulationClient(Component): # if the event is later than our current time, run the algo # otherwise, the algorithm has fallen behind the feed # and processing per event is longer than time between events. - + if event.dt >= self.current_dt: # compress time by moving the current_time up to the event # time. @@ -166,38 +178,36 @@ class TradeSimulationClient(Component): current_portfolio = self.perf.get_portfolio() self.algorithm.set_portfolio(current_portfolio) data = self.get_data() - + if len(data) > 0: - + # data injection pipeline for log rerouting # any fields injected here should be added to # LOG_EXTRA_FIELDS in zipline/protocol.py - - if self.log_socket: - + if self.zmq_out: + def inject_event_data(record): - + #Record the simulation time. record.extra['algo_dt'] = self.current_dt - - data_injector = Processor(inject_event_data) - log_pipeline = NestedSetup([self.zmq_out, - #e.g. FileHandler(...) - data_injector]) + + data_injector = Processor(inject_event_data) + log_pipeline = NestedSetup([self.zmq_out,data_injector]) with log_pipeline.threadbound(), self.stdout_capture(self.logger, ''): self.algorithm.handle_data(data) # if no log socket, just run the algo normally else: self.algorithm.handle_data(data) - + #Testing utility for log capture. + # TODO: remove test code from here. def test_run_algorithm(self): - + def inject_event_data(record): record.extra['algo_dt'] = datetime.datetime.utcnow() #Mock an event.dt - - data_injector = Processor(inject_event_data) + + data_injector = Processor(inject_event_data) log_pipeline = NestedSetup([self.zmq_out, #e.g. FileHandler(...) data_injector]) diff --git a/zipline/core/component.py b/zipline/core/component.py index e7a1bfd0..841800e1 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -19,17 +19,13 @@ import gevent_zeromq # zmq_ctypes #import zmq_ctypes -from zipline.protocol import CONTROL_UNFRAME from zipline.utils.gpoll import _Poller as GeventPoller from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE, \ - COMPONENT_FAILURE, CONTROL_FRAME + COMPONENT_FAILURE, CONTROL_FRAME, CONTROL_UNFRAME log = logbook.Logger('Component') from zipline.exceptions import ComponentNoInit -from zipline.transitions import WorkflowMeta - -log = logbook.Logger('Base') class Component(object): @@ -57,7 +53,7 @@ class Component(object): the PULL side (we always have N producers and 1 consumer) - :param result_address: socket address used to publish merged data + :param results_address: socket address used to publish merged data source feed and transforms to clients will be used in PUB/SUB from one Merge to one or many clients. Bind is always on the PUB side. @@ -83,8 +79,9 @@ class Component(object): self.out_socket = None self.killed = False self.controller = None - # timeout after a full minute - self.heartbeat_timeout = 60 *1000 + # timeout on heartbeat is very short to avoid burning + # cycles on heartbeating. unit is milliconds + self.heartbeat_timeout = 0 # TODO: state_flag is deprecated, remove # TODO: error_state is deprecated, remove self.state_flag = COMPONENT_STATE.OK @@ -98,6 +95,8 @@ class Component(object): self.note = None self.confirmed = False self.devel = False + self.socks = None + self.last_ping = None # Humanhashes make this way easier to debug because they stick # in your mind unlike a 32 byte string of random hex. @@ -196,6 +195,10 @@ class Component(object): The core logic of the all components is run here. """ + log.info("Start %r" % self) + log.info("Pid %s" % os.getpid()) + log.info("Group %s" % os.getpgrp()) + self.start_tic = time.time() self.done = False # TODO: use state flag @@ -218,33 +221,25 @@ class Component(object): # ... until the controller signals GO self.loop() - self.shutdown() - log.info("Shutdown %r" % self) self.stop_tic = time.time() def run(self, catch_exceptions=True): """ Run the component. - - Optionally takes an argument to catch and log all exceptions - raised during execution. Use this with care since it makes it - very hard to debug since it mucks up your stacktraces. """ + try: + self._run() + except Exception as exc: + exc_info = sys.exc_info() + self.signal_exception(exc) - if catch_exceptions: - try: - self._run() - except Exception as exc: - exc_info = sys.exc_info() - self.signal_exception(exc) - - # Reraise the exception - raise exc_info[0], exc_info[1], exc_info[2] - finally: - - self.shutdown() - self.teardown_sockets() + # Reraise the exception + raise exc_info[0], exc_info[1], exc_info[2] + finally: + self.shutdown() + self.teardown_sockets() + log.info("Exiting %r" % self) def working(self): """ @@ -261,12 +256,90 @@ class Component(object): Loop to do work while we still have work to do. """ while self.working(): + self.heartbeat() self.do_work() def runtime(self): if self.ready() and self.start_tic and self.stop_tic: return self.stop_tic - self.start_tic + def heartbeat(self, timeout=0): + # wait for synchronization reply from the host + self.socks = dict(self.poll.poll(timeout)) + + # ---------------- + # Control Dispatch + # ---------------- + assert self.control_in, 'Component does not have a control_in socket' + + # If we're in devel mode drop out because the controller + # isn't guaranteed to be around anymore + if self.devel: + return + + if self.socks.get(self.control_in) == zmq.POLLIN: + msg = self.control_in.recv() + event, payload = CONTROL_UNFRAME(msg) + + # =========== + # Heartbeat + # =========== + + # The controller will send out a single number packed in + # a CONTROL_FRAME with ``heartbeat`` event every + # (n)-seconds. The component then has n seconds to + # respond to it. If not then it will be considered as + # malfunctioning or maybe CPU bound. + + if event == CONTROL_PROTOCOL.HEARTBEAT: + # Heart outgoing + heartbeat_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.OK, + payload + ) + + self.last_ping = float(payload) + # Echo back the heartbeat identifier to tell the + # controller that this component is still alive and + # doing work + self.control_out.send(heartbeat_frame) + + + # ========= + # Soft Kill + # ========= + + # Try and clean up properly and send out any reports or + # data that are done during a clean shutdown. Inform the + # controller that we're done. + elif event == CONTROL_PROTOCOL.SHUTDOWN: + self.signal_done() + self.shutdown() + + # ========= + # Hard Kill + # ========= + + # Just exit. + elif event == CONTROL_PROTOCOL.KILL: + self.kill() + + # In case we didn't receive a ping, send a pre-emptive + # pong to the monitor. + elif self.last_ping and time.time() - self.last_ping > 1: + # send a ping ahead of schedule + pre_pong = time.time() + heartbeat_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.OK, + str(pre_pong) + ) + + # Echo back the heartbeat identifier to tell the + # controller that this component is still alive and + # doing work + self.control_out.send(heartbeat_frame) + self.last_ping = pre_pong + # ---------------------------- # Cleanup & Modes of Failure # ---------------------------- @@ -325,7 +398,7 @@ class Component(object): # messages. while self.waiting: - socks = dict(self.poll.poll(self.heartbeat_timeout)) + #socks = dict(self.poll.poll(self.heartbeat_timeout)) msg = self.control_in.recv() event, payload = CONTROL_UNFRAME(msg) @@ -418,19 +491,29 @@ class Component(object): self.state_flag = COMPONENT_STATE.DONE if self.out_socket: - self.out_socket.send(str(CONTROL_PROTOCOL.DONE)) + msg = zmq.Message(str(CONTROL_PROTOCOL.DONE)) + self.out_socket.send(msg) - #notify controller we're done + + + # notify controller we're done done_frame = CONTROL_FRAME( CONTROL_PROTOCOL.DONE, '' ) - self.control_out.send(done_frame) - #notify internal work look that we're done + self.control_out.send(done_frame) + log.info("[%s] sent control done" % self.get_id) + + # there is a narrow race condition where we finish just + # after the Monitor accepts our prior heartbeat, but just + # before the next one is sent. So, we hang around for one + # last heartbeat, and wait an unusually long time. + self.heartbeat(timeout=5000) + + # notify internal work look that we're done self.done = True # TODO: use state flag - #log.info("[%s] DONE" % self.get_id) # ----------- # Messaging @@ -465,10 +548,10 @@ class Component(object): return self.connect_push_socket(self.addresses['merge_address']) def bind_result(self): - return self.bind_push_socket(self.addresses['result_address']) + return self.bind_push_socket(self.addresses['results_address']) def connect_result(self): - return self.connect_pull_socket(self.addresses['result_address']) + return self.connect_pull_socket(self.addresses['results_address']) def bind_push_socket(self, addr): push_socket = self.context.socket(self.zmq.PUSH) @@ -508,7 +591,7 @@ class Component(object): def bind_pub_socket(self, addr): pub_socket = self.context.socket(self.zmq.PUB) pub_socket.bind(addr) - #pub_socket.setsockopt(self.zmq.LINGER,0) + #pub_socket.setsockopt(self.zmq.LINGER, 0) self.out_socket = pub_socket return pub_socket diff --git a/zipline/core/host.py b/zipline/core/host.py index 804b25e6..ea1ca0aa 100644 --- a/zipline/core/host.py +++ b/zipline/core/host.py @@ -92,6 +92,10 @@ class ComponentHost(object): def unregister_component(self, component_id): del self.components[component_id] + @property + def pids(self): + return [proc.pid for proc in self.subprocesses] + def open(self): assert hasattr(self, 'zmq_flavor'), \ """ You must specify a flavor of ZeroMQ for all Topology @@ -111,6 +115,7 @@ class ComponentHost(object): for component in self.components.itervalues(): self.launch_component(component) + def is_running(self): """ DEPRECATED, left in for compatability for now. diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index 7b742627..11b91fa8 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -6,9 +6,10 @@ import gevent import itertools import logbook import gevent_zeromq +from setproctitle import setproctitle from signal import SIGHUP, SIGINT -from collections import OrderedDict +from collections import OrderedDict, Counter from zipline.utils.gpoll import _Poller as GeventPoller from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \ @@ -32,8 +33,7 @@ class UnknownChatter(Exception): def __init__(self, name): self.named = name def __str__(self): - return """Component calling itself "%s" talking on unexpected channel"""\ - % self.named + return """Component calling itself "%s" talking on unexpected channel""" % self.named log = logbook.Logger('Controller') @@ -42,8 +42,8 @@ log = logbook.Logger('Controller') # the system. PARAMETERS = ndict(dict( - GENERATIONAL_PERIOD = 1, - ALLOWED_SKIPPED_HEARTBEATS = 3, + GENERATIONAL_PERIOD = 10, #seconds + ALLOWED_SKIPPED_HEARTBEATS = 10, ALLOWED_INVALID_HEARTBEATS = 3, PRESTART_HEARBEATS = 3, SOURCES_START_HEARTBEATS = 3, @@ -94,6 +94,8 @@ class Controller(object): self.error_replay = OrderedDict() + self.missed_beats = Counter() + log.warn("Running Controller in development mode, will ONLY synchronize start.") def init_zmq(self, flavor): @@ -158,6 +160,7 @@ class Controller(object): def run(self): self.running = True self.init_zmq(self.zmq_flavor) + setproctitle('Monitor') self.state = CONTROL_STATES.INIT @@ -289,26 +292,32 @@ class Controller(object): # Wait the responses while self.alive: - socks = dict(poller.poll(self.period)) + socks = dict(poller.poll(0)) tic = time.time() - if tic - self.ctime > self.period: - break - if socks.get(self.router) == self.zmq.POLLIN: rawmessage = self.router.recv() if rawmessage: buffer.append(rawmessage) - try: if not self.router.getsockopt(self.zmq.RCVMORE): self.handle_recv(buffer[:]) buffer = [] + except INVALID_CONTROL_FRAME: log.error('Invalid frame', rawmessage) pass + # We break out of this loop if the time between + # sending and receiving the heartbeat is more + # than our poll period. + + if tic - self.ctime > self.period: + log.info("heartbeat loop timedout: %s" % (tic - self.ctime)) + log.info(repr(self.responses)) + break + # ================ # Heartbeat Stats # ================ @@ -319,10 +328,10 @@ class Controller(object): # Topology Status # ================ - # Is the entire topology told us its DONE + # Has the entire topology told us its DONE done = len(self.finished) == len(self.topology) - # Is the entire topology shown up to the party + # Has the entire topology shown up to the party complete = len(self.tracked) == len(self.topology) if complete: @@ -361,6 +370,7 @@ class Controller(object): self.signal_hangup() if not self.alive: + log.info('Breaking out of Monitor Loop') break def signal_hangup(self): @@ -370,7 +380,7 @@ class Controller(object): it. """ if not self.nosignals: - ppid = os.getpid() + ppid = os.getppid() log.warning("Sending SIGHUP") os.kill(ppid, SIGHUP) else: @@ -410,10 +420,10 @@ class Controller(object): # triggers the end of the topology. good = self.tracked & self.responses - bad = self.tracked - good - new = self.responses - good + bad = self.tracked - good - self.finished + new = self.responses - good - self.finished - missing = self.topology - self.tracked + missing = self.topology - self.tracked - self.finished for component in new: self.new(component) @@ -424,12 +434,13 @@ class Controller(object): for component in bad: self.fail(component) + + for component in missing: + if self.debug: - log.info('Bad component %r' % component) + log.info('Missing component %r' % component) if self.debug: - for component in missing: - log.info('Missing component %r' % component) for component in self.tracked: if component not in self.topology: @@ -439,11 +450,6 @@ class Controller(object): # Init Handlers # -------------- - def new_source(self): - #if self.state is CONTROL_STATES.RUNNING: - #self.state = SOURCES_READY - pass - def new_universal(self): pass @@ -460,11 +466,9 @@ class Controller(object): log.info('Now Tracking "%s" ' % component) universal = self.new_universal - init_handlers = { - 'FEED' : self.new_source, - } + init_handlers = {} - if component in self.topology or self.freeform: + if component in (self.topology - self.finished) or self.freeform: init_handlers.get(component, universal)() self.tracked.add(component) else: @@ -477,7 +481,6 @@ class Controller(object): # ------------------ def fail_universal(self): - pass # TODO: this requires higher order functionality log.error('System in exception state, shutting down') self.shutdown(soft=True) @@ -489,14 +492,10 @@ class Controller(object): universal = self.fail_universal fail_handlers = { } - if component in self.topology or self.freeform: + if component in (self.topology - self.finished) or self.freeform: log.warning('Component "%s" missed heartbeat' % component) self.tracked.remove(component) - - # TODO: determine when this propogates to a true - # failure, missing one heartbeat could just mean that - # its CPU overloaded - #fail_handlers.get(component, universal)() + fail_handlers.get(component, universal)() # ------------------- # Completion Handling @@ -559,9 +558,13 @@ class Controller(object): # Go to your bosom; knock there, and ask your heart what # it doth know... self.responses.add(identity) - elif status < self.ctime: + elif float(status) < self.ctime: # False face must hide what the false heart doth know. - log.warning('Delayed heartbeat received.') + log.warning('Delayed heartbeat received: %s' % msg) + elif float(status) > self.ctime: + # Pre-emptive heartbeat from the component + # log.info("pre-emptive pong: %s" % msg) + self.responses.add(identity) else: # Otherwise its something weird and we don't know # what to do so just say so, probably line noise @@ -625,7 +628,7 @@ class Controller(object): def do_error_replay(self): for (component, time), error in self.error_replay.iteritems(): - log.debug('Component Log for -- %s --:\n%s' % (component, error)) + log.info('Component Log for -- %s --:\n%s' % (component, error)) def shutdown(self, hard=False, soft=True): @@ -644,8 +647,3 @@ class Controller(object): self.state = CONTROL_STATES.TERMINATE log.info('Soft Shutdown') self.send_softkill() - - #self.do_error_replay() - - #self.pub.close() - #self.router.close() diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index ab0c17e6..7844b706 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -214,8 +214,9 @@ class PerformanceTracker(object): Publish the performance results asynchronously to a socket. """ - assert isinstance(results_addr, basestring), type(results_addr) - self.results_addr = results_addr + #assert isinstance(results_addr, basestring), type(results_addr) + #self.results_addr = results_addr + self.results_socket = results_addr def to_dict(self): """ @@ -261,7 +262,7 @@ class PerformanceTracker(object): self.todays_performance.calculate_performance() def handle_market_close(self): - + # add the return results from today to the list of DailyReturn objects. todays_date = self.market_close.replace(hour=0, minute=0, second=0) todays_return_obj = risk.DailyReturn( @@ -347,12 +348,9 @@ class PerformanceTracker(object): 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) - # this signals that the simulation is complete. - self.results_socket.send("DONE") - class Position(object): diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index 8e687c49..8e52ad3d 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -255,7 +255,7 @@ class RiskMetrics(): cur_return = math.log(1.0 + r) + cur_return #this is a guard for a single day returning -100% except ValueError: - log.warn("{cur} return, zeroing the returns".format(cur=cur_return)) + log.debug("{cur} return, zeroing the returns".format(cur=cur_return)) cur_return = 0.0 compounded_returns.append(cur_return) diff --git a/zipline/lines.py b/zipline/lines.py index 23eb8690..f56c1eb3 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -122,26 +122,25 @@ class SimulatedTrading(object): self.leased_sockets = [] self.sim_context = None - sockets = self.allocate_sockets(8) + sockets = self.allocate_sockets(7) addresses = { 'sync_address' : sockets[0], 'data_address' : sockets[1], 'feed_address' : sockets[2], 'merge_address' : sockets[3], - 'result_address' : sockets[4], - 'order_address' : sockets[5] + # TODO: this refers to the results of the merge, a + # horribly confusing name for the socket. + 'results_address' : sockets[4], } self.con = Controller( + sockets[5], sockets[6], - sockets[7], devel = self.devel ) # TODO: Not freeform - self.con.manage( - 'freeform' - ) + self.con.manage('freeform') self.started = False @@ -152,19 +151,15 @@ class SimulatedTrading(object): self.trading_client = TradeSimulationClient( self.trading_environment, self.sim_style, - config['log_socket'] + config['results_socket'] ) self.add_client(self.trading_client) # setup all sources self.sources = {} - #self.order_source = OrderDataSource() - #self.add_source(self.order_source) #setup transforms - #self.transaction_sim = TransactionSimulator(self.sim_style) self.transforms = {} - #self.add_transform(self.transaction_sim) self.sim.register_controller( self.con ) @@ -260,10 +255,10 @@ class SimulatedTrading(object): order_count ) - if config.has_key('log_socket'): - log_socket = config['log_socket'] + if config.has_key('results_socket'): + results_socket = config['results_socket'] else: - log_socket = None + results_socket = None #------------------- # Simulation #------------------- @@ -273,7 +268,7 @@ class SimulatedTrading(object): 'allocator' : allocator, 'simulator_class' : simulator_class, 'simulation_style' : simulation_style, - 'log_socket' : log_socket, + 'results_socket' : results_socket, 'devel' : config.get('devel', False) }) #------------------- @@ -322,9 +317,6 @@ class SimulatedTrading(object): def get_cumulative_performance(self): return self.trading_client.perf.cumulative_performance.to_dict() - def publish_to(self, results_socket): - self.trading_client.perf.publish_to(results_socket) - def allocate_sockets(self, n): """ Allocate sockets local to this line, track them so @@ -399,16 +391,17 @@ class SimulatedTrading(object): # the supervisory layer # TODO: better way of identifying concurrency substrate - if self.sim.zmq_flavor == 'thread': - log.debug('Blocking') - for thread in self.sim.subthreads: - #log.debug('Waiting on %r' % thread) - log.debug('Waiting on %r' % thread) - thread.join() - log.debug('Yielded on %r' % thread) - else: - for process in self.sim.subprocesses: - process.join() + if blocking: + if self.sim.zmq_flavor == 'thread': + log.debug('Blocking') + for thread in self.sim.subthreads: + #log.debug('Waiting on %r' % thread) + log.debug('Waiting on %r' % thread) + thread.join() + log.debug('Yielded on %r' % thread) + else: + for process in self.sim.subprocesses: + process.join() @property def is_success(self): diff --git a/zipline/protocol.py b/zipline/protocol.py index 35cef68b..866e5410 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -513,9 +513,9 @@ def BT_UPDATE_FRAME(prefix, payload): def BT_UPDATE_UNFRAME(msg): """ - Risk and Perf framing methods prefix the payload with + Risk, Perf, and LOG framing methods prefix the payload with a shorthand for their type. That way, all messages received from the socket - can be PERF_FRAMED(), whether they are risk or perf. + can be PERF_FRAMED(), whether they are risk, perf, or log. """ prefix, payload = msgpack.loads(msg, use_list=True) return dict(prefix=prefix, payload=payload) @@ -608,6 +608,7 @@ SIMULATION_STYLE = Enum( LOG_FIELDS = set(['func_name', 'lineno', 'time', 'msg',\ 'level', 'channel', ]) LOG_EXTRA_FIELDS = set(['algo_dt',]) +LOG_DONE = "DONE" def LOG_FRAME(payload): """ @@ -621,14 +622,14 @@ def LOG_FRAME(payload): 'level' : 4, #Logbook enum 'channel' : 'MyLogger' } - + Frame checks that we have all expected fields and exports an event/payload dict as JSON. """ assert isinstance(payload, dict), \ "LOG_FRAME expected a dict" - + assert payload.has_key('algo_dt'), \ "LOG_FRAME with no algo_dt" assert payload.has_key('time'), \ @@ -639,21 +640,18 @@ def LOG_FRAME(payload): "LOG_FRAME with no level" assert payload.has_key('msg'),\ "LOG_FRAME with no message" - - data = {} - data['e'] = 'LOG' - data['p'] = payload - - return msgpack.dumps(data) + + + return BT_UPDATE_FRAME('LOG', payload) def LOG_UNFRAME(msg): """ - Expects a json serialized dictionary in event/payload format. + msg should be a tuple of ('LOG',dict) """ record = msgpack.loads(msg) - assert record['e'] == 'LOG' - assert record.has_key('p') - - return record['p'] - + assert isinstance(record, tuple) + assert len(record) == 2 + assert record[0] == 'LOG' + payload = record[1] + return payload diff --git a/zipline/transforms/base.py b/zipline/transforms/base.py index 6e406efc..9a232ad9 100644 --- a/zipline/transforms/base.py +++ b/zipline/transforms/base.py @@ -4,6 +4,12 @@ import zipline.protocol as zp from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \ CONTROL_FRAME, CONTROL_UNFRAME + +import logbook +import time + +log = logbook.Logger('BaseTransform') + class BaseTransform(Component): """ Top level execution entry point for the transform @@ -46,35 +52,15 @@ class BaseTransform(Component): - send the transformed event """ - socks = dict(self.poll.poll(self.heartbeat_timeout)) - # TODO: Abstract this out, maybe on base component - if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN: - msg = self.control_in.recv() - event, payload = CONTROL_UNFRAME(msg) - - # -- Heartbeat -- - if event == CONTROL_PROTOCOL.HEARTBEAT: - # Heart outgoing - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.OK, - payload - ) - self.control_out.send(heartbeat_frame) - - # -- Soft Kill -- - elif event == CONTROL_PROTOCOL.SHUTDOWN: - self.signal_done() - self.shutdown() - - # -- Hard Kill -- - elif event == CONTROL_PROTOCOL.KILL: - self.kill() - - if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN: + if self.feed_socket in self.socks and self.socks[self.feed_socket] == self.zmq.POLLIN: message = self.feed_socket.recv() + #import msgpack + #event = msgpack.loads(message) + #log.info(event) if message == str(CONTROL_PROTOCOL.DONE): + log.info("signaling done") self.signal_done() return diff --git a/zipline/utils/log_utils.py b/zipline/utils/log_utils.py index bf08cf41..6bcf80f8 100644 --- a/zipline/utils/log_utils.py +++ b/zipline/utils/log_utils.py @@ -4,13 +4,16 @@ import pytz import datetime from logbook import NOTSET -from logbook.handlers import Handler +from logbook.handlers import Handler, FileHandler from zipline.protocol import LOG_FRAME, LOG_FIELDS, \ LOG_EXTRA_FIELDS from contextlib import contextmanager + +log = logbook.Logger("LogUtils") + class redirecter(object): def __init__(self, logger, name): self.logger = logger @@ -30,10 +33,10 @@ class redirecter(object): self.logger.error(out_form) self.buffer = bytes() -class log_redirecter(object): +class log_redirecter(object): def __init__(self, logger): self.logger = logger - + def write(self, line): #Absorb blank lines from print statements. if line =='\n': @@ -42,7 +45,7 @@ class log_redirecter(object): else: #TODO: add logic to guarantee we made this self.logger.info(line.strip('\n')) - + def flush(self, final=False): pass @@ -70,7 +73,7 @@ def stdout_only_pipe(logger, pipe_name): import sys orig_fd = sys.stdout sys.stdout = log_redirecter(logger) - + yield sys.stdout.flush() sys.stdout = orig_fd @@ -82,8 +85,8 @@ class ZeroMQLogHandler(Handler): Setup is similar to logbook.queues.ZeroMQHandler, except we connect instead of binding and we extract record fields into a dict. """ - - def __init__(self, uri=None, level=NOTSET, filter=None, bubble=False, + + def __init__(self, socket=None, level=NOTSET, filter=None, bubble=False, context=None, fds = LOG_FIELDS, extra_fds = LOG_EXTRA_FIELDS): Handler.__init__(self, level, filter, bubble) try: @@ -94,11 +97,11 @@ class ZeroMQLogHandler(Handler): #: the zero mq context self.context = context #: the zero mq socket. - self.socket = self.context.socket(zmq.PUSH) + self.socket = socket #self.context.socket(zmq.PUSH) - self.uri = uri - if uri is not None: - self.socket.connect(uri) + #self.uri = uri + #if uri is not None: + # self.socket.connect(uri) self.fds = fds self.extra_fds = extra_fds @@ -109,21 +112,21 @@ class ZeroMQLogHandler(Handler): fields to make json happy. """ from zipline.utils.date_utils import EPOCH - + #Needed to extract record info from dictionary. record.pull_information() #Logbook stores record times as datetime objects, which - #can't be serialized by JSON, so we need to convert to + #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'): algo_dt = record.extra['algo_dt'] @@ -131,31 +134,30 @@ class ZeroMQLogHandler(Handler): if isinstance(algo_dt, datetime.datetime): algo_dt = EPOCH(algo_dt.replace(tzinfo = pytz.utc)) record.extra['algo_dt'] = algo_dt - + data = {} - + #Extract all the fields we care about from LogRecord's internal - #dictionary. + #dictionary. for field in iter(self.fds): if record.__dict__.has_key(field): data[field] = record.__dict__[field] else: data[field] = None - + for field in iter(self.extra_fds): if record.extra.has_key(field): data[field] = record.extra[field] else: data[field] = None return data - + def emit(self, record): """Extract relevant fields and send info as JSON over a zmq socket.""" payload = self.export_record(record) self.socket.send(LOG_FRAME(payload)) - + def close(self): - #self.socket.close() pass - + #self.socket.close()