From 65b70f2857b5e78bb5d8119a1e9cb573d730e839 Mon Sep 17 00:00:00 2001 From: fawce Date: Sat, 11 Feb 2012 14:24:19 -0500 Subject: [PATCH] moved virtual env directory --- docs/index.rst | 42 +++++++-- docs/qsim.rst | 31 ------- etc/jenkins.sh | 6 +- qsim/core.py | 81 ++++++++++++------ qsim/test/test_messaging.py | 5 +- qsim/transforms/__init__.py | 19 ++++ qsim/transforms/core.py | 162 +++++++++++++++++++++++++++++------ qsim/transforms/merge.py | 135 ----------------------------- qsim/transforms/technical.py | 2 +- 9 files changed, 252 insertions(+), 231 deletions(-) delete mode 100644 qsim/transforms/merge.py diff --git a/docs/index.rst b/docs/index.rst index 5cd17a3b..5cbf9309 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -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 `, 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 ` 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_. diff --git a/docs/qsim.rst b/docs/qsim.rst index f1bb8fb1..f1e7216c 100644 --- a/docs/qsim.rst +++ b/docs/qsim.rst @@ -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 `, 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 ` 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 =========================== diff --git a/etc/jenkins.sh b/etc/jenkins.sh index b45cab15..37482ec4 100755 --- a/etc/jenkins.sh +++ b/etc/jenkins.sh @@ -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 diff --git a/qsim/core.py b/qsim/core.py index e4d08d5c..50924da9 100644 --- a/qsim/core.py +++ b/qsim/core.py @@ -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() diff --git a/qsim/test/test_messaging.py b/qsim/test/test_messaging.py index b641e743..d5f33bfa 100644 --- a/qsim/test/test_messaging.py +++ b/qsim/test/test_messaging.py @@ -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." diff --git a/qsim/transforms/__init__.py b/qsim/transforms/__init__.py index e69de29b..789e9543 100644 --- a/qsim/transforms/__init__.py +++ b/qsim/transforms/__init__.py @@ -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. + +""" \ No newline at end of file diff --git a/qsim/transforms/core.py b/qsim/transforms/core.py index 7e2f5dbb..79464466 100644 --- a/qsim/transforms/core.py +++ b/qsim/transforms/core.py @@ -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 {} - \ No newline at end of file + +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)) + + + \ No newline at end of file diff --git a/qsim/transforms/merge.py b/qsim/transforms/merge.py deleted file mode 100644 index e60d1e95..00000000 --- a/qsim/transforms/merge.py +++ /dev/null @@ -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)) - diff --git a/qsim/transforms/technical.py b/qsim/transforms/technical.py index 9c51bd0f..43ff8477 100644 --- a/qsim/transforms/technical.py +++ b/qsim/transforms/technical.py @@ -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): """