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

Merge pull request #84 from quantopian/new_world_order

Browse Source 
New world order
This commit is contained in:
Scott Sanderson
2012-08-08 20:05:00 -07:00
parent 6c35e60b57 cf6bef6ed8
commit 5e798f131b
20 changed files with 5 additions and 1440 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_components.py
+2 -2
View File
@@ -3,7 +3,7 @@ import pytz
from pprint import pformat as pf
from datetime import datetime, timedelta
from unittest2 import TestCase
from unittest2 import TestCase, skip
from collections import defaultdict
from zipline.gens.composites import date_sorted_sources, merged_transforms
@@ -218,7 +218,7 @@ class ComponentTestCase(TestCase):
comp_c.proc.join()
mon_proc.join()
@skip
def test_full(self):
monitor = create_monitor(allocator)
tests/test_optimize.py
+2 -1
View File
@@ -6,7 +6,8 @@ from collections import defaultdict
import numpy as np
from zipline.core.devsimulator import AddressAllocator
from zipline.optimize.factory import create_predictable_zipline
# TODO: refactor the factory to use generators
# from zipline.optimize.factory import create_predictable_zipline
DEFAULT_TIMEOUT = 15 # seconds
EXTENDED_TIMEOUT = 90
zipline/components/__init__.py
-13
View File
@@ -1,13 +0,0 @@
from feed import Feed
from merge import Merge
from passthrough import PassthroughTransform
from datasource import DataSource
from tradesimulation import TradeSimulationClient
__all__ = [
Feed,
Merge,
PassthroughTransform,
DataSource,
TradeSimulationClient,
]
zipline/components/aggregator.py
-144
View File
@@ -1,144 +0,0 @@
"""
Abstract base class for Feed and Merge.
Component
|
Aggregate
|
/ \
Feed Merge
"""
import logbook
import zipline.protocol as zp
from zipline.core.component import Component
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE
from zipline.transitions import WorkflowMeta
from zipline.utils.protocol_utils import Enum
log = logbook.Logger('Aggregate')
# =================
# State Transitions
# =================
INIT, READY, DRAINING = AGGREGATE_STATES = \
Enum( 'INIT', 'READY', 'DRAINING')
AGGREGATE_TRANSITIONS = dict(
do_start = (-1 , INIT) ,
do_run = (INIT , READY) ,
do_drain = (READY , DRAINING) ,
)
# =========
# Component
# =========
class Aggregate(Component):
"""
Abstract superclass to Merge & Feed. Acts on two sockets
- pull_socket
- feed_socket
Both use ``sources`` for buffering.
Feed and Merge define these differently.
"""
abstract = True
__metaclass__ = WorkflowMeta
@property
def get_type(self):
return COMPONENT_TYPE.CONDUIT
def add_source(self, source_id):
self.sources[source_id] = []
# -------------
# Core Methods
# -------------
def do_work(self):
# -------------
# Work Dispatch
# -------------
if self.socks.get(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.sources) == self.ds_finished_counter:
# Drain any remaining messages in the buffer
log.debug("Draining Feed")
self.state = DRAINING
self.drain()
self.signal_done()
else:
event = self.unframe(message)
self.append(event)
if self.is_full():
event = self.next()
if event:
self.send(event)
else:
pass
# -------------
# Flow Control
# -------------
def drain(self):
"""
Send all messages in the buffer.
"""
while self.pending_messages() > 0:
event = self.next()
self.heartbeat()
if event:
self.send(event)
def send(self, event):
"""
Send the (chronologically) next message in the buffer.
"""
self.feed_socket.send(self.frame(event), self.zmq.NOBLOCK)
self.sent_counters[event.source_id] += 1
self.sent_count += 1
def is_full(self):
"""
Indicates whether the buffer has messages in buffer for all
un-DONE, blocking sources.
"""
for source_id, events in self.sources.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.sources.itervalues():
total += len(events)
return total
def __len__(self):
"""
Buffer's length is same as internal map holding separate
sorted arrays of events keyed by source id.
"""
return len(self.sources)
zipline/components/datasource.py
-70
View File
@@ -1,70 +0,0 @@
"""
Commonly used messaging components.
"""
import zipline.protocol as zp
from zipline.core.component import Component
from zipline.protocol import COMPONENT_TYPE
class DataSource(Component):
"""
Abstract 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 set_filter(self, name, value):
self.filter[name] = value
def setup_source(self):
self.filter = {}
self.cur_event = None
@property
def get_id(self):
"""
Returns this component id, this is fixed at a class level. This
should not and cannot be contingent on arguments to the init
function. Examples:
- "TradeDataSource"
- "RandomEquityTrades"
- "SpecificEquityTrades"
"""
raise NotImplementedError
@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.ndict)
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)
def do_work(self):
raise NotImplementedError()
zipline/components/feed.py
-111
View File
@@ -1,111 +0,0 @@
import logbook
from collections import defaultdict, Counter
from zipline.components.aggregator import Aggregate, \
AGGREGATE_STATES, AGGREGATE_TRANSITIONS
import zipline.protocol as zp
log = logbook.Logger('Feed')
# =========
# Component
# =========
class Feed(Aggregate):
"""
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.
"""
states = list(AGGREGATE_STATES)
transitions = AGGREGATE_TRANSITIONS
initial_state = -1
def init(self):
self.sent_count = 0
self.received_count = 0
self.ds_finished_counter = 0
self.sources = defaultdict(list)
# source_id -> integer count
self.sent_counters = Counter()
self.recv_counters = Counter()
self.state = AGGREGATE_STATES.INIT
@property
def get_id(self):
return "FEED"
@property
def draining(self):
return self.state == AGGREGATE_STATES.DRAINING
# -------
# Sockets
# -------
def open(self):
self.pull_socket = self.bind_data()
self.feed_socket = self.bind_feed()
# -------
# Framing
# -------
def unframe(self, msg):
return zp.DATASOURCE_UNFRAME(msg)
def frame(self, event):
return zp.FEED_FRAME(event)
# -------------
# Flow Control
# -------------
def append(self, event):
"""
Add an event to the buffer for the source specified by
source_id.
"""
self.sources[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.
"""
# TODO: this is redundant to the guard in aggregator.
# is_full and draining defined in aggregator
if not(self.is_full() or self.draining):
return
earliest_source = None
earliest_event = None
# iterate over the queues of source from all sources
# (1 queue per datasource)
for source in self.sources.itervalues():
if len(source) == 0:
continue
head = source[0]
if head.dt == None:
#this is a filler event, discard
source.pop(0)
continue
if (earliest_event == None) or (head.dt <= earliest_event.dt):
earliest_event = head
earliest_source = source
if earliest_event != None:
return earliest_source.pop(0)
else:
return False
zipline/components/merge.py
-83
View File
@@ -1,83 +0,0 @@
import zipline.protocol as zp
from zipline.components.aggregator import Aggregate, \
AGGREGATE_STATES, AGGREGATE_TRANSITIONS
from collections import defaultdict, Counter
class Merge(Aggregate):
"""
Merges multiple streams of events into single messages.
"""
states = list(AGGREGATE_STATES)
transitions = AGGREGATE_TRANSITIONS
initial_state = -1
def init(self):
self.sent_count = 0
self.received_count = 0
self.draining = False
self.ds_finished_counter = 0
self.sources = defaultdict(list)
# source_id -> integer count
self.sent_counters = Counter()
self.recv_counters = Counter()
@property
def get_id(self):
return "MERGE"
# -------
# Sockets
# -------
def open(self):
self.pull_socket = self.bind_merge()
self.feed_socket = self.bind_result()
# -------
# Framing
# -------
def unframe(self, msg):
return zp.TRANSFORM_UNFRAME(msg)
def frame(self, event):
return zp.MERGE_FRAME(event)
# ---------
# Data Flow
# ---------
def append(self, event):
"""
:param event: a ndict 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.sources[event.keys()[0]].append(event)
self.received_count += 1
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.
passthrough = self.sources[zp.TRANSFORM_TYPE.PASSTHROUGH]
result = passthrough.pop(0).PASSTHROUGH
for source, events in self.sources.iteritems():
if source == zp.TRANSFORM_TYPE.PASSTHROUGH:
continue
if len(events) > 0:
cur = events.pop(0)
result.merge(cur)
return result
zipline/components/passthrough.py
-18
View File
@@ -1,18 +0,0 @@
from zipline.transforms import BaseTransform
from zipline.protocol import FEED_FRAME, TRANSFORM_TYPE
class PassthroughTransform(BaseTransform):
"""
A bypass transform passes data through unchanged.
"""
def init(self):
self.props = { 'name': 'PASSTHROUGH' }
#TODO, could save some cycles by skipping the _UNFRAME call
# and just setting value to original msg string.
def transform(self, event):
return {
'name' : TRANSFORM_TYPE.PASSTHROUGH,
'value' : FEED_FRAME(event)
}
zipline/components/tradesimulation.py
-243
View File
@@ -1,243 +0,0 @@
import logbook
import datetime
import zmq
import zipline.protocol as zp
import zipline.finance.performance as perf
from zipline.core.component import Component
from zipline.finance.trading import TransactionSimulator
from zipline.utils.protocol_utils import ndict
from zipline.utils.log_utils import ZeroMQLogHandler, stdout_only_pipe
from logbook import Logger, NestedSetup, Processor
log = logbook.Logger('TradeSimulation')
class TradeSimulationClient(Component):
def init(self, trading_environment, sim_style, results_socket, algorithm):
self.received_count = 0
self.prev_dt = None
self.event_queue = None
self.txn_count = 0
self.order_count = 0
self.trading_environment = trading_environment
self.current_dt = trading_environment.period_start
self.last_iteration_dur = datetime.timedelta(seconds=0)
self.algorithm = algorithm
self.algorithm.set_order(self.order)
self.max_wait = datetime.timedelta(seconds=60)
self.last_msg_dt = datetime.datetime.utcnow()
self.txn_sim = TransactionSimulator(
open_orders={},
style=sim_style
)
self.event_data = ndict()
self.perf = perf.PerformanceTracker(
self.trading_environment,
self.algorithm.get_sid_filter()
)
self.zmq_out = None
self.results_socket = results_socket
self.algo_initialized = False
@property
def get_id(self):
return str(zp.FINANCE_COMPONENT.TRADING_CLIENT)
def open(self):
self.result_feed = self.connect_result()
if self.results_socket:
sock = self.context.socket(zmq.PUSH)
sock.connect(self.results_socket)
self.results_socket = sock
self.sockets.append(sock)
self.out_socket = sock
self.setup_logging(sock)
self.perf.publish_to(sock)
def initialize_algo(self):
""" Setup loggers for algorithm and run algorithm's own
initialize method.
"""
self.logger = Logger("Print")
self.algo_log = Logger("AlgoLog")
self.algorithm.set_logger(self.algo_log)
self.do_op(self.algorithm.initialize)
self.algo_initialized = True
def setup_logging(self, socket = None):
sock = socket or self.results_socket
self.zmq_out = ZeroMQLogHandler(
socket = sock,
)
# This is a class, which is instantiated later
# in run_algorithm. The class provides a generator.
self.stdout_capture = stdout_only_pipe
def do_work(self):
if not self.algo_initialized:
self.initialize_algo()
# see if the poller has results for the result_feed
if self.socks.get(self.result_feed) == self.zmq.POLLIN:
self.last_msg_dt = datetime.datetime.utcnow()
# get the next message from the result feed
msg = self.result_feed.recv()
# if the feed is done, shut 'er down
if msg == str(zp.CONTROL_PROTOCOL.DONE):
self.finish_simulation()
return
# result_feed is a merge component, so unframe accordingly
event = zp.MERGE_UNFRAME(msg)
self.received_count += 1
# update performance and relay the event to the algorithm
self.process_event(event)
if self.perf.exceeded_max_loss:
self.finish_simulation()
def finish_simulation(self):
log.info("TradeSimulation is Done")
# signal the performance tracker that the simulation has
# ended. Perf will internally calculate the full risk report.
self.perf.handle_simulation_end()
# signal Simulator, our ComponentHost, that this component is
# done and Simulator needn't block exit on this component.
self.signal_done()
def process_event(self, event):
# generate transactions, if applicable
txn = self.txn_sim.apply_trade_to_open_orders(event)
if txn:
event.TRANSACTION = txn
# track the number of transactions, for testing purposes.
self.txn_count += 1
else:
event.TRANSACTION = None
# the performance class needs to process each event, without
# skipping. Algorithm should wait until the performance has been
# updated, so that down stream components can safely assume that
# performance is up to date. Note that this is done before we
# mark the time for the algorithm's processing, thereby not
# running the algo's clock for performance book keeping.
self.perf.process_event(event)
# mark the start time for client's processing of this event.
event_start = datetime.datetime.utcnow()
# queue the event.
self.queue_event(event)
# if the event is later than our current time, run the algo
# otherwise, the algorithm has fallen behind the feed
# and processing per event is longer than time between events.
if event.dt >= self.current_dt:
# compress time by moving the current_time up to the event
# time.
self.current_dt = event.dt
self.run_algorithm()
# tally the time spent on this iteration
self.last_iteration_dur = datetime.datetime.utcnow() - event_start
# move the algorithm's clock forward to include iteration time
self.current_dt = self.current_dt + self.last_iteration_dur
def run_algorithm(self):
"""
As per the algorithm protocol:
- Set the current portfolio for the algorithm as per protocol.
- Construct data based on backlog of events, send to algorithm.
"""
data = self.get_data()
if len(data) > 0:
data.portfolio = self.perf.get_portfolio()
# data injection pipeline for log rerouting
# any fields injected here should be added to
# LOG_EXTRA_FIELDS in zipline/protocol.py
self.do_op(self.algorithm.handle_data, data)
def do_op(self, callable_op, *args, **kwargs):
""" Wrap a callable operation with the zmq logbook
handler if it exits."""
if self.zmq_out:
def inject_event_data(record):
# Record the simulation time.
record.extra['algo_dt'] = self.current_dt
data_injector = Processor(inject_event_data)
log_pipeline = NestedSetup([self.zmq_out,data_injector])
with log_pipeline.threadbound(), self.stdout_capture(self.logger, ''):
callable_op(*args, **kwargs)
# if no log socket, just run the algo normally
else:
callable_op(*args, **kwargs)
#Testing utility for log capture.
# TODO: remove test code from here.
def test_run_algorithm(self):
# since open is never called from some tests we need to
# set the logger explicitly
self.algorithm.set_logger(self.algo_log)
def inject_event_data(record):
# Mock an event.dt
record.extra['algo_dt'] = datetime.datetime.utcnow()
data_injector = Processor(inject_event_data)
log_pipeline = NestedSetup([self.zmq_out,
#e.g. FileHandler(...)
data_injector])
with log_pipeline.threadbound(), self.stdout_capture(self.logger, ''):
self.algorithm.handle_data('data')
#def connect_order(self):
# return self.connect_push_socket(self.addresses['order_address'])
def order(self, sid, amount):
order = zp.ndict({
'dt':self.current_dt,
'sid':sid,
'amount':amount
})
self.order_count += 1
self.perf.log_order(order)
self.txn_sim.add_open_order(order)
def queue_event(self, event):
if self.event_queue == None:
self.event_queue = []
self.event_queue.append(event)
def get_data(self):
for event in self.event_queue:
#alias the dt as datetime
event.datetime = event.dt
self.event_data[event['sid']] = event
self.event_queue = []
return self.event_data
zipline/core/__init__.py
-2
View File
@@ -1,9 +1,7 @@
from host import ComponentHost
from component import Component
from monitor import Monitor
__all__ = [
Component,
Monitor,
ComponentHost
]
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/monitor.py
+1 -1
View File
@@ -7,7 +7,7 @@ import logbook
from setproctitle import setproctitle
from signal import SIGHUP, SIGINT
from collections import OrderedDict, Counter
from collections import Counter
from zipline.protocol import (
CONTROL_PROTOCOL,
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