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

Merge pull request #78 from quantopian/hanging_chad

Browse Source 
Hanging chad
This commit is contained in:
fawce
2012-07-18 14:21:54 -07:00
parent 56da9d53f8 fb2b0c3be5
commit 95a987db39
13 changed files with 302 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_optimize.py
+2
View File
@@ -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,
zipline/components/aggregator.py
+3 -10
View File
@@ -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)
zipline/components/feed.py
+1
View File
@@ -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
zipline/components/tradesimulation.py
+53 -43
View File
@@ -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])
zipline/core/component.py
+119 -36
View File
@@ -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
zipline/core/host.py
+5
View File
@@ -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.
zipline/core/monitor.py
+41 -43
View File
@@ -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()
zipline/finance/performance.py
+5 -7
View File
@@ -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):
zipline/finance/risk.py
+1 -1
View File
@@ -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)
zipline/lines.py
+22 -29
View File
@@ -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):
zipline/protocol.py
+14 -16
View File
@@ -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
zipline/transforms/base.py
+11 -25
View File
@@ -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
zipline/utils/log_utils.py
+25 -23
View File
@@ -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()