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

Merge branch 'master' of github.com:quantopian/zipline

Browse Source 
Conflicts:
	zipline/protocol.py
	zipline/test/client.py
	zipline/test/test_messaging.py
This commit is contained in:
fawce
2012-02-29 23:43:27 -05:00
parent 5247ace0ca 554199d2d3
commit 2827c2036b
16 changed files with 925 additions and 355 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/conf.py
+8 -1
View File
@@ -25,7 +25,14 @@ sys.path.append(os.path.abspath('..'))
# Add any Sphinx extension module names here, as strings. They can be extensions
# coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
extensions = ['sphinx.ext.autodoc', 'sphinx.ext.intersphinx', 'sphinx.ext.todo', 'sphinx.ext.coverage', 'sphinx.ext.mathjax', 'sphinx.ext.viewcode']
extensions = [
'sphinx.ext.autodoc',
'sphinx.ext.intersphinx',
'sphinx.ext.todo',
'sphinx.ext.coverage',
'sphinx.ext.mathjax',
'sphinx.ext.viewcode'
]
# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']
docs/index.rst
+1 -1
View File
@@ -6,7 +6,7 @@
Contents:
.. toctree::
:maxdepth: 2
:maxdepth: 4
notes.rst
modules.rst
docs/zipline.rst
+24 -25
View File
@@ -9,6 +9,14 @@ zipline Package
:undoc-members:
:show-inheritance:
:mod:`cli` Module
-----------------
.. automodule:: zipline.cli
:members:
:undoc-members:
:show-inheritance:
:mod:`component` Module
-----------------------
@@ -17,30 +25,6 @@ zipline Package
:undoc-members:
:show-inheritance:
:mod:`daemon` Module
--------------------
.. automodule:: zipline.daemon
:members:
:undoc-members:
:show-inheritance:
:mod:`db` Module
----------------
.. automodule:: zipline.db
:members:
:undoc-members:
:show-inheritance:
:mod:`host_settings` Module
---------------------------
.. automodule:: zipline.host_settings
:members:
:undoc-members:
:show-inheritance:
:mod:`messaging` Module
-----------------------
@@ -65,6 +49,14 @@ zipline Package
:undoc-members:
:show-inheritance:
:mod:`serial` Module
--------------------
.. automodule:: zipline.serial
:members:
:undoc-members:
:show-inheritance:
:mod:`sources` Module
---------------------
@@ -73,6 +65,14 @@ zipline Package
:undoc-members:
:show-inheritance:
:mod:`topos` Module
-------------------
.. automodule:: zipline.topos
:members:
:undoc-members:
:show-inheritance:
:mod:`util` Module
------------------
@@ -86,7 +86,6 @@ Subpackages
.. toctree::
zipline.finance
zipline.test
zipline.transforms
docs/zipline.test.rst
+8 -24
View File
@@ -9,22 +9,6 @@ test Package
:undoc-members:
:show-inheritance:
:mod:`dummy` Module
-------------------
.. automodule:: zipline.test.dummy
:members:
:undoc-members:
:show-inheritance:
:mod:`factory` Module
---------------------
.. automodule:: zipline.test.factory
:members:
:undoc-members:
:show-inheritance:
:mod:`test_devsimulator` Module
-------------------------------
@@ -33,14 +17,6 @@ test Package
:undoc-members:
:show-inheritance:
:mod:`test_finance` Module
--------------------------
.. automodule:: zipline.test.test_finance
:members:
:undoc-members:
:show-inheritance:
:mod:`test_messaging` Module
----------------------------
@@ -49,6 +25,14 @@ test Package
:undoc-members:
:show-inheritance:
:mod:`test_monitor` Module
--------------------------
.. automodule:: zipline.test.test_monitor
:members:
:undoc-members:
:show-inheritance:
:mod:`test_sanity` Module
-------------------------
etc/requirements.txt
+1 -1
View File
@@ -3,4 +3,4 @@ pyzmq==2.1.11
gevent-zeromq==0.2.2
msgpack-python==0.1.12
humanhash==0.0.1
pymongo==2.1.1
ujson=1.18
zipline/component.py
+197 -46
View File
@@ -1,8 +1,13 @@
"""
Commonly used messaging components.
Contains the base class for all components.
"""
import os
import uuid
import time
import socket
import humanhash
@@ -10,66 +15,113 @@ import zipline.util as qutil
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE
class Component(object):
"""
Base class for components. Defines the the base messaging
interface for components.
:param addresses: a dict of name_string -> zmq port address strings.
Must have the following entries
:param sync_address: socket address used for synchronizing the start of
all workers, heartbeating, and exit notification
will be used in REP/REQ sockets. Bind is always on
the REP side.
:param data_address: socket address used for data sources to stream
their records. Will be used in PUSH/PULL sockets
between data sources and a ParallelBuffer (aka
the Feed). Bind will always be on the PULL side
(we always have N producers and 1 consumer)
:param feed_address: socket address used to publish consolidated feed
from serialization of data sources
will be used in PUB/SUB sockets between Feed and
Transforms. Bind is always on the PUB side.
:param merge_address: socket address used to publish transformed
values. will be used in PUSH/PULL from many
transforms to one MergedParallelBuffer (aka the
Merge). Bind will always be on the PULL side (we
always have N producers and 1 consumer)
:param result_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.
bind/connect methods will return the correct socket type for each
address.
"""
def __init__(self):
"""
:addresses: a dict of name_string -> zmq port address strings. Must have the following entries::
- sync_address: socket address used for synchronizing the start of all workers, heartbeating, and exit notification
will be used in REP/REQ sockets. Bind is always on the REP side.
- data_address: socket address used for data sources to stream their records.
will be used in PUSH/PULL sockets between data sources and a ParallelBuffer (aka the Feed). Bind
will always be on the PULL side (we always have N producers and 1 consumer)
- feed_address: socket address used to publish consolidated feed from serialization of data sources
will be used in PUB/SUB sockets between Feed and Transforms. Bind is always on the PUB side.
- merge_address: socket address used to publish transformed values.
will be used in PUSH/PULL from many transforms to one MergedParallelBuffer (aka the Merge). Bind
will always be on the PULL side (we always have N producers and 1 consumer)
- result_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.
Bind/Connect methods will return the correct socket type for each address. Any sockets on which recv is expected to be called
will also return a Poller.
"""
self.zmq = None
self.context = None
self.addresses = None
self.out_socket = None
self.gevent_needed = False
self.killed = False
self.controller = None
self.heartbeat_timeout = 2000
self.state_flag = COMPONENT_STATE.OK # OK | DONE | EXCEPTION
self._exception = None
self.start_tic = None
self.stop_tic = None
# Humanhashes make this way easier to debug because they
# stick in your mind unlike a 32 byte string of random hex.
self.guid = uuid.uuid4()
self.huid = humanhash.humanize(self.guid.hex)
self.init()
def init(self):
"""
Subclasses should override this to extend the setup for
the class. Shouldn't have side effects.
"""
pass
# ------------
# Core Methods
# ------------
def open(self):
raise NotImplementedError
def destroy(self):
"""
Tear down after normal operation.
Open the connections needed to start doing work.
"""
raise NotImplementedError
def kill(self):
def ready(self):
"""
Tear down ( fast ) as a mode of failure in the
simulation.
Return ``True`` if and only if the component has finished execution.
"""
raise NotImplementedError
return self.state_flag in [COMPONENT_STATE.DONE, \
COMPONENT_STATE.EXCEPTION]
def successful(self):
"""
Return ``True`` if and only if the component has finished execution
successfully, that is, without raising an error.
"""
return self.state_flag == COMPONENT_STATE.DONE and not \
self.exception
@property
def exception(self):
"""
Holds the exception that the component failed on, or
``None`` if the component has not failed.
"""
return self._exception
def do_work(self):
raise NotImplementedError
def _run(self):
self.start_tick = time.clock()
self.done = False # TODO: use state flag
self.sockets = []
@@ -81,7 +133,14 @@ class Component(object):
import zmq
self.zmq = zmq
# TODO: this can cause max fd errors on BSD machines with
# low ulimits, its perfectly fine to use one Context in
# multithreaded enviroments, its only in multiprocess
# systems where this becomes needed. Add this option.
#
# http://zeromq.github.com/pyzmq/morethanbindings.html#thread-safety
self.context = self.zmq.Context()
self.setup_poller()
self.open()
@@ -89,11 +148,13 @@ class Component(object):
self.setup_control()
self.loop()
#close all the sockets
for sock in self.sockets:
sock.close()
def run(self, catch_exceptions=False):
self.end_tick = time.clock()
# shouldn't block if we've done our job correctly
# self.context.term()
def run(self, catch_exceptions=True):
"""
Run the component.
@@ -114,14 +175,21 @@ class Component(object):
self.signal_exception(exc)
fail = exc
finally:
self.shutdown()
self.teardown_sockets()
if self.context:
self.context.destroy()
if fail:
raise fail
else:
self._run()
if(self.context != None):
self.context.destroy()
try:
self._run()
finally:
self.shutdown()
self.teardown_sockets()
def loop(self):
"""
@@ -141,12 +209,50 @@ class Component(object):
self.receive_sync_ack() # blocking
def runtime(self):
if self.ready():
return self.stop_tic - self.start_tic
# ----------------------------
# Cleanup & Modes of Failure
# ----------------------------
def teardown_sockets(self):
"""
Close all zmq sockets safely. This is universal, no matter
where this is running it will need the sockets closed.
"""
#close all the sockets
for sock in self.sockets:
sock.close()
def shutdown(self):
"""
Clean shutdown.
Tear down after normal operation.
"""
pass
def kill(self):
"""
Unclean shutdown.
Tear down ( fast ) as a mode of failure in the
simulation or on service halt.
Context specific.
"""
raise NotImplementedError
# ----------------------
# Internal Maintenance
# ----------------------
def signal_exception(self, exc=None):
self.state_flag = COMPONENT_STATE.EXCEPTION
self._exception = exc
qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id))
def signal_done(self):
@@ -258,15 +364,15 @@ class Component(object):
"""
assert self.controller
self.control_out = self.controller.message_sender()
self.control_in = self.controller.message_listener()
self.control_out = self.controller.message_sender(context=self.context)
self.control_in = self.controller.message_listener(context=self.context)
self.poll.register(self.control_in, self.zmq.POLLIN)
self.sockets.extend([self.control_in, self.control_out])
def setup_sync(self):
"""
Setup the sync socket and poller.
Setup the sync socket and poller. ( Connect )
"""
qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id))
@@ -275,7 +381,7 @@ class Component(object):
self.sync_socket.connect(self.addresses['sync_address'])
#self.sync_socket.setsockopt(self.zmq.LINGER,0)
# Explictly, a different poller for obvious reasons.
# Explictly a different poller for obvious reasons.
# I'm not fond of having this poller init'd as a side
# effect of a method call. Still thinking about where to
# put it at the moment though...
@@ -288,21 +394,66 @@ class Component(object):
# Description and Debug
# ---------------------
def extern_logger(self):
"""
Pipe logs out to a provided logging interface.
"""
pass
def setup_extern_logger(self):
"""
Pipe logs out to a provided logging interface.
"""
pass
@property
def get_id(self):
"""
The descriptive name of the component.
"""
return 'UNKNOWN COMPONENT'
@property
def get_type(self):
"""
The data flow type of the component.
- ``SOURCE``
- ``CONDUIT``
- ``SINK``
"""
raise NotImplementedError
@property
def get_pure(self):
"""
Describes whehter this component purely functional,
i.e. for a given set of inputs is it guaranteed to
always give the same output . Components that are
side-effectful are, generally, not pure.
"""
return False
def debug(self):
"""
Debug information about the component.
"""
return (
self.get_id ,
self.huid ,
socket.gethostname() ,
os.getpid() ,
hex(id(self)) ,
)
return {
'id' : self.get_id ,
'huid' : self.huid ,
'host' : socket.gethostname() ,
'pid' : os.getpid() ,
'memaddress' : hex(id(self)) ,
'ready' : self.successful() ,
'succesfull' : self.ready() ,
}
def __len__(self):
"""
Some components overload this for debug purposes
"""
raise NotImplementedError
def __repr__(self):
"""
zipline/messaging.py
+230 -146
View File
@@ -1,14 +1,14 @@
"""
Commonly used messaging components.
"""
import json
import uuid
import datetime
import zipline.protocol as zp
import ujson as json
import zipline.util as qutil
from zipline.component import Component
from zipline.protocol import CONTROL_PROTOCOL
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \
COMPONENT_STATE
class ComponentHost(Component):
"""
@@ -18,13 +18,21 @@ class ComponentHost(Component):
def __init__(self, addresses, gevent_needed=False):
Component.__init__(self)
self.addresses = addresses
self.addresses = addresses
self.gevent_needed = gevent_needed
self.running = False
self.init()
def init(self):
# Component Registry
# ----------------------
self.components = {}
# ----------------------
self.sync_register = {}
self.timeout = datetime.timedelta(seconds=5)
self.gevent_needed = gevent_needed
self.heartbeat_timeout = 2000
self.feed = ParallelBuffer()
self.merge = MergedParallelBuffer()
@@ -35,25 +43,38 @@ class ComponentHost(Component):
self.register_components([self.feed, self.merge, self.passthrough])
def register_controller(self, controller):
"""
Add the given components to the registry. Establish
communication with them.
"""
if self.controller != None:
raise Exception("There can be only one!")
self.controller = controller
# Propogate the controller to all the subcomponents
for component in self.components.itervalues():
component.controller = controller
def register_components(self, components):
for component in components:
component.gevent_needed = self.gevent_needed
component.addresses = self.addresses
"""
Add the given components to the registry. Establish
communication with them.
"""
assert isinstance(components, list)
if self.controller:
component.controller = self.controller
for component in components:
component.gevent_needed = self.gevent_needed
component.addresses = self.addresses
component.controller = self.controller
self.components[component.get_id] = component
self.sync_register[component.get_id] = datetime.datetime.utcnow()
if(isinstance(component, DataSource)):
if isinstance(component, DataSource):
self.feed.add_source(component.get_id)
if(isinstance(component, BaseTransform)):
if isinstance(component, BaseTransform):
self.merge.add_source(component.get_id)
def unregister_component(self, component_id):
@@ -62,6 +83,7 @@ class ComponentHost(Component):
def setup_sync(self):
"""
Setup the sync socket and poller. ( Bind )
"""
qutil.LOGGER.debug("Connecting sync server.")
@@ -108,17 +130,17 @@ class ComponentHost(Component):
if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN:
msg = self.sync_socket.recv()
parts = msg.split(':')
if len(parts) != 2:
qutil.LOGGER.info("got bad confirm: {msg}".format(msg=msg))
continue
sync_id, status = parts
try:
parts = msg.split(':')
sync_id, status = parts
except ValueError as exc:
self.signal_exception(exc)
if status == str(CONTROL_PROTOCOL.DONE): # TODO: other way around
qutil.LOGGER.info("{id} is DONE".format(id=sync_id))
self.unregister_component(sync_id)
self.state_flag = COMPONENT_STATE.DONE
else:
self.sync_register[sync_id] = datetime.datetime.utcnow()
@@ -126,28 +148,18 @@ class ComponentHost(Component):
# send synchronization reply
self.sync_socket.send('ack', self.zmq.NOBLOCK)
# ------------------
# Simulation Control
# ------------------
def launch_controller(self, controller):
raise NotImplementedError
def launch_component(self, component):
raise NotImplementedError
class SimulatorBase(ComponentHost):
"""
Simulator coordinates the launch and communication of source, feed, transform, and merge components.
"""
def __init__(self, addresses, gevent_needed=False):
"""
"""
ComponentHost.__init__(self, addresses, gevent_needed)
def simulate(self):
self.run()
def get_id(self):
return "Simulator"
def teardown_component(self, component):
raise NotImplementedError
class ParallelBuffer(Component):
@@ -160,20 +172,30 @@ class ParallelBuffer(Component):
def __init__(self):
Component.__init__(self)
self.sent_count = 0
self.received_count = 0
self.draining = False
#data source component ID -> List of messages
self.data_buffer = {}
self.ds_finished_counter = 0
# Depending on the size of this, might want to use a data
# structure with better asymptotics.
self.data_buffer = {}
def init(self):
pass
@property
def get_id(self):
return "FEED"
def add_source(self, source_id):
self.data_buffer[source_id] = []
@property
def get_type(self):
return COMPONENT_TYPE.CONDUIT
# -------------
# Core Methods
# -------------
def open(self):
self.pull_socket = self.bind_data()
@@ -183,32 +205,66 @@ class ParallelBuffer(Component):
# wait for synchronization reply from the host
socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds.
if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN:
msg = self.control_in.recv()
if self.pull_socket in socks and socks[self.pull_socket] == self.zmq.POLLIN:
message = self.pull_socket.recv()
if message == str(CONTROL_PROTOCOL.DONE):
self.ds_finished_counter += 1
if len(self.data_buffer) == self.ds_finished_counter:
#drain any remaining messages in the buffer
#drain any remaining messages in the buffer
self.drain()
self.signal_done()
else:
event = self.unframe(message)
self.append(event)
self.send_next()
def __len__(self):
"""
Buffer's length is same as internal map holding separate
sorted arrays of events keyed by source id.
"""
return len(self.data_buffer)
try:
event = json.loads(message)
def append(self, event):
# JSON deserialization error
except ValueError as exc:
return self.signal_exception(exc)
try:
self.append(event[u'id'], event)
self.send_next()
# Invalid message
except KeyError as exc:
return self.signal_exception(exc)
# -------------
# Flow Control
# -------------
def drain(self):
"""
Send all messages in the buffer.
"""
self.draining = True
while self.pending_messages() > 0:
self.send_next()
def send_next(self):
"""
Send the (chronologically) next message in the buffer.
"""
if(not(self.is_full() or self.draining)):
return
event = self.next()
if event != None:
self.feed_socket.send(json.dumps(event), self.zmq.NOBLOCK)
self.sent_count += 1
def append(self, source_id, value):
"""
Add an event to the buffer for the source specified by
source_id.
"""
self.data_buffer[event.source_id].append(event)
self.data_buffer[source_id].append(value)
self.received_count += 1
def next(self):
@@ -224,7 +280,7 @@ class ParallelBuffer(Component):
if len(events) == 0:
continue
cur = events
if (earliest == None) or (cur[0].dt <= earliest[0].dt):
if (earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']):
earliest = cur
if earliest != None:
@@ -250,31 +306,18 @@ class ParallelBuffer(Component):
total += len(events)
return total
def drain(self):
def add_source(self, source_id):
"""
Send all messages in the buffer
Add a data source to the buffer.
"""
self.draining = True
while(self.pending_messages() > 0):
self.send_next()
self.data_buffer[source_id] = []
def send_next(self):
def __len__(self):
"""
Send the (chronologically) next message in the buffer.
Buffer's length is same as internal map holding separate
sorted arrays of events keyed by source id.
"""
if(not(self.is_full() or self.draining)):
return
event = self.next()
if(event != None):
self.feed_socket.send(self.frame(event), self.zmq.NOBLOCK)
self.sent_count += 1
def unframe(self, msg):
return zp.DATASOURCE_UNFRAME(msg)
def frame(self, event):
return zp.FEED_FRAME(event)
return len(self.data_buffer)
class MergedParallelBuffer(ParallelBuffer):
@@ -285,6 +328,19 @@ class MergedParallelBuffer(ParallelBuffer):
def __init__(self):
ParallelBuffer.__init__(self)
self.init()
def init(self):
pass
@property
def get_id(self):
return "MERGE"
@property
def get_type(self):
return COMPONENT_TYPE.CONDUIT
def open(self):
self.pull_socket = self.bind_merge()
self.feed_socket = self.bind_result()
@@ -295,58 +351,50 @@ class MergedParallelBuffer(ParallelBuffer):
return
#get the raw event from the passthrough transform.
result = self.data_buffer["PASSTHROUGH"].pop(0).PASSTHROUGH
result = self.data_buffer["PASSTHROUGH"].pop(0)['value']
for source, events in self.data_buffer.iteritems():
if source == "PASSTHROUGH":
continue
if len(events) > 0:
cur = events.pop(0)
result.merge(cur)
result[source] = cur['value']
return result
@property
def get_id(self):
return "MERGE"
def unframe(self, msg):
return zp.TRANSFORM_UNFRAME(msg)
def frame(self, event):
return zp.MERGE_FRAME(event)
#
def append(self, event):
"""
:param event: a namedict with one entry. key is the name of the transform, value is the transformed value.
Add an event to the buffer for the source specified by
source_id.
"""
self.data_buffer[event.__dict__.keys()[0]].append(event)
self.received_count += 1
class BaseTransform(Component):
"""Top level execution entry point for the transform::
"""
Top level execution entry point for the transform
- connects to the feed socket to subscribe to events
- connets to the result socket (most oftened bound by a TransformsMerge) to PUSH transforms
- processes all messages received from feed, until DONE message received
- pushes all transforms
- sends DONE to result socket, closes all sockets and context
- connects to the feed socket to subscribe to events
- connets to the result socket (most oftened bound by a TransformsMerge) to PUSH transforms
- processes all messages received from feed, until DONE message received
- pushes all transforms
- sends DONE to result socket, closes all sockets and context
Parent class for feed transforms. Subclass and override transform
method to create a new derived value from the combined feed."""
method to create a new derived value from the combined feed.
"""
def __init__(self, name):
Component.__init__(self)
self.state = {}
self.state['name'] = name
self.state = {
'name': name
}
self.init()
def init(self):
pass
@property
def get_id(self):
return self.state['name']
@property
def get_type(self):
return COMPONENT_TYPE.CONDUIT
def open(self):
"""
Establishes zmq connections.
@@ -359,21 +407,45 @@ class BaseTransform(Component):
def do_work(self):
"""
Loops until feed's DONE message is received:
- receive an event from the data feed
- call transform (subclass' method) on event
- send the transformed event
- receive an event from the data feed
- call transform (subclass' method) on event
- send the transformed event
"""
socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds.
socks = dict(self.poll.poll(self.heartbeat_timeout))
if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN:
msg = self.control_in.recv()
if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN:
message = self.feed_socket.recv()
if message == str(CONTROL_PROTOCOL.DONE):
self.signal_done()
return
event = zp.FEED_UNFRAME(message)
cur_state = self.transform(event)
qutil.LOGGER.info("state of transform is: {state}".format(state=cur_state))
self.result_socket.send(zp.TRANSFORM_FRAME(cur_state['name'], cur_state['value']), self.zmq.NOBLOCK)
try:
event = json.loads(message)
except ValueError as exc:
return self.signal_exception(exc)
try:
cur_state = self.transform(event)
cur_state['dt'] = event['dt']
cur_state['id'] = self.state['name']
# This is overloaded, so it can fail in all sorts of
# unknown ways. Its best to catch it in the
# Transformer itself.
except Exception as exc:
return self.signal_exception(exc)
try:
json_frame = json.dumps(cur_state)
except ValueError as exc:
return self.signal_exception(exc)
self.result_socket.send(json_frame, self.zmq.NOBLOCK)
def transform(self, event):
"""
@@ -392,25 +464,30 @@ class BaseTransform(Component):
class PassthroughTransform(BaseTransform):
"""
A bypass transform which is also an identity transform::
+-------+
+---| f |--->
+-------+
+------id------->
"""
def __init__(self):
BaseTransform.__init__(self, "PASSTHROUGH")
def do_work(self):
"""
Loops until feed's DONE message is received:
- receive an event from the data feed
- call transform (subclass' method) on event
- send the transformed event
"""
socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds.
if self.feed_socket in socks and socks[self.feed_socket] == self.zmq.POLLIN:
message = self.feed_socket.recv()
if message == str(CONTROL_PROTOCOL.DONE):
self.signal_done()
return
#message is already FEED_FRAMEd, send it as the value.
self.result_socket.send(zp.TRANSFORM_FRAME("PASSTHROUGH", message), self.zmq.NOBLOCK)
self.init()
def init(self):
pass
@property
def get_type(self):
return COMPONENT_TYPE.CONDUIT
def transform(self, event):
return { 'value': event }
class DataSource(Component):
@@ -419,31 +496,38 @@ class DataSource(Component):
means looping through all records in a store, converting to a dict, and
calling send(map).
"""
def __init__(self, source_id, *args):
def __init__(self, source_id):
Component.__init__(self)
self.source_id = source_id
self.id = source_id
self.init()
def init(self):
self.cur_event = None
self.init_ds(args)
def init_ds(*args):
pass
@property
def get_id(self):
return self.source_id
return self.id
@property
def get_type(self):
return COMPONENT_TYPE.SOURCE
def open(self):
#create the data sink. Based on http://zguide.zeromq.org/py:tasksink2
self.data_socket = self.connect_data()
def get_type(self):
raise NotImplementedError
def send(self, event):
"""
Emit data.
"""
assert isinstance(event, dict)
event['id'] = self.id
event['type'] = self.get_type()
try:
json_frame = json.dumps(event)
except ValueError as exc:
return self.signal_exception(exc)
self.data_socket.send(json_frame)
zipline/monitor.py
+76 -18
View File
@@ -1,16 +1,55 @@
import zmq
from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME
class Controller(object):
"""
A broker of sorts.
A N to M messaging system for inter component communication.
Ostensibly a broker of sorts. Putting messages to the broker
is durable, if the broker goes down messages will queue up
until the HWM and then go out when the new broker comes up.
The other end is not durable, it is simply PUB/SUB which has
the benefit of of allowing more fluid time evolution of the
whole system since the messaging passing topology will not
alter itself as a result of more nodes listening.
The actual brokerin' is either a Python loop ( slow ) or a
zmq.FORWARDER device ( fast ).
:param pull_socket: Socket to subscribe to for republication of
published messages. The endpoint for
:func message_sender: .
:param pub_socket: Socket to publish messages, the starting
point of :func message_listener: .
:param logging: Logging interface for tracking broker state
Defaults to None
Usage::
controller = Controller(
'tcp://127.0.0.1:5000',
'tcp://127.0.0.1:5001',
)
# typically you'd want to run this async to your main
# program since it blocks indefinetely.
controller.run()
sub = self.controller.message_listener()
push = self.controller.message_sender()
push.send('DIE')
sub.recv()
"""
polling = False
debug = False
def __init__(self, pull_socket, pub_socket, logging = None):
self._ctx = None
polling = False
self.associated = []
@@ -18,6 +57,7 @@ class Controller(object):
self.push_socket = pull_socket # same port
self.pub_socket = pub_socket
self.sub_socket = pub_socket # same port
self.terminated = False
if logging:
self.logging = logging
@@ -30,6 +70,9 @@ class Controller(object):
self.failed = 0
def run(self, debug=False, context=None):
"""
Run's the loop for the broker.
"""
self.polling = True
if not context:
@@ -37,10 +80,10 @@ class Controller(object):
else:
self._ctx = context
if debug:
return self._poll_fast() # the c loop
else:
return self._poll() # use a python loop
#if not debug:
#return self._poll_fast() # the c loop
#else:
return self._poll() # use a python loop
def _poll_fast(self):
"""
@@ -49,6 +92,9 @@ class Controller(object):
self.pull = self._ctx.socket(zmq.PULL)
self.pub = self._ctx.socket(zmq.PUB)
self.pull.bind(self.pull_socket)
self.pub.bind(self.pub_socket)
zmq.device(zmq.FORWARDER, self.pull, self.pub)
def _poll(self):
@@ -60,15 +106,14 @@ class Controller(object):
self.pull = self._ctx.socket(zmq.PULL)
self.pub = self._ctx.socket(zmq.PUB)
self.associated.extend([self.pull, self.pub])
self.pull.bind(self.pull_socket)
self.pub.bind(self.pub_socket)
self.associated.extend([self.pull, self.pub])
while self.polling:
try:
msg = self.pull.recv()
print msg
self.pub.send(msg)
except KeyboardInterrupt:
self.polling = False
@@ -84,6 +129,8 @@ class Controller(object):
self.failed += 1
continue
self.terminated = True
# -------------------
# Hooks for Endpoints
# -------------------
@@ -117,20 +164,31 @@ class Controller(object):
self.associated.append(s)
return s
def shutdown(self, context=None):
self.polling = False
if not context:
context = zmq.Context()
#logging.info('Shutdown controller')
s = self.message_sender(context)
s.send(CONTROL_FRAME(
'controller',
CONTROL_PROTOCOL.SHUTDOWN,
))
#for asoc in self.associated:
#asoc.close()
def destroy(self):
"""
Manual cleanup.
"""
self.polling = False
self.shutdown()
for asoc in self.associated:
asoc.close()
#if self._ctx:
#self._ctx.destroy()
def __del__(self):
self.destroy()
#def __del__(self):
#self.shutdown()
def qos(self):
if not self.debug:
zipline/protocol.py
+13 -8
View File
@@ -78,13 +78,13 @@ def FrameExceptionFactory(name):
class namedict(object):
"""
So that you can use:
So that you can use::
foo.BAR
-- or --
foo['BAR']
For more complex strcuts use collections.namedtuple:
For more complex structs use collections.namedtuple:
"""
def __init__(self, dct=None):
@@ -130,18 +130,18 @@ CONTROL_PROTOCOL = Enum(
)
def CONTROL_FRAME(id, status):
assert isinstance(basestring, id)
assert isinstance(int, status)
assert isinstance(id, basestring,)
assert isinstance(status, int)
return msgpack.dumps(tuple([id, status]))
def CONTORL_UNFRAME(msg):
assert isinstance(basestring, msg)
assert isinstance(msg, basestring)
try:
id, status = msgpack.loads(msg)
assert isinstance(basestring, id)
assert isinstance(int, status)
assert isinstance(id, basestring)
assert isinstance(status, int)
return id, status
except TypeError:
@@ -167,6 +167,12 @@ HEARTBEAT_PROTOCOL = namedict({
# Component State
# ==================
COMPONENT_TYPE = Enum(
'SOURCE' , # 0
'CONDUIT' , # 1
'SINK' , # 2
)
COMPONENT_STATE = Enum(
'OK' , # 0
'DONE' , # 1
@@ -426,7 +432,6 @@ def UNPACK_DATE(payload):
del(payload.__dict__['micros'])
payload['dt'] = dt
return payload
FINANCE_PROTOCOL = Enum(
'ORDER' , # 0 - req
zipline/serial.py
+91
View File
@@ -0,0 +1,91 @@
"""
Format serializer for Zipline.
Because I'm opinionated about how you should send things over
ZeroMQ. :)
"""
import zlib
#import blosc
import hmac
import base64
import numpy
import pandas
import cPickle as pickle
# Pickle does the equivelant of builtin ``eval``. Be afraid, be
# very afraid.
def send_zipped_pickle(socket, obj, flags=0, protocol=-1):
"""
Pickle an object, and zip the pickle before sending it.
"""
p = pickle.dumps(obj, protocol)
z = zlib.compress(p)
return socket.send(z, flags=flags)
def recv_zipped_pickle(socket, flags=0, protocol=-1):
"""
Unpickle and uncompress a received object.
"""
z = socket.recv(flags)
p = zlib.uncompress(z)
return pickle.loads(p, protocol=protocol)
# HDF5, Numpy Byte Strings, Pandas arrays should use
# blosc and reconstruct the Python container from the byte string
# on the other side.
def send_numpy(socket, obj, flags=0):
packed = blosc.pack_array(obj)
return socket.send(packed, flags=flags)
def recv_numpy(socket, flags=0):
packed = blosc.unpack_array(socket.recv(flags))
return socket.send(packed, flags=flags)
def send_pandas(socket, obj, flags=0):
ndarray = obj._data.blocks[0].values
socket.send_multipart(ndarray, flags=flags)
spec = (
obj._data.index,
obj._data.columns,
obj._data.blocks[0].dtype
)
return socket.send_multipart(spec, flags)
def recv_pandas(socket, flags=0):
ndarray = socket.recv_multipart(flags)
spec = socket.recv_multipart(flags)
return pandas.DataFrame._init_ndarray(ndarray, *spec)
def send_hdf5(self):
pass
def recv_hdf5(self):
pass
# Cryptographically secure wire protocol for ZeroMQ Using HMAC.
# Compare byte strings, backported from Python 3.
def byte_eq(a, b):
return not sum(0 if x==y else 1 for x, y in zip(a, b)) and len(a) == len(b)
def send_secure(socket, data, key, flags=0):
msg = base64.b64encode(data)
sig = base64.b64encode(hmac.new(key, msg).digest())
return socket.send(bytes('!') + sig + bytes('?') + msg, flags=flags)
def recv_secure(socket, data, key, flags):
data = socket.recv(flags=flags)
try:
sig, msg = data.split(bytes('?'), 1)
except ValueError:
raise Exception('Invalid signature/message pair.')
if byte_eq(sig[1:], base64.b64encode(hmac.new(key, msg).digest())):
return base64.b64decode(msg)
else:
raise Exception('Cryptographically invalid message received')
zipline/test/client.py
+42 -33
View File
@@ -1,61 +1,70 @@
import json
#import logging
import ujson as json
import zipline.util as qutil
import zipline.messaging as qmsg
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE
from zipline.finance.trading import TradeSimulationClient
import zipline.protocol as zp
#qlogger = logging.getLogger('qexec')
class TestClient(qmsg.Component):
"""no-op client - Just connects to the merge and counts messages. compares received message count to the expected count."""
def __init__(self, expected_msg_count=0):
qmsg.Component.__init__(self)
self.received_count = 0
# Generally shouldn't reference unit tests here.
self.utest = utest
self.expected_msg_count = expected_msg_count
self.prev_dt = None
self.heartbeat_timeout = 2000
self.init()
def init(self):
self.received_count = 0
self.prev_dt = None
@property
def get_id(self):
return "TEST_CLIENT"
@property
def get_type(self):
return COMPONENT_TYPE.SINK
def open(self):
self.data_feed = self.connect_result()
def do_work(self):
socks = dict(self.poll.poll(self.heartbeat_timeout))
if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN:
msg = self.data_feed.recv()
if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN:
msg = self.control_in.recv()
if msg == str(zp.CONTROL_PROTOCOL.DONE):
if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN:
msg = self.data_feed.recv()
#logger.info('msg:' + str(msg))
if msg == str(CONTROL_PROTOCOL.DONE):
qutil.LOGGER.info("Client is DONE!")
self.signal_done()
if(self.expected_msg_count > 0):
assert self.received_count == self.expected_msg_count
return
self.received_count += 1
event = zp.MERGE_UNFRAME(msg)
if(self.prev_dt != None):
if(not event.dt >= self.prev_dt):
raise Exception("Message out of order: {date} after {prev}".format(date=event['dt'], prev=prev_dt))
self.prev_dt = event.dt
if(self.received_count % 100 == 0):
try:
event = json.loads(msg)
# JSON deserialization error
except ValueError as exc:
return self.signal_exception(exc)
if self.prev_dt != None:
if not event['dt'] >= self.prev_dt:
raise Exception(
"Message out of order: {date} after {prev}".format(
date = event['dt'], prev = self.prev_dt
)
)
else:
self.prev_dt = event['dt']
if self.received_count % 100 == 0:
qutil.LOGGER.info("received {n} messages".format(n=self.received_count))
class TestTradingClient(TradeSimulationClient):
def __init__(self, count):
TradeSimulationClient.__init__(self)
self.count = count
self.incr = 0
def handle_events(self, event_queue):
#place an order for 100 shares of sid:133
if(self.incr < self.count):
self.order(133, 100)
self.incr += 1
zipline/test/test_devsimulator.py
-10
View File
@@ -1,10 +0,0 @@
from unittest2 import TestCase
from zipline.test.test_messaging import SimulatorTestCase
from zipline.test.dummy import ThreadPoolExecutorMixin
class ThreadPoolExecutor(SimulatorTestCase, TestCase):
def test_universe(self):
# first order logic is working today. Yay!
self.assertTrue(True != False)
zipline/test/test_messaging.py
+120 -37
View File
@@ -1,19 +1,74 @@
"""
Test suite for the messaging infrastructure of Zipline.
Test suite for the messaging infrastructure of QSim.
"""
#don't worry about excessive public methods pylint: disable=R0904
import zipline.messaging as qmsg
from collections import defaultdict
from zipline.transforms.technical import MovingAverage
from zipline.sources import RandomEquityTrades
from zipline.test.dummy import ThreadPoolExecutorMixin
from zipline.test.client import TestClient
from zipline.test.transform import DivideByZeroTransform
# Should not inherit form TestCase since test runners will pick
# it up as a test. Its a Mixin of sorts at this point.
class SimulatorTestCase(ThreadPoolExecutorMixin):
class SimulatorTestCase(object):
leased_sockets = defaultdict(list)
def setUp(self):
self.setup_logging()
# TODO: how to make Nose use this cross-process????
self.setup_allocator()
def tearDown(self):
pass
#self.unallocate_sockets()
# Assert the sockets were properly cleaned up
#self.assertEmpty(self.leased_sockets[self.id()].values())
# Assert they were returned to the heap
#self.allocator.socketheap.assert
def get_simulator(self):
"""
Return a new simulator instance to be tested.
"""
raise NotImplementedError
def get_controller(self):
"""
Return a new controler for simulator instance to be tested.
"""
raise NotImplementedError
def setup_allocator(self):
"""
Setup the socket allocator for this test case.
"""
raise NotImplementedError
def allocate_sockets(self, n):
"""
Allocate sockets local to this test case, track them so
we can gc after test run.
"""
assert isinstance(n, int)
assert n > 0
leased = self.allocator.lease(n)
self.leased_sockets[self.id()].extend(leased)
return leased
def unallocate_sockets(self):
self.allocator.reaquire(*self.leased_sockets[self.id()])
# -------
# Cases
# -------
@@ -52,20 +107,68 @@ class SimulatorTestCase(ThreadPoolExecutorMixin):
# Simulation
# ----------
sim.simulate()
sim_context = sim.simulate()
sim_context.join()
# Stop Running
# ------------
# TODO: less abrupt later, just shove a StopIteration
# down the pipe to make it stop spinning
sim.cuc._Thread__stop()
self.assertTrue(sim.ready())
self.assertFalse(sim.exception)
self.assertEqual(sim.feed.pending_messages(), 0,
"The feed should be drained of all messages, found {n} remaining."
.format(n=sim.feed.pending_messages())
)
def test_simplefail(self):
# Simple test just to make sure that the archiecture is
# responding.
# Base Simuation
# --------------
# Allocate sockets for the simulator components
sockets = self.allocate_sockets(5)
addresses = {
'sync_address' : sockets[0],
'data_address' : sockets[1],
'feed_address' : sockets[2],
'merge_address' : sockets[3],
'result_address' : sockets[4]
}
sim = self.get_simulator(addresses)
con = self.get_controller()
# Simulation Components
# ---------------------
ret1 = RandomEquityTrades(133, "ret1", 1)
ret2 = RandomEquityTrades(134, "ret2", 1)
fail_transform = DivideByZeroTransform("fail")
client = TestClient(self, expected_msg_count=ret1.count + ret2.count)
sim.register_controller( con )
sim.register_components([ret1, ret2, fail_transform, client])
# Simulation
# ----------
sim_context = sim.simulate()
sim_context.join()
# Stop Running
# ------------
self.assertTrue(fail_transform.exception)
self.assertFalse(fail_transform.successful())
self.assertEqual(sim.feed.pending_messages(), 0,
"The feed should be drained of all messages, found {n} remaining."
.format(n=sim.feed.pending_messages())
)
def test_sources_only(self):
@@ -98,14 +201,13 @@ class SimulatorTestCase(ThreadPoolExecutorMixin):
# Simulation
# ----------
sim.simulate()
sim_context = sim.simulate()
sim_context.join()
# Stop Running
# ------------
# TODO: less abrupt later, just shove a StopIteration
# down the pipe to make it stop spinning
sim.cuc._Thread__stop()
self.assertTrue(sim.ready())
self.assertFalse(sim.exception)
self.assertEqual(sim.feed.pending_messages(), 0,
"The feed should be drained of all messages, found {n} remaining."
@@ -145,34 +247,15 @@ class SimulatorTestCase(ThreadPoolExecutorMixin):
# Simulation
# ----------
sim.simulate()
sim_context = sim.simulate()
sim_context.join()
# Stop Running
# ------------
# TODO: less abrupt later, just shove a StopIteration
# down the pipe to make it stop spinning
sim.cuc._Thread__stop()
self.assertTrue(sim.ready())
self.assertFalse(sim.exception)
self.assertEqual(sim.feed.pending_messages(), 0,
"The feed should be drained of all messages, found {n} remaining."
.format(n=sim.feed.pending_messages())
)
# TODO used?
def dtest_error_in_feed(self):
ret1 = RandomEquityTrades(133, "ret1", 400)
ret2 = RandomEquityTrades(134, "ret2", 400)
sources = {"ret1":ret1, "ret2":ret2}
mavg1 = MovingAverage("mavg1", 30)
mavg2 = MovingAverage("mavg2", 60)
transforms = {"mavg1":mavg1, "mavg2":mavg2}
client = TestClient(expected_msg_count=0)
sim = self.get_simulator(sources, transforms, client)
# TODO: way too long
sim.feed = DataFeedErr(sources.keys(), sim.data_address, sim.feed_address, sim.performance_address, qmsg.Sync(sim, "DataFeedErrorGenerator"))
sim.simulate()
zipline/test/transform.py
+22
View File
@@ -0,0 +1,22 @@
from zipline.messaging import BaseTransform
from zipline.protocol import COMPONENT_TYPE
class DivideByZeroTransform(BaseTransform):
"""
A transform that fails.
"""
def __init__(self, name):
BaseTransform.__init__(self, "PASSTHROUGH")
self.state['name'] = name
self.init()
def init(self):
pass
@property
def get_type(self):
return COMPONENT_TYPE.CONDUIT
def transform(self, event):
return { 'value': 0/0 }
zipline/topology.py
+80
View File
@@ -0,0 +1,80 @@
"""
Contains the various deployable topologies of ziplines.
This is mostly hardcoded at the moment but as the topologies
becomes more sophisiticated this logic will be the primary
router of sockets.
Ontology of Stream Processing
=============================
Source
******
A producer of data. The data could be in a datastore, coming from a
socket, etc. To access this data, we pull from the source. Sources increase the
total amount of data flowing through the system. Sources are generally not
pure since they involve IO.
Sink
****
A consumer of data. Basic examples would be a sum function (adding up a
stream of numbers fed in), a datastore sink, a socket etc. We push data
into a sink. When / If a sink completes processing, it may return some
value that exists outside of the system.
Sinks decrease the total amount of information flowing through the system.
Conduit
*******
A transformer of data. We push data into a conduit. Similar to a sink,
but instead of returning a single value at the end, a conduit can
return multiple outputs every time it is pushed to. The returned values
remain in the system.
Conduits may or may not be pure, it is usefull to distinguish between the
two since pure conduits have a variety of nice properties under composition
"""
from zipline.protocol import COMPONENT_TYPE
class Topology(object):
pass
class DiamondTopology(Topology):
"""
Exposes a feed, merge, and passthrough bypass::
+--------+
+---------->| |---------------+
| +--------+ |
| v
+---+----+ +---+----+ +--------+ +--------+ +---+----+
| +-->| +----->| |---------->| |--->| |
+---+----+ +---+----+ +--------+ +--------+ +---+----+
| ^
| +--------+ |
+---------->| |---------------+
| +--------+ |
| |
+------------passthru----------------+
"""
flow = {
'flow' : COMPONENT_TYPE.SOURCE ,
'serializers' : COMPONENT_TYPE.CONDUIT ,
'transforms' : COMPONENT_TYPE.CONDUIT ,
'merges' : COMPONENT_TYPE.CONDUIT ,
'clients' : COMPONENT_TYPE.SINK ,
}
def __init__(self):
self.sources = []
self.serializers = []
self.transforms = []
self.merges = []
self.clients = []
zipline/transforms/technical.py
+12 -5
View File
@@ -5,6 +5,7 @@ TODO: add trailing stop
"""
import datetime
from zipline.messaging import BaseTransform
import zipline.util as qutil
@@ -16,18 +17,25 @@ class MovingAverage(BaseTransform):
def __init__(self, name, days):
BaseTransform.__init__(self, name)
self.window = datetime.timedelta(days = days)
self.init()
def init(self):
self.events = []
self.current_total = 0
self.window = datetime.timedelta(days = days)
def transform(self, event):
"""Update the moving average with the latest data point."""
"""
Update the moving average with the latest data point.
"""
self.events.append(event)
self.current_total += event.price
event_date = event.dt
index = 0
for cur_event in self.events:
cur_date = cur_event.dt
if(cur_date - event_date) >= self.window:
@@ -37,11 +45,10 @@ class MovingAverage(BaseTransform):
else:
break
if(len(self.events) == 0):
if len(self.events) == 0:
return 0.0
self.average = self.current_total/len(self.events)
self.state['value'] = self.average
return self.state
return self.state