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

moved virtual env directory

Browse Source
This commit is contained in:
fawce
2012-02-11 14:24:19 -05:00
parent 9822c30ff6
commit 65b70f2857
9 changed files with 252 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/index.rst
+36 -6
View File
@@ -3,12 +3,6 @@
You can adapt this file completely to your liking, but it should at least
contain the root `toctree` directive.
Quantopian Simulator: QSim
================================
Qsim runs backtests using asynchronous components and zeromq messaging for communication and coordination.
Contents:
.. toctree::
@@ -18,6 +12,42 @@ Contents:
modules.rst
messaging.rst
Quantopian Simulator: QSim
================================
Qsim runs backtests using asynchronous components and zeromq messaging for communication and coordination.
Simulator is the heart of QSim, and the primary access point for creating, launching, and tracking simulations. You can find it in :py:class:`~qsim.core.Simulator`
Simulator Sub-Components
==========================
Each simulation contains numerous subcomponents, each operating asynchronously from all others, and communicating
via zeromq.
DataSources
--------------------
A DataSource represents a historical event record, which will be played back during simulation. A simulation may have one or more DataSources, which will be combined in DataFeed. Generally, datasources read records from a persistent store (db, csv file, remote service), format the messages for downstream simulation components, and send them to a PUSH socket. See the base class for all datasources :py:class:`~qsim.sources.DataSource`
DataFeed
--------------------
All simulations start with a collection of :py:class:`DataSources <qsim.sources.DataSource>`, which need to be fed to an algorithm. Each :py:class:`~qsim.sources.DataSource`can contain events of differing content (trades, quotes, corporate event) and frequency (quarterly, intraday). To simplify the process of managing the data sources, :py:class:`~qsim.core.DataFeed` can receive events from multiple :py:class:`DataSources <qsim.sources.DataSource>` and combine them into a serial chronological stream.
Transforms
--------------------
Often, an algorithm will require a running calculation on top of a :py:class:`~qsim.sources.DataSource`, or on the consolidated feed. A simple example is a technical indicator or a moving average. Transforms can be described in :py:class:`~qsim.core.Simulator`'s configuration. Subclass :py:class:`~qsim.transforms.core.Transform` to add your own Transform. Transforms must hold their own state between events, and serialize their current values into messages.
Data Alignment
--------------------
Like Datasources, Transforms have differing frequencies and results. Simulator manages the results of parallel transforms and **aligns** transform results with the raw DataFeed. Client algorithms simply receive a map of data, which includes the current event and all the transformed values.
Time Compression
--------------------
Review the unit test coverage_.
docs/qsim.rst
-31
View File
@@ -2,37 +2,6 @@
qsim Package
============
Simulator is the heart of QSim, and the primary access point for creating, launching, and tracking simulations. You can find it in :py:class:`~qsim.core.Simulator`
Simulator Sub-Components
==========================
Each simulation contains numerous subcomponents, each operating asynchronously from all others, and communicating
via zeromq.
DataSources
--------------------
A DataSource represents a historical event record, which will be played back during simulation. A simulation may have one or more DataSources, which will be combined in DataFeed. Generally, datasources read records from a persistent store (db, csv file, remote service), format the messages for downstream simulation components, and send them to a PUSH socket. See the base class for all datasources :py:class:`~qsim.sources.DataSource`
DataFeed
--------------------
All simulations start with a collection of :py:class:`DataSources <qsim.sources.DataSource>`, which need to be fed to an algorithm. Each :py:class:`~qsim.sources.DataSource`can contain events of differing content (trades, quotes, corporate event) and frequency (quarterly, intraday). To simplify the process of managing the data sources, :py:class:`~qsim.core.DataFeed` can receive events from multiple :py:class:`DataSources <qsim.sources.DataSource>` and combine them into a serial chronological stream.
Transforms
--------------------
Often, an algorithm will require a running calculation on top of a :py:class:`~qsim.sources.DataSource`, or on the consolidated feed. A simple example is a technical indicator or a moving average. Transforms can be described in :py:class:`~qsim.core.Simulator`'s configuration. Subclass :py:class:`~qsim.transforms.core.Transform` to add your own Transform. Transforms must hold their own state between events, and serialize their current values into messages.
Data Alignment
--------------------
Like Datasources, Transforms have differing frequencies and results. Simulator manages the results of parallel transforms and **aligns** transform results with the raw DataFeed. Client algorithms simply receive a map of data, which includes the current event and all the transformed values.
QSim API
===========================
etc/jenkins.sh
+3 -3
View File
@@ -2,10 +2,10 @@
#setup virtualenvironment
export VIRTUALENVWRAPPER_PYTHON=/usr/bin/python2.7
if [ ! -d $HOME/.venvs ]; then
mkdir $HOME/.venvs
export WORKON_HOME=/mnt/jenkins_backup/virtual_envs
if [ ! -d $WORKON_HOME ]; then
mkdir $WORKON_HOME
fi
export WORKON_HOME=$HOME/.venvs
source /usr/local/bin/virtualenvwrapper.sh
#create the scientific python virtualenv and copy to provide qsim base
qsim/core.py
+55 -26
View File
@@ -5,6 +5,8 @@ Provides simulated data feed services.
import qsim.sources as sources
import qsim.util as qutil
import qsim.messaging as qmsg
import qsim.transforms.technical as ta
import zmq
import time
import logging
@@ -12,7 +14,7 @@ import json
class Simulator(object):
"""
Simulator is the heart of QSim. The beating heart...
Simulator translates configuration data into running source, feed, transform, and merge components.
"""
def __init__(self, config):
@@ -21,44 +23,74 @@ class Simulator(object):
client algorithms that simulator should create.
"""
self.config = config
self.data_workers = {}
def launch(self):
"""
Create all components specified in config...
"""
pass
self.feed = DataFeed(self.config.sources.keys())
self.start_data_sources(self.config.sources)
self.create_transforms(self.config.transforms)
def start_data_sources(self, configs):
"""
:configs: array of dicts with properties
"""
for name, info in configs.iteritems():
if(info['class'] == "EquityMinuteTrades"):
emt = EquityMinuteTrades(info['sid'], self.feed, name)
self.data_workers[name] = emt
elif(info['class'] == "RandomEquityTrades"):
ret = sources.RandomEquityTrades(info['sid'], self.feed, name, info['count'])
self.data_workers[name] = ret
qutil.LOGGER.info("starting {id}".format(id=source_id))
self.data_workers[name].start()
qutil.LOGGER.info("datasources processes launched")
def start_transforms(self, configs):
"""
:configs: Must be an array of dicts holding properties needed for each transform. See the classes in :py:module:`qsim.transforms`
Create transforms based on configs, set each transform's result address to
transforms_address. Each transform will connect to transforms_address that all transformed events will be PUSH'd
to this object.
"""
self.transforms = {}
for props in configs:
class_name = props['class']
if(class_name == 'MovingAverage'):
mavg = ta.MovingAverage(self.feed, props, self.transform_address)
self.transforms[mavg.config.name] = mavg
keys = copy.copy(self.transforms.keys())
keys.append("feed") #for the raw feed
self.data_buffer = qmsg.MergedParallelBuffer(keys)
self.buffers = {}
for name, transform in self.transforms.iteritems():
self.buffers[name] = []
qutil.LOGGER.info("starting {name}".format(name=name))
proc = multiprocessing.Process(target=transform.run)
proc.start()
class DataFeed(object):
"""DataFeed is the heart of a simulation. It is initialized with a configuration for """
def __init__(self, config):
qutil.LOGGER = qutil.LOGGER
def __init__(self, source_list):
"""
:source_list: list of source IDs
"""
self.data_address = "tcp://127.0.0.1:{port}".format(port=10101)
self.sync_address = "tcp://127.0.0.1:{port}".format(port=10102)
self.feed_address = "tcp://127.0.0.1:{port}".format(port=10103)
self.client_register = {}
self.data_workers = {}
self.config = config
for name, info in config.iteritems():
if(info['class'] == "EquityMinuteTrades"):
emt = EquityMinuteTrades(info['sid'], self, name)
self.data_workers[name] = emt
elif(info['class'] == "RandomEquityTrades"):
ret = sources.RandomEquityTrades(info['sid'], self, name, info['count'])
self.data_workers[name] = ret
self.data_buffer = qmsg.ParallelBuffer(self.data_workers.keys())
def start_data_workers(self):
"""Start a sub-process for each datasource."""
for source_id, source in self.data_workers.iteritems():
qutil.LOGGER.info("starting {id}".format(id=source_id))
source.start()
qutil.LOGGER.info("ds processes launched")
self.data_buffer = qmsg.ParallelBuffer(source_id_list)
def register_sync(self, sync_id):
self.client_register[sync_id] = "UNCONFIRMED"
@@ -104,9 +136,6 @@ class DataFeed(object):
self.data_buffer.out_socket = self.feed_socket
#start the data source workers
self.start_data_workers()
#wait for all feed subscribers
self.sync_clients()
qsim/test/test_messaging.py
+2 -3
View File
@@ -7,7 +7,7 @@ import unittest2 as unittest
import multiprocessing
from qsim.core import DataFeed
from qsim.transforms.merge import MergedTransformsFeed
from qsim.transforms.core import MergedTransformsFeed
from qsim.transforms.technical import MovingAverage
import qsim.util as qutil
@@ -41,8 +41,7 @@ class MessagingTestCase(unittest.TestCase):
feed_proc = multiprocessing.Process(target=self.feed.run)
feed_proc.start()
client.run()
self.assertEqual(self.feed.data_buffer.pending_messages(), 0,
"The feed should be drained of all messages, found {n} remaining."
qsim/transforms/__init__.py
+19
View File
@@ -0,0 +1,19 @@
"""
Transforms
==========
Transforms provide re-useable components for stream processing. All
Transforms expect to receive data events from qsim.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.
"""
qsim/transforms/core.py
+136 -26
View File
@@ -1,28 +1,14 @@
"""
Transforms
==========
Transforms provide re-useable components for stream processing. All
Transforms expect to receive data events from qsim.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.
"""
import zmq
import json
import qsim.messaging as qmsg
import qsim.util as qutil
import qsim.config as config
import copy
import multiprocessing
import zmq
import qsim.util as qutil
import qsim.messaging as qmsg
import qsim.config as config
class BaseTransform(object):
"""Parent class for feed transforms. Subclass and override transform
@@ -30,11 +16,11 @@ class BaseTransform(object):
def __init__(self, feed, config_dict, result_address):
"""
feed_address - zmq socket address, Transform will CONNECT a PULL socket and receive messages until "DONE" is received.
result_address - zmq socket address, Transform will CONNECT a PUSH socket and send messaes until feed_socket receives "DONE"
sync_address - zmq socket address, Transform will CONNECT a REQ socket and send/receive one message before entering feed loop
config - must be a dict that can be wrapped in a config.Config object with at least an entry for 'name':string value
server - if True, transform will bind to the result address (and act as a server), if False it will connect. The
:feed_address: zmq socket address, Transform will CONNECT a PULL socket and receive messages until "DONE" is received.
:result_address: zmq socket address, Transform will CONNECT a PUSH socket and send messaes until feed_socket receives "DONE"
:sync_address: zmq socket address, Transform will CONNECT a REQ socket and send/receive one message before entering feed loop
:config: must be a dict that can be wrapped in a config.Config object with at least an entry for 'name':string value
:server: if True, transform will bind to the result address (and act as a server), if False it will connect. The
the last transform in a series should be server=True so that clients can connect.
"""
@@ -119,4 +105,128 @@ class BaseTransform(object):
self.state['value'] = transformed_value
"""
return {}
class MergedTransformsFeed(BaseTransform):
""" Merge data feed and array of transform feeds into a single result vector.
PULL from feed
PULL from child transforms
PUSH merged message to client
"""
def __init__(self, feed, props):
"""
:props: - must have an entry for 'transforms':array of dicts, which are
converted to configs.
"""
BaseTransform.__init__(self, feed, props, "tcp://127.0.0.1:20202")
self.transform_address = "tcp://127.0.0.1:{port}".format(port=10104)
self.transform_socket = None
self.create_transforms(self.config.transforms)
def create_transforms(self, configs):
"""
:configs: an array of config objects with a class property. Each type of transform needs
Create transforms based on configs, set each transform's result address to
this object's transform_address, so that all transformed events will be delivered
to this object.
"""
self.transforms = {}
for props in configs:
class_name = props['class']
if(class_name == 'MovingAverage'):
mavg = ta.MovingAverage(self.feed, props, self.transform_address)
self.transforms[mavg.config.name] = mavg
keys = copy.copy(self.transforms.keys())
keys.append("feed") #for the raw feed
self.data_buffer = qmsg.MergedParallelBuffer(keys)
self.buffers = {}
for name, transform in self.transforms.iteritems():
self.buffers[name] = []
def open(self):
"""Establish zmq context, feed socket, result socket for client, and transform
socket to receive transformed events. Create and launch transforms. Will confirm
ready with the DataFeed at the conclusion."""
self.context = zmq.Context()
qutil.LOGGER.info("starting {name} transform".format(name = self.state['name']))
#create the feed SUB.
self.feed_socket = self.context.socket(zmq.SUB)
self.feed_socket.connect(self.feed.feed_address)
self.feed_socket.setsockopt(zmq.SUBSCRIBE,'')
#create the result PUSH
self.result_socket = self.context.socket(zmq.PUSH)
self.result_socket.bind(self.result_address)
#create the transform PULL.
self.transform_socket = self.context.socket(zmq.PULL)
self.transform_socket.bind(self.transform_address)
self.data_buffer.out_socket = self.result_socket
# Initialize poll set
self.poller = zmq.Poller()
self.poller.register(self.feed_socket, zmq.POLLIN)
self.poller.register(self.transform_socket, zmq.POLLIN)
self.sync.confirm()
def close(self):
"""
Close all zmq sockets and context.
"""
self.transform_socket.close()
BaseTransform.close(self)
def process_all(self):
"""
Uses a Poller to receive messages from all transforms and the feed.
All transforms corresponding to the same event are merged with each other
and the original feed event into a single message. That message is then
sent to the result socket.
"""
done_count = 0
while True:
socks = dict(self.poller.poll())
if self.feed_socket in socks and socks[self.feed_socket] == zmq.POLLIN:
message = self.feed_socket.recv()
if(message == "DONE"):
qutil.LOGGER.info("finished receiving feed to merge")
done_count += 1
else:
self.received_count += 1
event = json.loads(message)
self.data_buffer.append("feed",event)
if self.transform_socket in socks and socks[self.transform_socket] == zmq.POLLIN:
t_message = self.transform_socket.recv()
if(t_message == "DONE"):
qutil.LOGGER.info("finished receiving a transform to merge")
done_count += 1
else:
self.received_count += 1
t_event = json.loads(t_message)
self.data_buffer.append(t_event['name'], t_event)
if(done_count >= len(self.data_buffer)):
break #done!
self.data_buffer.send_next()
qutil.LOGGER.info("Transform {name} received {r} and sent {s}".format(name=self.state['name'], r=self.data_buffer.received_count, s=self.data_buffer.sent_count))
qutil.LOGGER.info("about to drain {n} messages from merger's buffer".format(n=self.data_buffer.pending_messages()))
#drain any remaining messages in the buffer
self.data_buffer.drain()
#signal to client that we're done
self.result_socket.send("DONE")
qutil.LOGGER.info("Transform {name} received {r} and sent {s}".format(name=self.state['name'], r=self.data_buffer.received_count, s=self.data_buffer.sent_count))
qsim/transforms/merge.py
-135
View File
@@ -1,135 +0,0 @@
import copy
import multiprocessing
import zmq
import technical as ta
from core import BaseTransform
import qsim.util as qutil
import qsim.messaging as qmsg
class MergedTransformsFeed(BaseTransform):
""" Merge data feed and array of transform feeds into a single result vector.
PULL from feed
PULL from child transforms
PUSH merged message to client
"""
def __init__(self, feed, props):
"""
config - must have an entry for 'transforms':array of dicts, which are
convertedto configs.
"""
BaseTransform.__init__(self, feed, props, "tcp://127.0.0.1:20202")
self.transform_address = "tcp://127.0.0.1:{port}".format(port=10104)
self.transform_socket = None
self.create_transforms(self.config.transforms)
def create_transforms(self, configs):
"""
Create transforms based on configs, set each transform's result address to
this object's transform_address, so that all transformed events will be delivered
to this object.
"""
self.transforms = {}
for props in configs:
class_name = props['class']
if(class_name == 'MovingAverage'):
mavg = ta.MovingAverage(self.feed, props, self.transform_address)
self.transforms[mavg.config.name] = mavg
keys = copy.copy(self.transforms.keys())
keys.append("feed") #for the raw feed
self.data_buffer = qmsg.MergedParallelBuffer(keys)
self.buffers = {}
for name, transform in self.transforms.iteritems():
self.buffers[name] = []
def open(self):
"""Establish zmq context, feed socket, result socket for client, and transform
socket to receive transformed events. Create and launch transforms. Will confirm
ready with the DataFeed at the conclusion."""
self.context = zmq.Context()
qutil.LOGGER.info("starting {name} transform".format(name = self.state['name']))
#create the feed SUB.
self.feed_socket = self.context.socket(zmq.SUB)
self.feed_socket.connect(self.feed.feed_address)
self.feed_socket.setsockopt(zmq.SUBSCRIBE,'')
#create the result PUSH
self.result_socket = self.context.socket(zmq.PUSH)
self.result_socket.bind(self.result_address)
#create the transform PULL.
self.transform_socket = self.context.socket(zmq.PULL)
self.transform_socket.bind(self.transform_address)
self.data_buffer.out_socket = self.result_socket
# Initialize poll set
self.poller = zmq.Poller()
self.poller.register(self.feed_socket, zmq.POLLIN)
self.poller.register(self.transform_socket, zmq.POLLIN)
for name, transform in self.transforms.iteritems():
qutil.LOGGER.info("starting {name}".format(name=name))
proc = multiprocessing.Process(target=transform.run)
proc.start()
self.sync.confirm()
def close(self):
"""
Close all zmq sockets and context.
"""
self.transform_socket.close()
BaseTransform.close(self)
def process_all(self):
"""
Uses a Poller to receive messages from all transforms and the feed.
All transforms corresponding to the same event are merged with each other
and the original feed event into a single message. That message is then
sent to the result socket.
"""
done_count = 0
while True:
socks = dict(self.poller.poll())
if self.feed_socket in socks and socks[self.feed_socket] == zmq.POLLIN:
message = self.feed_socket.recv()
if(message == "DONE"):
qutil.LOGGER.info("finished receiving feed to merge")
done_count += 1
else:
self.received_count += 1
event = json.loads(message)
self.data_buffer.append("feed",event)
if self.transform_socket in socks and socks[self.transform_socket] == zmq.POLLIN:
t_message = self.transform_socket.recv()
if(t_message == "DONE"):
qutil.LOGGER.info("finished receiving a transform to merge")
done_count += 1
else:
self.received_count += 1
t_event = json.loads(t_message)
self.data_buffer.append(t_event['name'], t_event)
if(done_count >= len(self.data_buffer)):
break #done!
self.data_buffer.send_next()
qutil.LOGGER.info("Transform {name} received {r} and sent {s}".format(name=self.state['name'], r=self.data_buffer.received_count, s=self.data_buffer.sent_count))
qutil.LOGGER.info("about to drain {n} messages from merger's buffer".format(n=self.data_buffer.pending_messages()))
#drain any remaining messages in the buffer
self.data_buffer.drain()
#signal to client that we're done
self.result_socket.send("DONE")
qutil.LOGGER.info("Transform {name} received {r} and sent {s}".format(name=self.state['name'], r=self.data_buffer.received_count, s=self.data_buffer.sent_count))
qsim/transforms/technical.py
+1 -1
View File
@@ -7,7 +7,7 @@ TODO: add trailing stop
import datetime
import qsim.util as qutil
from core import BaseTransform
from qsim.transforms.core import BaseTransform
class MovingAverage(BaseTransform):
"""