diff --git a/zipline/__init__.py b/zipline/__init__.py index f47dbf47..60a39553 100644 --- a/zipline/__init__.py +++ b/zipline/__init__.py @@ -1,3 +1,16 @@ """ -QSim provides asynchronous simulation of historic data streams, simulated trade execution, and data stream transformations. -""" \ No newline at end of file +Zipline +""" + +# This is *not* a place to dump arbitrary classes/modules for convenience, +# it is a place to expose the public interfaces. + +import protocol +from core.monitor import Controller +from lines import SimulatedTrading + +__all__ = [ + SimulatedTrading, + Controller, + protocol, +] diff --git a/zipline/core/component.py b/zipline/core/component.py new file mode 100644 index 00000000..d82c8fb9 --- /dev/null +++ b/zipline/core/component.py @@ -0,0 +1,562 @@ +""" +Commonly used messaging components. + +Contains the base class for all components. +""" + +import os +import sys +import uuid +import time +import socket +import gevent +import traceback +import humanhash + +# pyzmq +import zmq +# gevent_zeromq +import gevent_zeromq +# zmq_ctypes +#import zmq_ctypes + +from datetime import datetime + +import zipline.util as qutil +from zipline.gpoll import _Poller as GeventPoller +from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE, \ + COMPONENT_FAILURE, BACKTEST_STATE, CONTROL_FRAME + + +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 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 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): + self.zmq = None + self.context = None + self.addresses = None + + self.out_socket = None + self.killed = False + self.controller = None + # timeout after a full minute + self.heartbeat_timeout = 60 *1000 + self.state_flag = COMPONENT_STATE.OK + self.error_state = COMPONENT_FAILURE.NOFAILURE + self.on_done = None + + self._exception = None + self.fail_time = None + self.start_tic = None + self.stop_tic = None + self.note = None + self.confirmed = False + + # 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): + """ + Open the connections needed to start doing work. + """ + raise NotImplementedError + + def ready(self): + """ + Return ``True`` if and only if the component has finished execution. + """ + 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 init_zmq(self, flavor): + """ + ZMQ in all flavors. Have it your way. + + mp - Distinct contexts | pyzmq + thread - Same context | pyzmq + green - Same context | gevent_zeromq + pypy - Same context | zmq_ctypes + + """ + + if flavor == 'mp': + self.zmq = zmq + self.context = self.zmq.Context() + self.zmq_poller = self.zmq.Poller + return + if flavor == 'thread': + self.zmq = zmq + self.context = self.zmq.Context.instance() + self.zmq_poller = self.zmq.Poller + return + if flavor == 'green': + self.zmq = gevent_zeromq.zmq + self.context = self.zmq.Context.instance() + self.zmq_poller = GeventPoller + return + if flavor == 'pypy': + self.zmq = zmq + self.context = self.zmq.Context.instance() + self.zmq_poller = self.zmq.Poller + return + + raise Exception("Unknown ZeroMQ Flavor") + + def _run(self): + self.start_tic = time.time() + + self.done = False # TODO: use state flag + self.sockets = [] + + self.init_zmq(self.zmq_flavor) + + self.setup_poller() + + self.open() + self.setup_sync() + self.setup_control() + + self.loop() + self.shutdown() + + 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 ues this with care since it makes it very hard to + debug since it mucks up your stacktraces. + """ + + 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() + + def working(self): + """ + Controls when the work loop will start and end + + If we encounter an exception or signal done exit. + + Overload for higher order behavior. + """ + return (not self.done) + + def loop(self, lockstep=True): + """ + Loop to do work while we still have work to do. + """ + while self.working(): + self.confirm() + self.do_work() + + def confirm(self): + """ + Send a synchronization request to the host. + """ + if not self.confirmed: + # TODO: proper framing + self.sync_socket.send(self.get_id + ":RUN") + + self.receive_sync_ack() # blocking + self.confirmed = True + + def runtime(self): + if self.ready() and self.start_tic and self.stop_tic: + 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. + """ + if self.on_done: + self.on_done() + + 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, scope=None): + """ + This is *very* important error tracking handler. + + Will inform the system that the component has failed and + how it has failed. + """ + + if scope == 'algo': + self.error_state = COMPONENT_FAILURE.ALGOEXCEPT + else: + self.error_state = COMPONENT_FAILURE.HOSTEXCEPT + + self.state_flag = COMPONENT_STATE.EXCEPTION + # mark the time of failure so we can track the failure + # progogation through the system. + + self.stop_tic = time.time() + + self._exception = exc + exc_type, exc_value, exc_traceback = sys.exc_info() + trace = '\n>>>'.join(traceback.format_exception(exc_type, exc_value, exc_traceback)) + + exception_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.EXCEPTION, + trace + ) + self.control_out.send(exception_frame) + + qutil.LOGGER.exception("Unexpected error in run for {id}.".format(id=self.get_id)) + + def signal_done(self): + """ + Notify down stream components that we're done. + """ + + self.state_flag = COMPONENT_STATE.DONE + + if self.out_socket: + self.out_socket.send(str(CONTROL_PROTOCOL.DONE)) + + #notify host we're done + # TODO: proper framing + self.sync_socket.send(self.get_id + ":" + str(CONTROL_PROTOCOL.DONE)) + + #notify controller we're done + done_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.DONE, + '' + ) + self.control_out.send(done_frame) + + self.receive_sync_ack() + #notify internal work look that we're done + self.done = True # TODO: use state flag + + qutil.LOGGER.info("[%s] DONE" % self.get_id) + + # ----------- + # Messaging + # ----------- + + def setup_poller(self): + """ + Setup the poller used for multiplexing the incoming data + handling sockets. + """ + + # Initializes the poller class specified by the flavor of + # ZeroMQ. Either zmq.Poller or gpoll.Poller . + self.poll = self.zmq_poller() + + def receive_sync_ack(self): + """ + Wait for synchronization reply from the host. + + DEPRECATED, left in for compatability for now. + """ + + socks = dict(self.sync_poller.poll(self.heartbeat_timeout)) + if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: + message = self.sync_socket.recv() + #else: + #raise Exception("Sync ack timed out on response for {id}".format(id=self.get_id)) + + def bind_data(self): + return self.bind_pull_socket(self.addresses['data_address']) + + def connect_data(self): + return self.connect_push_socket(self.addresses['data_address']) + + def bind_feed(self): + return self.bind_pub_socket(self.addresses['feed_address']) + + def connect_feed(self): + return self.connect_sub_socket(self.addresses['feed_address']) + + def bind_merge(self): + return self.bind_pull_socket(self.addresses['merge_address']) + + def connect_merge(self): + return self.connect_push_socket(self.addresses['merge_address']) + + def bind_result(self): + return self.bind_pub_socket(self.addresses['result_address']) + + def connect_result(self): + return self.connect_sub_socket(self.addresses['result_address']) + + def bind_pull_socket(self, addr): + pull_socket = self.context.socket(self.zmq.PULL) + pull_socket.bind(addr) + self.poll.register(pull_socket, self.zmq.POLLIN) + + self.sockets.append(pull_socket) + + return pull_socket + + def connect_push_socket(self, addr): + push_socket = self.context.socket(self.zmq.PUSH) + push_socket.connect(addr) + #push_socket.setsockopt(self.zmq.LINGER,0) + self.sockets.append(push_socket) + self.out_socket = push_socket + + return push_socket + + 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) + self.out_socket = pub_socket + + return pub_socket + + def connect_sub_socket(self, addr): + sub_socket = self.context.socket(self.zmq.SUB) + sub_socket.connect(addr) + sub_socket.setsockopt(self.zmq.SUBSCRIBE,'') + self.sockets.append(sub_socket) + + self.poll.register(sub_socket, self.zmq.POLLIN) + + return sub_socket + + def setup_control(self): + """ + Set up the control socket. Used to monitor the + overall status of the simulation and to forcefully tear + down the simulation in case of a failure. + """ + + # Allow for the possibility of not having a controller, + # possibly the zipline devsimulator may not want this. + if not self.controller: + return + + self.control_out = self.controller.message_sender( + identity = self.get_id, + 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. ( Connect ) + + DEPRECATED, left in for compatability for now. + """ + + qutil.LOGGER.debug("Connecting sync client for {id}".format(id=self.get_id)) + + self.sync_socket = self.context.socket(self.zmq.REQ) + self.sync_socket.connect(self.addresses['sync_address']) + #self.sync_socket.setsockopt(self.zmq.LINGER,0) + + self.sync_poller = self.zmq_poller() + self.sync_poller.register(self.sync_socket, self.zmq.POLLIN) + + self.sockets.append(self.sync_socket) + + # --------------------- + # 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. + """ + # Prevents the bug that Thomas ran into + raise NotImplementedError + + @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 note(self): + """ + Information about the component. Mostly used for testing. + """ + + def get_note(self): + return self.note or '' + + def debug(self): + """ + Debug information about the component. + """ + 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): + """ + Return a usefull string representation of the component + to indicate its type, unique identifier, and computational + context identifier name. + """ + + return "<{name} {uuid} at {host} {pid} {pointer}>".format( + name = self.get_id , + uuid = self.huid , + host = socket.gethostname() , + pid = os.getpid() , + pointer = hex(id(self)) , + ) diff --git a/zipline/core/messaging.py b/zipline/core/messaging.py new file mode 100644 index 00000000..fd1875c1 --- /dev/null +++ b/zipline/core/messaging.py @@ -0,0 +1,636 @@ +""" +Commonly used messaging components. +""" + +import datetime + +from collections import Counter + +import zipline.util as qutil +from zipline.component import Component +import zipline.protocol as zp +from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \ + COMPONENT_STATE, CONTROL_FRAME, CONTROL_UNFRAME + +class ComponentHost(Component): + """ + Components that can launch multiple sub-components, synchronize their + start, and then wait for all components to be finished. + """ + + def __init__(self, addresses): + Component.__init__(self) + self.addresses = addresses + self.running = False + + self.init() + + def init(self): + assert hasattr(self, 'zmq_flavor'), \ + """ You must specify a flavor of ZeroMQ for all + ComponentHost subclasses. """ + + # Component Registry, keyed by get_id + # ---------------------- + self.components = {} + # ---------------------- + # Internal Registry, keyed by guid + self._components = {} + # ---------------------- + + self.sync_register = {} + self.timeout = datetime.timedelta(seconds=60) + + self.feed = Feed() + self.merge = Merge() + self.passthrough = PassthroughTransform() + self.controller = None + + #register the feed and the merge + 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 + self.controller.zmq_flavor = self.zmq_flavor + + # Propogate the controller to all the subcomponents + for component in self.components.itervalues(): + component.controller = controller + + def register_components(self, components): + """ + Add the given components to the registry. Establish + communication with them. + """ + assert isinstance(components, list) + for component in components: + + component.addresses = self.addresses + component.controller = self.controller + + # Hosts share their zmq flavor with hosted components + component.zmq_flavor = self.zmq_flavor + + self._components[component.guid] = component + self.components[component.get_id] = component + self.sync_register[component.get_id] = datetime.datetime.utcnow() + + if isinstance(component, DataSource): + self.feed.add_source(component.get_id) + if isinstance(component, BaseTransform): + self.merge.add_source(component.get_id) + + def unregister_component(self, component_id): + del self.components[component_id] + del self.sync_register[component_id] + + def setup_sync(self): + """ + Setup the sync socket and poller. ( Bind ) + """ + qutil.LOGGER.debug("Connecting sync server.") + + self.sync_socket = self.context.socket(self.zmq.REP) + self.sync_socket.bind(self.addresses['sync_address']) + + self.sync_poller = self.zmq_poller() + self.sync_poller.register(self.sync_socket, self.zmq.POLLIN) + + self.sockets.append(self.sync_socket) + + def open(self): + for component in self.components.values(): + self.launch_component(component) + self.launch_controller() + + def is_running(self): + """ + DEPRECATED, left in for compatability for now. + """ + + cur_time = datetime.datetime.utcnow() + + if len(self.components) == 0: + qutil.LOGGER.info("Component register is empty.") + return False + + return True + + def loop(self, lockstep=True): + + while self.is_running(): + # wait for synchronization request at start, and DONE at end. + # don't timeout. + socks = dict(self.sync_poller.poll()) + + if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: + msg = self.sync_socket.recv() + + 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.debug("{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() + + #qutil.LOGGER.info("confirmed {id}".format(id=msg)) + # 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 + + def teardown_component(self, component): + raise NotImplementedError + + +class Feed(Component): + """ + Connects to N PULL sockets, publishing all messages received to a PUB + socket. Published messages are guaranteed to be in chronological order + based on message property dt. Expects to be instantiated in one execution + context (thread, process, etc) and run in another. + """ + + def __init__(self): + Component.__init__(self) + + self.sent_count = 0 + self.received_count = 0 + self.draining = False + self.ds_finished_counter = 0 + + # Depending on the size of this, might want to use a data + # structure with better asymptotics. + self.data_buffer = {} + + # source_id -> integer count + self.sent_counters = Counter() + self.recv_counters = Counter() + + def init(self): + pass + + @property + def get_id(self): + return "FEED" + + @property + def get_type(self): + return COMPONENT_TYPE.CONDUIT + + # ------------- + # Core Methods + # ------------- + + def open(self): + self.pull_socket = self.bind_data() + self.feed_socket = self.bind_feed() + + def do_work(self): + # wait for synchronization reply from the host + 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.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 + qutil.LOGGER.debug("draining feed") + self.drain() + self.signal_done() + else: + try: + event = self.unframe(message) + # deserialization error + except zp.INVALID_DATASOURCE_FRAME as exc: + return self.signal_exception(exc) + + try: + self.append(event) + self.send_next() + + # Invalid message + except zp.INVALID_DATASOURCE_FRAME as exc: + return self.signal_exception(exc) + + def unframe(self, msg): + return zp.DATASOURCE_UNFRAME(msg) + + def frame(self, event): + return zp.FEED_FRAME(event) + + # ------------- + # 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(self.frame(event), self.zmq.NOBLOCK) + self.sent_counters[event.source_id] += 1 + self.sent_count += 1 + + def append(self, event): + """ + Add an event to the buffer for the source specified by + source_id. + """ + self.data_buffer[event.source_id].append(event) + self.recv_counters[event.source_id] += 1 + self.received_count += 1 + + def next(self): + """ + Get the next message in chronological order. + """ + if not(self.is_full() or self.draining): + return + + cur_source = None + earliest_source = None + earliest_event = None + #iterate over the queues of events from all sources + #(1 queue per datasource) + for events in self.data_buffer.values(): + if len(events) == 0: + continue + cur_source = events + first_in_list = events[0] + if first_in_list.dt == None: + #this is a filler event, discard + events.pop(0) + continue + + if (earliest_event == None) or (first_in_list.dt <= earliest_event.dt): + earliest_event = first_in_list + earliest_source = cur_source + + if earliest_event != None: + return earliest_source.pop(0) + + def is_full(self): + """ + Indicates whether the buffer has messages in buffer for + all un-DONE, blocking sources. + """ + for source_id, events in self.data_buffer.iteritems(): + if len(events) == 0: + return False + return True + + def pending_messages(self): + """ + Returns the count of all events from all sources in the + buffer. + """ + total = 0 + for events in self.data_buffer.values(): + total += len(events) + return total + + def add_source(self, source_id): + """ + Add a data source to the buffer. + """ + self.data_buffer[source_id] = [] + + 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) + + +class Merge(Feed): + """ + Merges multiple streams of events into single messages. + """ + + def __init__(self): + Feed.__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() + + def next(self): + """Get the next merged message from the feed buffer.""" + if not (self.is_full() or self.draining): + return + + if self.pending_messages() == 0: + return + + # + #get the raw event from the passthrough transform. + result = self.data_buffer[zp.TRANSFORM_TYPE.PASSTHROUGH].pop(0).PASSTHROUGH + for source, events in self.data_buffer.iteritems(): + if source == zp.TRANSFORM_TYPE.PASSTHROUGH: + continue + if len(events) > 0: + cur = events.pop(0) + result.merge(cur) + return result + + 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.keys()[0]].append(event) + self.received_count += 1 + + +class BaseTransform(Component): + """ + Top level execution entry point for the transform + + - connects to the feed socket to subscribe to events + - connects 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. + """ + + def __init__(self, name): + Component.__init__(self) + + 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. + """ + #create the feed. + self.feed_socket = self.connect_feed() + #create the result PUSH + self.result_socket = self.connect_merge() + + 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)) + + # 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: + message = self.feed_socket.recv() + + if message == str(CONTROL_PROTOCOL.DONE): + self.signal_done() + return + + try: + event = self.unframe(message) + except zp.INVALID_FEED_FRAME as exc: + return self.signal_exception(exc) + + try: + cur_state = self.transform(event) + + # 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: + transform_frame = self.frame(cur_state) + except zp.INVALID_TRANSFORM_FRAME as exc: + return self.signal_exception(exc) + + self.result_socket.send(transform_frame, self.zmq.NOBLOCK) + + def frame(self, cur_state): + return zp.TRANSFORM_FRAME(cur_state['name'], cur_state['value']) + + def unframe(self, msg): + return zp.FEED_UNFRAME(msg) + + def transform(self, event): + """ + Must return the transformed value as a map with:: + + {name:"name of new transform", value: "value of new field"} + + Transforms run in parallel and results are merged into a single map, so + transform names must be unique. Best practice is to use the self.state + object initialized from the transform configuration, and only set the + transformed value:: + + self.state['value'] = transformed_value + """ + raise NotImplementedError + + +class PassthroughTransform(BaseTransform): + """ + A bypass transform which is also an identity transform:: + + +-------+ + +---| f |---> + +-------+ + +------id-------> + + """ + + def __init__(self): + BaseTransform.__init__(self, "PASSTHROUGH") + self.init() + + def init(self): + pass + + @property + def get_type(self): + return COMPONENT_TYPE.CONDUIT + + #TODO, could save some cycles by skipping the _UNFRAME call and just setting value to original msg string. + def transform(self, event): + return {'name':zp.TRANSFORM_TYPE.PASSTHROUGH, 'value': zp.FEED_FRAME(event) } + + +class DataSource(Component): + """ + Baseclass for data sources. Subclass and implement send_all - usually this + means looping through all records in a store, converting to a dict, and + calling send(map). + + Every datasource has a dict property to hold filters:: + - key -- name of the filter, e.g. SID + - value -- a primitive representing the filter. e.g. a list of ints. + + Modify the datasource's filters via the set_filter(name, value) + """ + def __init__(self, source_id): + Component.__init__(self) + + self.id = source_id + self.init() + self.filter = {} + + def init(self): + self.cur_event = None + + def set_filter(self, name, value): + self.filter[name] = value + + @property + def get_id(self): + return self.id + + @property + def get_type(self): + return COMPONENT_TYPE.SOURCE + + def open(self): + self.data_socket = self.connect_data() + + def send(self, event): + """ + Emit data. + """ + assert isinstance(event, zp.namedict) + + event['source_id'] = self.get_id + event['type'] = self.get_type + + try: + ds_frame = self.frame(event) + except zp.INVALID_DATASOURCE_FRAME as exc: + return self.signal_exception(exc) + + self.data_socket.send(ds_frame) + + def frame(self, event): + return zp.DATASOURCE_FRAME(event) diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py new file mode 100644 index 00000000..627323ba --- /dev/null +++ b/zipline/core/monitor.py @@ -0,0 +1,622 @@ +import time +import gevent +import itertools +# pyzmq +import zmq +import gevent_zeromq + +from collections import OrderedDict + +from protocol import CONTROL_PROTOCOL, CONTROL_FRAME, \ + CONTROL_UNFRAME, CONTROL_STATES, INVALID_CONTROL_FRAME \ + +states = CONTROL_STATES + +from gpoll import _Poller as GeventPoller + +# Roll Call ( Discovery ) +# ----------------------- +# +# Controller ( 'foo', 'bar', 'fizz', 'pop' ) +# ------------------ +# | | | | +# +---+ +# | 0 | ? ? ? +# +---+ +# | +# IDENTITY: foo +# get message: PROTOCOL.HEARTBEAT +# reply with PROTOCOL.OK +# +# Controller topology = ( 'foo', 'bar', 'fizz', 'pop' ) +# 'foo' in topology = YES -> +# track 'foo' +# ------------------ +# | | | | +# +---+ +# | 1 | ? ? ? +# +---+ + +# Heartbeating +# ------------ +# +# Controller ( time = 2.717828 ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 0 | | 0 | | 0 | | 0 | +# +---+ +---+ +---+ +---+ +# | +# IDENTITY: foo +# get message: time = 2.717828 +# reply with [ foo, 2.71828 ] +# +# Controller ( foo.status = OK ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 1 | | 0 | | 0 | | 0 | +# +---+ +---+ +---+ +---+ +# | +# Controller tracks this node as good +# for this heartbeat + +# Shutdown +# -------- +# +# Controller ( state = RUNNING ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 1 | | 1 | | 1 | | 1 | +# +---+ +---+ +---+ +---+ +# | +# IDENTITY: foo +# send [ DONE ] + +# Controller ( state = SHUTDOWN ) +# Controller topology.remove('foo') +# ------------------ +# | | | +# +---+ +---+ +---+ +---+ +# | | | 1 | | 1 | | 1 | +# +---+ +---+ +---+ +---+ +# | +# IDENTITY: foo +# yield, stop sending messages + +# Termination +# ------------ +# +# Controller ( state = TERMINATE ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 1 | | 1 | | 1 | | 1 | +# +---+ +---+ +---+ +---+ +# | +# get message PROTOCOL.KILL + +# Controller ( state = TERMINATE ) +# ------------------ +# | | | | +# +---+ +---+ +---+ +---+ +# | 0 | | 0 | | 0 | | 0 | +# +---+ +---+ +---+ +---+ + +INIT, SOURCES_READY, RUNNING, TERMINATE = CONTROL_STATES + +state_transitions = frozenset([ + (-1 , INIT), + (INIT , SOURCES_READY), + (SOURCES_READY , RUNNING), + (INIT , TERMINATE), + (SOURCES_READY , TERMINATE), + (RUNNING , TERMINATE), +]) + +class UnknownChatter(Exception): + def __init__(self, name): + self.named = name + def __str__(self): + return """Component calling itself "%s" talking on unexpected channel"""\ + % self.named + +class Controller(object): + """ + A N to M messaging system for inter component communication. + + :param pub_socket: Socket to publish messages, the starting + point of :func message_listener: . + + :param route_socket: Socket to listen for status updates for + the individual components. + :func message_sender: . + + :param logging: Logging interface for tracking broker state + Defaults to None + + Topology is the set of components we expect to show up. + States are the transitions the sytems go through. The + simplest is from RUNNING -> NOT RUNNING . + + 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.manage( + [ TOPOLOGY ] + [ STATES ] + ) + + """ + + debug = False + period = 1 + + def __init__(self, pub_socket, route_socket, logging = None): + + self.context = None + self.zmq = None + self.zmq_poller = None + + self.running = False + self.polling = False + self.tracked = set() + self.responses = set() + + self.ctime = 0 + self.tic = time.time() + self.freeform = False + self._state = -1 + + self.associated = [] + + self.pub_socket = pub_socket + self.route_socket = route_socket + + self.error_replay = OrderedDict() + + if logging: + self.logging = logging + else: + import util as qutil + self.logging = qutil.LOGGER + + def init_zmq(self, flavor): + + assert self.zmq_flavor in ['thread', 'mp', 'green'] + + if flavor == 'mp': + self.zmq = zmq + self.context = self.zmq.Context() + self.zmq_poller = self.zmq.Poller + return + if flavor == 'thread': + self.zmq = zmq + self.context = self.zmq.Context.instance() + self.zmq_poller = self.zmq.Poller + return + if flavor == 'green': + self.zmq = gevent_zeromq.zmq + self.context = self.zmq.Context.instance() + self.zmq_poller = GeventPoller + return + if flavor == 'pypy': + self.zmq = zmq + self.context = self.zmq.Context.instance() + self.zmq_poller = self.zmq.Poller + return + + def manage(self, topology, states=None, context=None): + """ + Give the controller a set set of components to manage and + a set of state transitions for the entire system. + """ + + # A freeform topology is where we heartbeat with anything + # that shows up. + if topology == 'freeform': + self.freeform = True + self.topology = frozenset([]) + else: + self.freeform = False + self.topology = frozenset(topology) + + self.polling = True + self.state = CONTROL_STATES.INIT + + @property + def state(self): + return self._state + + @state.setter + def state(self, new): + old, self._state = self._state, new + + if (old, new) not in state_transitions: + raise RuntimeError("[Controller] Invalid State Transition : %s -> %s" %(old, new)) + else: + self.logging.info("[Controller] State Transition : %s -> %s" %(old, new)) + + def run(self): + self.running = True + self.init_zmq(self.zmq_flavor) + + try: + return self._poll() # use a python loop + except KeyboardInterrupt: + self.logging.info('Shutdown event loop') + + def log_status(self): + """ + Snapshot of the tracked components at every period. + """ + #self.logging.info("[Controller] Tracking : %s" % ([c for c in self.tracked],)) + pass + + def replay_errors(self): + """ + Replay the errors in the order they were reported to the + controller. + """ + return [ a for a in sorted(self.replay_errors.keys())] + + # ------------- + # Publications + # ------------- + + def send_heart(self): + if not self.running: + return + + heartbeat_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.HEARTBEAT, + str(self.ctime) + ) + self.pub.send(heartbeat_frame) + + def send_hardkill(self): + if not self.running: + return + + kill_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.KILL, + '' + ) + self.pub.send(kill_frame) + + def send_softkill(self): + if not self.running: + return + + soft_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.SHUTDOWN, + '' + ) + self.pub.send(soft_frame) + + # ----------- + # Event Loops + # ----------- + + def _poll(self): + + assert self.route_socket + assert self.pub_socket + assert self.cancel_socket + + # -- Publish -- + # ============= + self.pub = self.context.socket(self.zmq.PUB) + self.pub.bind(self.pub_socket) + + # -- Cancel -- + # ============= + assert isinstance(self.cancel_socket,basestring), self.cancel_socket + self.cancel = self.context.socket(self.zmq.REP) + self.cancel.connect(self.cancel_socket) + + # -- Router -- + # ============= + self.router = self.context.socket(self.zmq.ROUTER) + self.router.bind(self.route_socket) + + + poller = self.zmq.Poller() + poller.register(self.router, self.zmq.POLLIN) + poller.register(self.cancel, self.zmq.POLLIN) + + self.associated += [self.pub, self.router, self.cancel] + + # TODO: actually do this + self.state = CONTROL_STATES.SOURCES_READY + + buffer = [] + + for i in itertools.count(0): + self.log_status() + self.responses = set() + + self.ctime = time.time() + self.send_heart() + + while self.polling: + # Reset the responses for this cycle + + socks = dict(poller.poll(self.period)) + 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: + 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 + + self.beat() + + if self.zmq_flavor == 'green': + gevent.sleep(0) + + if self.state is CONTROL_STATES.TERMINATE: + break + + if not self.polling: + break + + # After loop exits + self.terminated = True + + def beat(self): + + # These the set overloaded operations + # A & B ~ set.intersection + # A - B ~ set.difference + + # * good - Components we are currently tracking and who just sent + # us back the right response. + # * bad - Components we are currently tracking but who did not + # send us back a response. + # * new - Components we haven't heard from yet, but sent back the + # right response. + + good = self.tracked & self.responses + bad = self.tracked - good + new = self.responses - good + + for component in new: + self.new(component) + + for component in bad: + self.fail(component) + + # -------------- + # Init Handlers + # -------------- + + def new_source(self): + if self.state is CONTROL_STATES.RUNNING: + self.state = SOURCES_READY + + def new_universal(self): + pass + + # The various "states of being that a component can inform us + # of + def new(self, component): + if self.state is CONTROL_STATES.TERMINATE: + return + + self.logging.info('[Controller] Now Tracking "%s" ' % component) + + universal = self.new_universal + init_handlers = { + 'FEED' : self.new_source, + } + + if component in self.topology or self.freeform: + init_handlers.get(component, universal)() + self.tracked.add(component) + else: + # Some sort of socket collision has occured, this is + # a very bad failure mode. + raise UnknownChatter(component) + + # ------------------ + # Epic Fail Handling + # ------------------ + + def fail_universal(self): + pass + # TODO: this requires higher order functionality + #self.logging.error('[Controller] System in exception state, shutting down') + #self.shutdown(soft=True) + + def fail(self, component): + if self.state is CONTROL_STATES.TERMINATE: + return + + universal = self.fail_universal + fail_handlers = { } + + if component in self.topology or self.freeform: + self.logging.info('[Controller] Component "%s" timed out' % component) + self.tracked.remove(component) + fail_handlers.get(component, universal)() + + # ------------------- + # Completion Handling + # ------------------- + + def done(self, component): + self.logging.info('[Controller] Component "%s" done.' % component) + + # -------------- + # Error Handling + # -------------- + + def exception_universal(self): + """ + Shutdown the system on failure. + """ + self.logging.error('[Controller] System in exception state, shutting down') + self.shutdown(soft=True) + + def exception(self, component, failure): + universal = self.exception_universal + exception_handlers = { } + + if component in self.topology or self.freeform: + self.error_replay[(component, time.time())] = failure + self.logging.error('[Controller] Component "%s" in exception state' % component) + + exception_handlers.get(component, universal)() + else: + raise UnknownChatter(component) + + # ----------------- + # Protocol Handling + # ----------------- + + def handle_recv(self, msg): + """ + Check for proper framing at the transport layer. + Seperates the proper frames from anything else that might + be coming over the wire. Which shouldn't happen ... right? + """ + identity = msg[0] + id, status = CONTROL_UNFRAME(msg[1]) + + # A component is telling us its alive: + if id is CONTROL_PROTOCOL.OK: + + if status == str(self.ctime): + self.responses.add(identity) + else: + # Otherwise its something weird and we don't know + # what to do so just say so + self.logging.error("Weird stuff happened: %s" % msg) + + # A component is telling us it failed, and how + if id is CONTROL_PROTOCOL.EXCEPTION: + self.exception(identity, status) + + # A component is telling us its done with work and won't + # be talking to us anymore + if id is CONTROL_PROTOCOL.DONE: + self.done(identity) + + # ------------------- + # Hooks for Endpoints + # ------------------- + + # These are all connects so no complex allocation logic is + # needed. Dealers and Subscribers can all come and go as a + # function of time without impacting flow of the whole + # system. + + def message_sender(self, identity, context = None): + """ + Spin off a socket used for sending messages to this + controller. + """ + + if not context: + context = self.zmq.Context.instance() + + s = context.socket(zmq.DEALER) + s.setsockopt(zmq.IDENTITY, identity) + s.connect(self.route_socket) + + self.associated.append(s) + return s + + def message_listener(self, context = None): + """ + Spin off a socket used for receiving messages from this + controller. + """ + + if not context: + context = self.zmq.Context.instance() + + s = context.socket(zmq.SUB) + s.connect(self.pub_socket) + s.setsockopt(zmq.SUBSCRIBE, '') + + self.associated.append(s) + return s + + def do_error_replay(self): + for (component, time), error in self.error_replay.iteritems(): + self.logging.info('[Controller] Error Log for -- %s --:\n%s' % + (component, error)) + + def shutdown(self, hard=False, soft=True, context=None): + + if not self.polling: + return + + self.polling = False + + assert hard or soft, """ Must specify kill hard or soft """ + + if hard: + self.state = CONTROL_STATES.TERMINATE + + self.logging.info('[Controller] Hard Shutdown') + + #for asoc in self.associated: + #asoc.close() + + if soft: + self.state = CONTROL_STATES.TERMINATE + + self.logging.info('[Controller] Soft Shutdown') + self.send_softkill() + + #for asoc in self.associated: + #asoc.close() + + self.do_error_replay() + +if __name__ == '__main__': + + print 'Running on '\ + 'tcp://127.0.0.1:5000 '\ + 'tcp://127.0.0.1:5001 ' + + controller = Controller( + 'tcp://127.0.0.1:5000', + 'tcp://127.0.0.1:5001', + ) + controller.zmq_flavor = 'green' + + controller.manage( + 'freeform', + [] + ) + controller.run() diff --git a/zipline/sources.py b/zipline/finance/sources.py similarity index 100% rename from zipline/sources.py rename to zipline/finance/sources.py diff --git a/zipline/profile/__init__.py b/zipline/profile/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/zipline/profile/prof.py b/zipline/profile/prof.py new file mode 100644 index 00000000..d292b300 --- /dev/null +++ b/zipline/profile/prof.py @@ -0,0 +1,104 @@ +""" + +Viscosity - Tools for benchmarking ZeroMQ data flow. + +""" + +import time as timer +import logging +import pycounters +from contextlib import contextmanager, nested +from pycounters import base +from pycounters.shortcuts import frequency, time +from pycounters import shortcuts, reporters, start_auto_reporting, register_reporter +from pycounters import shortcuts,reporters,report_value, output_report, \ +counters, register_counter, _reporting_decorator_context_manager + +JSONFile = "counters.json" + +logger = logging.getLogger('simple_example') +logger.setLevel(logging.DEBUG) + +ch = logging.StreamHandler() +ch.setLevel(logging.DEBUG) +logger.addHandler(ch) + +reporter = reporters.JSONFileReporter(output_file=JSONFile) +logreport = reporters.LogReporter(logger) +register_reporter(logreport) +register_reporter(reporter) + +class timecontext: + + def __init__(self, name): + self.name = name + + def __enter__(self): + cntr = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False) + if not cntr: + counter = counters.AverageTimeCounter(self.name) + register_counter(counter) + self.tic = timer.time() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if not exc_type: + shortcuts.value(self.name, timer.time() - self.tic) + +class ttimecontext: + + def __init__(self, name): + self.name = name + + def __enter__(self): + counter = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False) + + if not counter: + counter = counters.EventCounter(self.name) + counter.value = 0 + register_counter(counter) + + self.counter = counter + self.tic = timer.time() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if not exc_type: + val = (timer.time() - self.tic) + if not self.counter.value: + self.counter.value = long(0.0) + self.counter.value += val + +class occurancecontext: + + def __init__(self, name): + self.name = name + + def __enter__(self): + cntr = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False) + if not cntr: + cntr = counters.TotalCounter(self.name) + counter = counters.TotalCounter(self.name) + register_counter(counter) + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + shortcuts.value(self.name, 1) + +if __name__ == '__main__': + + with timecontext('average time'): + for i in xrange(5): + x = [2] * 1000 + timer.sleep(0.01) + + with occurancecontext('totalcount'): + for i in xrange(5): + x = [2] * 1000 + + with ttimecontext('total time'): + for i in xrange(5): + x = [2] * 1000 + timer.sleep(1) + + pycounters.output_report() diff --git a/zipline/topology.py b/zipline/topology.py deleted file mode 100644 index b5b92125..00000000 --- a/zipline/topology.py +++ /dev/null @@ -1,80 +0,0 @@ -""" -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 = [] diff --git a/zipline/toys/__init__.py b/zipline/toys/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/zipline/date_utils.py b/zipline/utils/date_utils.py similarity index 100% rename from zipline/date_utils.py rename to zipline/utils/date_utils.py diff --git a/zipline/gpoll.py b/zipline/utils/gpoll.py similarity index 100% rename from zipline/gpoll.py rename to zipline/utils/gpoll.py diff --git a/zipline/util.py b/zipline/utils/logging.py similarity index 100% rename from zipline/util.py rename to zipline/utils/logging.py diff --git a/zipline/utils/protocol_utils.py b/zipline/utils/protocol_utils.py new file mode 100644 index 00000000..60c90814 --- /dev/null +++ b/zipline/utils/protocol_utils.py @@ -0,0 +1,221 @@ +import copy +import pandas +from ctypes import Structure, c_ubyte +from collections import MutableMapping +from itertools import izip + +def Enum(*options): + """ + Fast enums are very important when we want really tight zmq + loops. These are probably going to evolve into pure C structs + anyways so might as well get going on that. + """ + class cstruct(Structure): + _fields_ = [(o, c_ubyte) for o in options] + __iter__ = lambda s: iter(range(len(options))) + return cstruct(*range(len(options))) + +def FrameExceptionFactory(name): + """ + Exception factory with a closure around the frame class name. + """ + class InvalidFrame(Exception): + def __init__(self, got): + self.got = got + + def __str__(self): + return "Invalid {framecls} Frame: {got}".format( + framecls = name, + got = self.got, + ) + + return InvalidFrame + +class namedict(MutableMapping): + """ + + Namedicts are dict like objects that have fields accessible by attribute lookup + as well as being indexable and iterable:: + + HEARTBEAT_PROTOCOL = namedict({ + 'REQ' : b'\x01', + 'REP' : b'\x02', + }) + + HEARTBEAT_PROTOCOL.REQ # syntactic sugar + HEARTBEAT_PROTOCOL.REP # oh suga suga + + For more complex structs use collections.namedtuple: + """ + + def __init__(self, dct=None): + if(dct): + self.__dict__.update(dct) + + def __setitem__(self, key, value): + """ + Required for use by pymongo as_class parameter to find. + """ + if(key == '_id'): + self.__dict__['id'] = value + else: + self.__dict__[key] = value + + def __getitem__(self, key): + return self.__dict__[key] + + def __delitem__(self, key): + del self.__dict__[key] + + def __iter__(self): + return self.__dict__.iterkeys() + + def __len__(self): + return len(self.__dict__) + + def keys(self): + return self.__dict__.keys() + + def as_dict(self): + # shallow copy is O(n) + return copy.copy(self.__dict__) + + def delete(self, key): + del(self.__dict__[key]) + + def merge(self, other_nd): + assert isinstance(other_nd, namedict) + self.__dict__.update(other_nd.__dict__) + + def __repr__(self): + return "namedict: " + str(self.__dict__) + + def __eq__(self, other): + # !!!!!!!!!!!!!!!!!!!! + # !!!! DANGEROUS !!!!! + # !!!!!!!!!!!!!!!!!!!! + return other != None and self.__dict__ == other.__dict__ + + def has_attr(self, name): + return self.__dict__.has_key(name) + + def as_series(self): + s = pandas.Series(self.__dict__) + s.name = self.sid + return s + +class ndict(MutableMapping): + """ + Xtreme Namedicts 2.0 + + Ndicts are dict like objects that have fields accessible by attribute + lookup as well as being indexable and iterable. Done right + this time. + """ + + def __init__(self, dct=None): + self.__internal = dict() + self.cls = frozenset(dir(self)) + + if dct: + self.__internal.update(dct) + + # Abstact Overloads + # ----------------- + + def __setitem__(self, key, value): + """ + Required for use by pymongo as_class parameter to find. + """ + if key == '_id': + self.__internal['id'] = value + else: + self.__internal[key] = value + + + def __getattr__(self, key): + if key in self.cls: + return self.__dict__[key] + else: + return self.__internal[key] + + def __getitem__(self, key): + return self.__internal[key] + + def __delitem__(self, key): + del self.__internal[key] + + def __iter__(self): + return self.__internal.iterkeys() + + def __len__(self): + return len(self.__internal) + + # Compatability with namedicts + # ---------------------------- + + # for compat, not the Python way to do things though... + # Deprecated, use builtin ``del`` operator. + delete = __delitem__ + + def has_attr(self, key): + """ + Deprecated, use builtin ``in`` operator. + """ + return self.__contains__(key) + + def has_key(self, key): + return self.__contains__(key) + + # Custom Methods + # -------------- + + def copy(self): + return ndict(copy.copy(self.__internal)) + + def as_dataframe(self): + """ + Return the representation as a Pandas dataframe. + """ + d = pandas.DataFrame(self.__internal) + return d + + def as_series(self): + """ + Return the representation as a Pandas time series. + """ + s = pandas.Series(self.__internal) + s.name = self.sid + return s + + def as_dict(self): + """ + Return the representation as a vanilla Python dict. + """ + # shallow copy is O(n) + return copy.copy(self.__internal) + + def merge(self, other_nd): + """ + Merge in place with another ndict. + """ + assert isinstance(other_nd, ndict) + self.__internal.update(other_nd.__internal) + + def __repr__(self): + return "namedict: " + str(self.__internal) + + # Faster dictionary comparison? + #def __eq__(self, other): + #assert isinstance(other, ndict) + + #keyeq = set(self.keys()) == set(other.keys()) + + #if not keyeq: + #return False + + #for i, j in izip(self.itervalues(), other.itervalues()): + #if i != j: + #return False + + #return True diff --git a/zipline/serial.py b/zipline/utils/serial.py similarity index 100% rename from zipline/serial.py rename to zipline/utils/serial.py diff --git a/zipline/zmq_utils.py b/zipline/utils/zmq_utils.py similarity index 100% rename from zipline/zmq_utils.py rename to zipline/utils/zmq_utils.py diff --git a/zipline/version.py b/zipline/version.py new file mode 100644 index 00000000..fc9fc57e --- /dev/null +++ b/zipline/version.py @@ -0,0 +1,9 @@ +BANNER = """ +Zipline {version} +Released under BSD3 +""".strip() + +VERSION = ( 0, 0, 1, 'dev' ) + +def pretty_version(): + return BANNER.format(version='.'.join(VERSION))