wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-23 12:50:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

all manner of changes to support logging

Browse Source
This commit is contained in:
scottsanderson
2012-06-28 16:28:10 -04:00
parent a222e1f0d8
commit 4c40a3d55c
5 changed files with 111 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/components/tradesimulation.py
+13 -16
View File
@@ -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
zipline/core/monitor.py
+12 -12
View File
@@ -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()
zipline/lines.py
+9 -5
View File
@@ -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
})
#-------------------
zipline/protocol.py
+55
View File
@@ -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']
zipline/test_algorithms.py
+22
View File
@@ -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