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

re-enabled the monitor communication in all components.

Browse Source
This commit is contained in:
fawce
2012-07-17 11:01:15 -04:00
parent 4981b600f0
commit 1cc56b52e7
6 changed files with 59 additions and 55 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
+1 -1
View File
@@ -71,7 +71,7 @@ class Aggregate(Component):
# ----------------
# Control Dispatch
# ----------------
#do_handle_control_events(self, socks)
do_handle_control_events(self, socks)
# -------------
# Work Dispatch
zipline/components/tradesimulation.py
+31 -28
View File
@@ -11,6 +11,8 @@ from zipline.utils.protocol_utils import ndict
from zipline.utils.log_utils import ZeroMQLogHandler, stdout_only_pipe
from logbook import Logger, NestedSetup, Processor, queues
from zipline.core.controlled import do_handle_control_events
log = logbook.Logger('TradeSimulation')
@@ -33,15 +35,15 @@ class TradeSimulationClient(Component):
self.event_data = ndict()
self.perf = perf.PerformanceTracker(self.trading_environment)
self.log_socket = log_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,26 +55,20 @@ 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!
self.setup_logging(self.context)
#Initialize log capture for testing purposes.
def setup_logging(self, context):
@@ -84,6 +80,13 @@ class TradeSimulationClient(Component):
def do_work(self):
# poll all the sockets
socks = dict(self.poll.poll(self.heartbeat_timeout))
# ----------------
# Control Dispatch
# ----------------
do_handle_control_events(self, socks)
# see if the poller has results for the result_feed
if socks.get(self.result_feed) == self.zmq.POLLIN:
@@ -101,7 +104,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 +118,7 @@ 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 +147,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,22 +169,22 @@ 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:
def inject_event_data(record):
#Record the simulation time.
record.extra['algo_dt'] = self.current_dt
data_injector = Processor(inject_event_data)
data_injector = Processor(inject_event_data)
log_pipeline = NestedSetup([self.zmq_out,
#e.g. FileHandler(...)
data_injector])
@@ -190,14 +193,14 @@ class TradeSimulationClient(Component):
# if no log socket, just run the algo normally
else:
self.algorithm.handle_data(data)
#Testing utility for log capture.
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
+8 -4
View File
@@ -422,7 +422,11 @@ class Component(object):
self.state_flag = COMPONENT_STATE.DONE
if self.out_socket:
self.out_socket.send(str(CONTROL_PROTOCOL.DONE))
log.info("[%s] sending DONE" % self.get_id)
msg = zmq.Message(str(CONTROL_PROTOCOL.DONE))
self.out_socket.send(msg)
log.info("[%s] sent DONE" % self.get_id)
#notify controller we're done
done_frame = CONTROL_FRAME(
@@ -431,10 +435,10 @@ class Component(object):
)
self.control_out.send(done_frame)
#notify internal work look that we're done
# notify internal work look that we're done
self.done = True # TODO: use state flag
#log.info("[%s] DONE" % self.get_id)
#log.info("[%s] DONE" % self.get_id)
# -----------
# Messaging
@@ -512,7 +516,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/monitor.py
+1 -1
View File
@@ -415,7 +415,7 @@ class Controller(object):
bad = self.tracked - good
new = self.responses - good
missing = self.topology - self.tracked
missing = self.topology - self.tracked - self.finished
for component in new:
self.new(component)
zipline/transforms/base.py
+16 -21
View File
@@ -4,6 +4,13 @@ import zipline.protocol as zp
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \
CONTROL_FRAME, CONTROL_UNFRAME
from zipline.core.controlled import do_handle_control_events
import logbook
import time
log = logbook.Logger('BaseTransform')
class BaseTransform(Component):
"""
Top level execution entry point for the transform
@@ -46,35 +53,23 @@ 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)
# ----------------
# Control Dispatch
# ----------------
do_handle_control_events(self, socks)
# -- 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:
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