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

purging

Browse Source
This commit is contained in:
fawce
2012-08-08 22:58:54 -04:00
parent 853c2ea61e
commit cf6bef6ed8
9 changed files with 0 additions and 752 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/core/controlled.py
-74
View File
@@ -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()
zipline/core/host.py
-143
View File
@@ -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
zipline/core/process.py
-90
View File
@@ -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()
zipline/finance/movingaverage.py
-72
View File
@@ -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:]
zipline/finance/sources.py
-119
View File
@@ -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
zipline/finance/vwap.py
-68
View File
@@ -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
zipline/transforms/__init__.py
-25
View File
@@ -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,
]
zipline/transforms/base.py
-107
View File
@@ -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
zipline/transforms/technical.py
-54
View File
@@ -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