diff --git a/zipline/core/controlled.py b/zipline/core/controlled.py deleted file mode 100644 index 19073fa1..00000000 --- a/zipline/core/controlled.py +++ /dev/null @@ -1,74 +0,0 @@ -""" -Poller logic for a component which is controlled by the monitor, this is -largely universal and thus we break it out into a seperate module and -splice it into the dispatch loops for each component instance. - -Example usage:: - - def do_work(): - socks = self.poll.poll() - - # Handle control events - do_handle_control_events() - - # Handle other events - if socks.get(socket) == zmq.POLLIN: - ... -""" - -import zmq -from zipline.core.component import Component -from zipline.protocol import CONTROL_PROTOCOL, CONTROL_FRAME, CONTROL_UNFRAME - -def do_handle_control_events(cls, poller): - assert isinstance(cls, Component) - assert cls.control_in, 'Component does not have a control_in socket' - - # If we're in devel mode drop out because the controller - # isn't guaranteed to be around anymore - if cls.devel: - return - - if poller.get(cls.control_in) == zmq.POLLIN: - msg = cls.control_in.recv() - event, payload = CONTROL_UNFRAME(msg) - - # =========== - # Heartbeat - # =========== - - # The controller will send out a single number packed in - # a CONTROL_FRAME with ``heartbeat`` event every - # (n)-seconds. The component then has n seconds to - # respond to it. If not then it will be considered as - # malfunctioning or maybe CPU bound. - - if event == CONTROL_PROTOCOL.HEARTBEAT: - # Heart outgoing - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.OK, - payload - ) - # Echo back the heartbeat identifier to tell the - # controller that this component is still alive and - # doing work - cls.control_out.send(heartbeat_frame) - - # ========= - # Soft Kill - # ========= - - # Try and clean up properly and send out any reports or - # data that are done during a clean shutdown. Inform the - # controller that we're done. - elif event == CONTROL_PROTOCOL.SHUTDOWN: - cls.signal_done() - cls.shutdown() - - # ========= - # Hard Kill - # ========= - - # Just exit. - elif event == CONTROL_PROTOCOL.KILL: - cls.kill() diff --git a/zipline/core/host.py b/zipline/core/host.py deleted file mode 100644 index 37de82aa..00000000 --- a/zipline/core/host.py +++ /dev/null @@ -1,143 +0,0 @@ -import os -import sys -import logbook - -from zipline.transforms import BaseTransform -from zipline.components import Feed, Merge, PassthroughTransform, \ - DataSource -from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE - -log = logbook.Logger('Topology') - -class ComponentHost(object): - """ - Components that can launch multiple sub-components, synchronize - their start, and then wait for all components to be finished. - """ - - def __init__(self, addresses): - self.addresses = addresses - self.running = False - - # Component Registry, keyed by unique string - # ---------------------- - self.components = {} - # ---------------------- - # Internal Registry, keyed by guid - self._components = {} - # ---------------------- - - self.exception = None - - self.feed = Feed() - self.merge = Merge() - self.passthrough = PassthroughTransform() - self.controller = None - - self.register_components([self.feed, self.merge, self.passthrough]) - - def _run(self): - self.open() - - def run(self, catch_exceptions=True): - """ - Run the host. - """ - log.info('===== PARENT PID: %s' % os.getppid()) - log.info('===== PID: %s' % os.getpid()) - - self.open() - #self.shutdown() - - def shutdown(self, ensure_clean=True): - raise NotImplementedError - - 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 - - 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] - - @property - def pids(self): - return [proc.pid for proc in self.subprocesses] - - def open(self): - assert hasattr(self, 'zmq_flavor'), \ - """ You must specify a flavor of ZeroMQ for all Topology - subclasses. """ - - - log.info('== Roll Call ==') - log.info('Monitor') - - self.launch_controller() - - for component in self.components.itervalues(): - log.info(component) - - log.info('== End Roll Call ==') - - for component in self.components.itervalues(): - self.launch_component(component) - - - def is_running(self): - """ - DEPRECATED, left in for compatability for now. - """ - - if len(self.components) == 0: - log.info("Component register is empty.") - return False - - return True - - def ready(self): - return True - - # ------------------ - # Simulation Control - # ------------------ - - # Overloaded by simulator - - def launch_controller(self, controller): - raise NotImplementedError - - def launch_component(self, component): - raise NotImplementedError diff --git a/zipline/core/process.py b/zipline/core/process.py deleted file mode 100644 index dc1fcd3c..00000000 --- a/zipline/core/process.py +++ /dev/null @@ -1,90 +0,0 @@ -""" -The process simulator. Each component in a separate -multiprocessing.process. -""" - -import logbook -import multiprocessing -from zipline.core.host import ComponentHost - -log = logbook.Logger('Process Simulator') - -class ProcessSimulator(ComponentHost): - """ - The process simulator. - """ - - zmq_flavor = 'mp' - - def __init__(self, addresses): - ComponentHost.__init__(self, addresses) - self.subprocesses = [] - self.running = False - self.mapping = {} - - def define(self, key, val): - """ - Returns the mapping between a component and its - pid. - """ - self.mapping[key] = val - - @property - def get_id(self): - return 'Multiprocess Simulator' - - # ========= - # Launchers - # ========= - # - # invoked by the host's open() - - def launch_controller(self): - proc = multiprocessing.Process(target=self.monitor.run) - proc.start() - self.con = proc - - # Process specific - self.monitor_process = proc - self.mapping[proc.pid] = 'Monitor' - - def launch_component(self, component): - proc = multiprocessing.Process(target=component.run) - proc.start() - self.subprocesses.append(proc) - - self.mapping[proc.pid] = component.get_id - return proc - - def simulate(self): - """ - Kick off the simulation - """ - self.run() - - def did_clean_shutdown(self): - cleanly = not any([s.is_alive() for s in self.subprocesses]) - if not cleanly: - for process in self.subprocesses: - if process.is_alive(): - log.error('Failed to Yield', self.mapping[process.pid]) - return cleanly - - def shutdown(self, ensure_clean=True): - """ - Shutdown the simulation. - """ - for component in self.components.itervalues(): - component.shutdown() - - for process in self.subprocesses: - process.join(timeout=1) - process.terminate() - - self.monitor.shutdown(soft=True) - self.running = False - - self.con.terminate() - - if ensure_clean: - assert self.did_clean_shutdown() diff --git a/zipline/finance/movingaverage.py b/zipline/finance/movingaverage.py deleted file mode 100644 index f0e401fe..00000000 --- a/zipline/finance/movingaverage.py +++ /dev/null @@ -1,72 +0,0 @@ -from datetime import timedelta -from collections import defaultdict - -from zipline.transforms.base import BaseTransform - -class MovingAverageTransform(BaseTransform): - - - def init(self, name, days=3): - self.state = {} - self.state['name'] = name - self.days = days - self.by_sid = defaultdict(self._create) - - @property - def get_id(self): - return self.state['name'] - - def transform(self, event): - cur = self.by_sid[event.sid] - cur.update(event) - self.state['value'] = cur.average - return self.state - - def _create(self): - return MovingAverage(self.days) - -class MovingAverage(object): - - def __init__(self, days): - self.window = EventWindow(days) - self.total = 0.0 - self.average = 0.0 - - def update(self, event): - self.window.update(event) - - self.total += event.price - - for dropped in self.window.dropped_ticks: - self.total -= dropped.price - - if len(self.window.ticks) > 0: - self.average = self.total / len(self.window.ticks) - else: - self.average = 0.0 - -class EventWindow(object): - """ - Tracks a window of the event history. Use an instance to track the events - inside your window to efficiently calculate rolling statistics. - """ - def __init__(self, days): - self.ticks = [] - self.dropped_ticks = [] - self.delta = timedelta(days=days) - - def update(self, event): - # add new event - self.ticks.append(event) - # determine which events are expired - last_date = event['dt'] - first_date = last_date - self.delta - - self.dropped_ticks = [] - for tick in self.ticks: - if tick['dt'] <= first_date: - self.dropped_ticks.append(tick) - - # remove the expired events - slice_index = len(self.dropped_ticks) - self.ticks = self.ticks[slice_index:] diff --git a/zipline/finance/sources.py b/zipline/finance/sources.py deleted file mode 100644 index b9f818ea..00000000 --- a/zipline/finance/sources.py +++ /dev/null @@ -1,119 +0,0 @@ -""" -Provides data handlers that can push messages to a zipline.core.DataFeed - -:: - DataSource - | - TradeDataSource - / \ - RandomEquityTrades SpecificEquityTrades - -""" -import pytz -import random -import datetime -from mock import Mock - -from zipline.components import DataSource -from zipline.utils import ndict - -import zipline.protocol as zp - -class TradeDataSource(DataSource): - - def init(self): - self.setup_source() - - #@property - #def get_id(self): - # return 'TradeDataSource' - - def send(self, event): - """ - Sends the event iff it matches the internal sid filter. - :param dict event: is a trade event with data as per - :py:func: `zipline.protocol.TRADE_FRAME` - :rtype: None - """ - - event.source_id = self.get_id - - if event.sid in self.filter['sid']: - message = zp.DATASOURCE_FRAME(event) - self.data_socket.send(message) - - -class RandomEquityTrades(TradeDataSource): - """ - Generates a random stream of trades for testing. - """ - - def init(self, sid, count): - self.count = count - self.incr = 0 - self.sid = sid - self.trade_start = datetime.datetime.now().replace(tzinfo=pytz.utc) - self.day = datetime.timedelta(days=1) - self.price = random.uniform(5.0, 50.0) - - self.setup_source() - - @property - def get_id(self): - return 'RandomEquityTrades' - - def do_work(self): - if not self.incr < self.count: - self.signal_done() - return - - self.price = self.price + random.uniform(-0.05, 0.05) - volume = random.randrange(100,10000,100) - - event = zp.ndict({ - "type" : zp.DATASOURCE_TYPE.TRADE, - "sid" : self.sid, - "price" : self.price, - "volume" : volume, - "dt" : self.trade_start + (self.day * self.incr), - }) - self.send(event) - self.incr += 1 - - -class SpecificEquityTrades(TradeDataSource): - """ - Generates a non-random stream of trades for testing. - """ - - def init(self, event_list): - """ - :param event_list: should be a chronologically ordered list of - dictionaries in the following form:: - - event = { - 'sid' : an integer for security id, - 'dt' : datetime object, - 'price' : float for price, - 'volume' : integer for volume - } - """ - self.event_list = event_list - self.count = 0 - - # TODO temporary hack - self.control_out = Mock() - self.setup_source() - - @property - def get_id(self): - return "SpecificEquityTrades" - - def do_work(self): - if(len(self.event_list) == 0): - self.signal_done() - return - - event = self.event_list.pop(0) - self.send(zp.ndict(event)) - self.count +=1 diff --git a/zipline/finance/vwap.py b/zipline/finance/vwap.py deleted file mode 100644 index 09941fe1..00000000 --- a/zipline/finance/vwap.py +++ /dev/null @@ -1,68 +0,0 @@ -from datetime import timedelta -from collections import defaultdict - -from zipline.transforms.base import BaseTransform -from zipline.finance.movingaverage import EventWindow - -class VWAPTransform(BaseTransform): - - def init(self, name, daycount=3): - self.props = {} - self.props['name'] = name - self.daycount = daycount - self.by_sid = defaultdict(self.create_vwap) - - @property - def get_id(self): - return self.props['name'] - - def transform(self, event): - cur = self.by_sid[event.sid] - cur.update(event) - self.props['value'] = cur.vwap - return self.props - - def create_vwap(self): - return DailyVWAP(self.daycount) - -class DailyVWAP(object): - """ - A class that tracks the volume weighted average price based on tick - updates. - """ - def __init__(self, days=3): - self.window = EventWindow(days) - self.flux = 0.0 - self.volume = 0 - self.vwap = 0.0 - self.delta = timedelta(days=days) - - def update(self, event): - - # update the event window - self.window.update(event) - - # add the current event's flux and volume to the tracker - flux, volume = self.calculate_flux([event]) - self.flux += flux - self.volume += volume - - # subract the expired events flux and volume from the tracker - dropped = self.window.dropped_ticks - dropped_flux, dropped_volume = self.calculate_flux(dropped) - - self.flux -= dropped_flux - self.volume -= dropped_volume - - if(self.volume != 0): - self.vwap = self.flux / self.volume - else: - self.vwap = None - - def calculate_flux(self, ticks): - flux = 0.0 - volume = 0 - for tick in ticks: - flux += tick['volume'] * tick['price'] - volume += tick['volume'] - return flux, volume diff --git a/zipline/transforms/__init__.py b/zipline/transforms/__init__.py deleted file mode 100644 index fb244e2e..00000000 --- a/zipline/transforms/__init__.py +++ /dev/null @@ -1,25 +0,0 @@ -""" -Transforms -========== - -Transforms provide re-useable components for stream processing. All -Transforms expect to receive data events from zipline.core.DataFeed -asynchronously via zeromq. Each transform is designed to run in independent -process space, independently of all other transforms, to allow for parallel -computation. - -Each transform must maintain the state necessary to calculate the transform of -each new feed events. - -To simplify the consumption of feed and transform data events, this module -also provides the TransformsMerge class. TransformsMerge initializes as set of -transforms and subscribes to their output. Each feed event is then combined with -all the transforms of that event into a single new message. - -""" - -from base import BaseTransform - -__all__ = [ - BaseTransform, -] diff --git a/zipline/transforms/base.py b/zipline/transforms/base.py deleted file mode 100644 index 9a232ad9..00000000 --- a/zipline/transforms/base.py +++ /dev/null @@ -1,107 +0,0 @@ -from zipline.core.component import Component - -import zipline.protocol as zp -from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE, \ - CONTROL_FRAME, CONTROL_UNFRAME - - -import logbook -import time - -log = logbook.Logger('BaseTransform') - -class BaseTransform(Component): - """ - Top level execution entry point for the transform - - - 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): - pass - - @property - def get_id(self): - return self.props['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 - - """ - - if self.feed_socket in self.socks and self.socks[self.feed_socket] == self.zmq.POLLIN: - message = self.feed_socket.recv() - #import msgpack - #event = msgpack.loads(message) - #log.info(event) - - if message == str(CONTROL_PROTOCOL.DONE): - log.info("signaling done") - self.signal_done() - return - - 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.props object initialized from the transform - configuration, and only set the transformed value:: - - self.props['value'] = transformed_value - """ - raise NotImplementedError diff --git a/zipline/transforms/technical.py b/zipline/transforms/technical.py deleted file mode 100644 index 13b5b1e5..00000000 --- a/zipline/transforms/technical.py +++ /dev/null @@ -1,54 +0,0 @@ -""" -Transformations for common technical indicators. -TODO: add MACD transform -TODO: add trailing stop -""" - -import datetime - -from zipline.messaging import BaseTransform -import zipline.util as qutil - -class MovingAverage(BaseTransform): - """ - Calculate a unweighted moving average for props['sid'] security - TODO: add sid -> mvavg dict. - """ - - 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 - - def transform(self, event): - """ - 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: - self.events.pop(index) - self.current_total -= cur_event.price - index += 1 - else: - break - - if len(self.events) == 0: - return 0.0 - - self.average = self.current_total/len(self.events) - self.state['value'] = self.average - - return self.state