diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index c8b585d3..72437d2f 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -37,13 +37,10 @@ class TradeSimulationClient(Component): self.log_socket = log_socket - #If we have a log socket,setup context managers for capturing user logs. + #If we have a log socket,setup context managers for capturing user prints. if log_socket: - #Temporarily putting this here to fix circular import bug. - - - log = Logger("User Log") + log = Logger("Print") self.stdout_capture = stdout_only_pipe(log, 'user algo stdout') handler = queues.ZeroMQLogHandler(uri = log_socket) @@ -62,7 +59,7 @@ class TradeSimulationClient(Component): # register the trading_client's order method with the algorithm self.algorithm.set_order(self.order) - # ask the algorithm to initialize, routing stdout to a zmq PUB socket. + # ask the algorithm to initialize, routing stdout to a zmq PUSH socket. if self.log_socket: with self.zmq_out, self.stdout_capture: self.algorithm.initialize() @@ -166,19 +163,19 @@ class TradeSimulationClient(Component): # data injection pipeline for log rerouting # any fields injected here should be added to - # LOG_EXTRA_FIELDS in qexec/web/client_protocol + # LOG_EXTRA_FIELDS in zipline/protocol.py - def inject_event_data(record): - record.extra['algo_dt'] = event.dt - #record.extra['whatever else we want'] = event.whatever - - data_injector = Processor(inject_event_data) - log_pipeline = NestedSetup([self.zmq_out, - #e.g. FileHandler(...) - data_injector]) # try to run algo with log rerouting - if log_socket: + if self.log_socket: + def inject_event_data(record): + record.extra['algo_dt'] = event.dt + #record.extra['whatever else we want'] = event.whatever + + data_injector = Processor(inject_event_data) + log_pipeline = NestedSetup([self.zmq_out, + #e.g. FileHandler(...) + data_injector]) with log_pipeline, self.stdout_capture: self.algorithm.handle_data(data) # if no log socket, just run the algo normally diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index 5b2384f4..a42508c9 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -309,7 +309,7 @@ class Controller(object): assert self.route_socket assert self.pub_socket - assert self.cancel_socket + #assert self.cancel_socket # -- Publish -- # ============= @@ -318,9 +318,9 @@ class Controller(object): # -- Cancel -- # ============= - assert isinstance(self.cancel_socket,basestring), self.cancel_socket - self.cancel = self.context.socket(self.zmq.REP) - self.cancel.connect(self.cancel_socket) + #assert isinstance(self.cancel_socket,basestring), self.cancel_socket + #self.cancel = self.context.socket(self.zmq.REP) + #self.cancel.connect(self.cancel_socket) # -- Router -- # ============= @@ -330,9 +330,9 @@ class Controller(object): poller = self.zmq.Poller() poller.register(self.router, self.zmq.POLLIN) - poller.register(self.cancel, self.zmq.POLLIN) + #poller.register(self.cancel, self.zmq.POLLIN) - self.associated += [self.pub, self.router, self.cancel] + self.associated += [self.pub, self.router]# self.cancel] # TODO: actually do this self.state = CONTROL_STATES.SOURCES_READY @@ -369,12 +369,12 @@ class Controller(object): self.logging.error('Invalid frame', rawmessage) pass - if socks.get(self.cancel) == self.zmq.POLLIN: - self.logging.info('[Controller] Received Cancellation') - rawmessage = self.cancel.recv() - self.cancel.send('') - self.shutdown(soft=True) - break + #if socks.get(self.cancel) == self.zmq.POLLIN: + # self.logging.info('[Controller] Received Cancellation') + # rawmessage = self.cancel.recv() + # self.cancel.send('') + # self.shutdown(soft=True) + # break self.beat() diff --git a/zipline/lines.py b/zipline/lines.py index 8aace760..4213fb78 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -135,9 +135,7 @@ class SimulatedTrading(object): sockets[7], logger = LOGGER ) - - self.con.cancel_socket = self.allocator.lease(1)[0] - + # TODO: Not freeform self.con.manage( 'freeform' @@ -151,7 +149,7 @@ class SimulatedTrading(object): self.trading_client = TradeSimulationClient( self.trading_environment, self.sim_style, - config.log_socket + config['log_socket'] ) self.add_client(self.trading_client) @@ -258,6 +256,11 @@ class SimulatedTrading(object): order_amount, order_count ) + + if config.has_key('log_socket'): + log_socket = config['log_socket'] + else: + log_socket = None #------------------- # Simulation #------------------- @@ -266,7 +269,8 @@ class SimulatedTrading(object): 'trading_environment' : trading_environment, 'allocator' : allocator, 'simulator_class' : simulator_class, - 'simulation_style' : simulation_style + 'simulation_style' : simulation_style, + 'log_socket' : log_socket }) #------------------- diff --git a/zipline/protocol.py b/zipline/protocol.py index 7081291f..fb27b2d2 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -601,3 +601,58 @@ SIMULATION_STYLE = Enum( 'FIXED_SLIPPAGE', 'NOOP' ) + +#Global variables for the fields we extract out of a standard logbook record. +LOG_FIELDS = set(['func_name', 'lineno', 'time', 'msg',\ + 'level', 'channel', ]) +LOG_EXTRA_FIELDS = set(['algo_dt',]) + +def LOG_FRAME(payload): + """ + Expects a dictionary of the form: + { + 'algo_dt' : 1199223000, #Algo simulation date. + 'time' : 1199223001, #Realtime date of log creation. + 'func_name' : 'foo', + 'lineno' : 46, + 'message' : 'Successfully disintegrated llama #3' + '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 received:"+str(type(log_record)) + + assert payload.has_key('algo_dt'), \ + "LOG_FRAME with no algo_dt" + assert payload.has_key('time'), \ + "LOG_FRAME with no time" + assert payload.has_key('channel'),\ + "LOG_FRAME with no channel" + assert payload.has_key('level'),\ + "LOG_FRAME with no level" + assert payload.has_key('message'),\ + "LOG_FRAME with no message" + + data = {} + data['e'] = 'LOG' + data['p'] = payload + + return json.dumps(data) + +def LOG_UNFRAME(msg): + """ + Expects a json serialized dictionary in event/payload format. + """ + record = json.loads(data) + assert record['e'] == 'LOG' + assert record.has_key('p') + + return record['p'] + + + diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index 116fb2b9..fa68366e 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -137,3 +137,25 @@ class NoopAlgorithm(object): def get_sid_filter(self): return None + +class LogTestAlgorithm(object): + + def __init__(self): + import logbook + self.log = logbook.Logger() + + def initialize(self): + self.log.info("Initialize") + + def set_order(self, order_callable): + pass + + def set_portfolio(self, portfolio): + pass + + def handle_data(self, data): + self.log.info("handle_data") + pass + + def get_sid_filter(self): + return None