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

cleanup

Browse Source
This commit is contained in:
fawce
2012-02-09 22:46:35 -05:00
parent 27df787f84
commit d3d3ef833c
9 changed files with 762 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
qsim/messaging.py
+8 -3
View File
@@ -5,7 +5,7 @@ import json
import uuid
import zmq
import util as qutil
import qsim.util as qutil
class ParallelBuffer(object):
""" holds several queues of events by key, allows retrieval in date order
@@ -20,13 +20,16 @@ class ParallelBuffer(object):
self.data_buffer[key] = []
def __len__(self):
"""buffer's length is same as internal map holding separate sorted arrays of events keyed by source id"""
return len(self.data_buffer)
def append(self, key, value):
self.data_buffer[key].append(value)
def append(self, source_id, value):
"""add an event to the buffer for the source specified by source_id"""
self.data_buffer[source_id].append(value)
self.received_count += 1
def next(self):
"""Get the next message in chronological order"""
if(not(self.is_full() or self.draining)):
return
@@ -43,12 +46,14 @@ class ParallelBuffer(object):
return earliest.pop(0)
def is_full(self):
"""indicates whether the buffer has messages in buffer for all un-DONE sources"""
for source, events in self.data_buffer.iteritems():
if (len(events) == 0):
return False
return True
def pending_messages(self):
""""""
total = 0
for source, events in self.data_buffer.iteritems():
total += len(events)
qsim/simulator/qbt.py
-67
View File
@@ -1,67 +0,0 @@
"""
QBT - Quantopian Backtest
====================================
qbt runs backtests using multiple processes and zeromq messaging for communication and coordination.
Backtest is the primary process. It maintains both server and client sockets:
zmq sockets for internal processing::
- data sink, ZMQ.REQ. Port = port_start + 1
- backtest will connect to socket, and then spawn one process per datasource, passing the data sink url as a startup arg. Each
datasource process will bind to the socket, and start processing
- backtest is responsible for merging the data events from all sources into a serialized stream and relaying it to the
aggregators, merging agg results, and transmitting consolidated stream to event feed.
- agg source, ZMQ.PUSH. Port = port_start + 2
- agg sink, ZMQ.PULL. Port = port_start + 3
- control source, ZMQ.PUB. Port = port_start + 4
- all child processes must subscribe to this socket. Control commands:
- START -- begin processing
- TIME -- current simulated time in backtest
- KILL -- exit immediately
zmq sockets for backtest clients:
=================================
- orders sink, ZMQ.RESP. Port = port_start + 5
- backtest will connect (can you bind?) to this socket and await orders from the client. Order data will be processed against the streaming datafeed.
- event feed, ZMQ.RESP. Port = port_start + 6
- backtest will bind to this socket and respond to requests from client for more data. Response data will be the queue of events that
transpired since the last request.
"""
import copy
import multiprocessing
import zmq
import qsim.util as qutil
import qsim.simulator.sources as sources
CONTROLLER_PORT = 9000
DATA_SINK_PORT = 10000
DATA_FEED_PORT = 30000
class Backtest(object):
def __init__(self, db, logger):
qutil.logger = logger
self.db = db
self.feed = DataFeed(db, logger)
def start_feed(self):
proc1 = multiprocessing.Process(target=feed.run)
proc1.start()
def run(self):
# Prepare our context and sockets
self.context = zmq.Context()
#create the feed sink.
self.feed_address = self.feed.data_address
self.feed_socket = self.context.connect(self.feed_address)
self.feed_socket.connect(zmq.PULL)
qsim/test/test_messaging.py
+31 -10
View File
@@ -1,7 +1,9 @@
"""
Test suite for the messaging infrastructure of QSim.
"""
#don't worry about excessive public methods pylint: disable=R0904
import unittest2 as unittest
import zmq
import logging
import tornado
import multiprocessing
from qsim.simulator.feed import DataFeed
@@ -12,9 +14,12 @@ import qsim.util as qutil
from qsim.test.client import TestClient
class MessagingTestCase(unittest.TestCase):
class MessagingTestCase(unittest.TestCase):
"""Tests the message passing: datasources -> feed -> transforms -> merge -> client"""
def setUp(self):
"""generate some config objects for the datafeed, sources, and transforms."""
qutil.configure_logging()
qutil.logger.info("testing...")
self.total_data_count = 800
@@ -25,9 +30,12 @@ class MessagingTestCase(unittest.TestCase):
self.config = {}
self.config['name'] = '**merged feed**'
self.config['transforms'] = [{'name':'mavg1', 'class':'MovingAverage', 'hours':1},{'name':'mavg2', 'class':'MovingAverage', 'hours':2}]
self.config['transforms'] = [{'name':'mavg1', 'class':'MovingAverage', 'hours':1},
{'name':'mavg2', 'class':'MovingAverage', 'hours':2}]
def test_client(self):
def test_client(self):
"""directly connect the test client to the feed, using two random data sources"""
#subscribe a client to the multiplexed feed
client = TestClient(self.feed, self.feed.feed_address)
@@ -36,11 +44,18 @@ class MessagingTestCase(unittest.TestCase):
client.run()
self.assertEqual(self.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages, found {n} remaining.".format(n=self.feed.data_buffer.pending_messages()))
self.assertEqual(self.total_data_count, client.received_count, "The client should have received ({n}) the same number of messages as the feed sent ({m}).".format(n=client.received_count, m=self.total_data_count))
self.assertEqual(self.feed.data_buffer.pending_messages(), 0,
"The feed should be drained of all messages, found {n} remaining."
.format(n=self.feed.data_buffer.pending_messages()))
self.assertEqual(self.total_data_count, client.received_count,
"The client should have received ({n}) the same number of messages as the feed sent ({m})."
.format(n=client.received_count, m=self.total_data_count))
def dtest_moving_average_to_client(self):
"""2 datasources -> feed -> moving average transform -> testclient
verify message count at client."""
mavg = MovingAverage(self.feed, self.config['transforms'][0], result_address="tcp://127.0.0.1:20202")
mavg_proc = multiprocessing.Process(target=mavg.run)
mavg_proc.start()
@@ -52,9 +67,14 @@ class MessagingTestCase(unittest.TestCase):
client.run()
self.assertEqual(self.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages.")
self.assertEqual(self.total_data_count, client.received_count, "The client should have received the same number of messages as the feed sent.")
self.assertEqual(self.total_data_count, client.received_count,
"The client should have received the same number of messages as the feed sent.")
def dtest_merged_to_client(self):
"""
2 datasources -> feed -> 2 moving average transforms -> transform merge -> testclient
verify message count at client.
"""
merger = MergedTransformsFeed(self.feed, self.config)
merger_proc = multiprocessing.Process(target=merger.run)
merger_proc.start()
@@ -66,5 +86,6 @@ class MessagingTestCase(unittest.TestCase):
client.run()
self.assertEqual(self.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages.")
self.assertEqual(self.total_data_count, client.received_count, "The client should have received the same number of messages as the feed sent.")
self.assertEqual(self.total_data_count, client.received_count,
"The client should have received the same number of messages as the feed sent.")