From 4b29999f6a843a7b737aa980c449aed297f8a1db Mon Sep 17 00:00:00 2001 From: fawce Date: Wed, 8 Feb 2012 22:14:58 -0500 Subject: [PATCH] delinting sources.py --- docs/qsim.data.rst | 11 ----- nosetests.xml | 2 +- qsim/data/__init__.py | 0 qsim/messaging/__init__.py | 0 qsim/simulator/feed.py | 4 +- qsim/simulator/qbt.py | 2 +- qsim/{data/equity.py => simulator/sources.py} | 46 ++++++++++++++----- 7 files changed, 38 insertions(+), 27 deletions(-) delete mode 100644 docs/qsim.data.rst delete mode 100644 qsim/data/__init__.py delete mode 100644 qsim/messaging/__init__.py rename qsim/{data/equity.py => simulator/sources.py} (54%) diff --git a/docs/qsim.data.rst b/docs/qsim.data.rst deleted file mode 100644 index 393b1123..00000000 --- a/docs/qsim.data.rst +++ /dev/null @@ -1,11 +0,0 @@ -data Package -============ - -:mod:`equity` Module --------------------- - -.. automodule:: qsim.data.equity - :members: - :undoc-members: - :show-inheritance: - diff --git a/nosetests.xml b/nosetests.xml index a13d8ef0..3841115e 100644 --- a/nosetests.xml +++ b/nosetests.xml @@ -1 +1 @@ - \ No newline at end of file + \ No newline at end of file diff --git a/qsim/data/__init__.py b/qsim/data/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/qsim/messaging/__init__.py b/qsim/messaging/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/qsim/simulator/feed.py b/qsim/simulator/feed.py index e3256df4..04e435d6 100644 --- a/qsim/simulator/feed.py +++ b/qsim/simulator/feed.py @@ -1,5 +1,5 @@ -import qsim.data.equity as qequity +import qsim.simulator.sources as sources import qsim.util as qutil import zmq import time @@ -24,7 +24,7 @@ class DataFeed(object): emt = EquityMinuteTrades(info['sid'], self, name) self.data_workers[name] = emt elif(info['class'] == "RandomEquityTrades"): - ret = qequity.RandomEquityTrades(info['sid'], self, name, info['count']) + ret = sources.RandomEquityTrades(info['sid'], self, name, info['count']) self.data_workers[name] = ret self.data_buffer = qutil.ParallelBuffer(self.data_workers.keys()) diff --git a/qsim/simulator/qbt.py b/qsim/simulator/qbt.py index 98f1a21c..030b2d96 100644 --- a/qsim/simulator/qbt.py +++ b/qsim/simulator/qbt.py @@ -35,7 +35,7 @@ import multiprocessing import zmq import qsim.util as qutil -import qsim.data.equity as qequity +import qsim.simulator.sources as sources diff --git a/qsim/data/equity.py b/qsim/simulator/sources.py similarity index 54% rename from qsim/data/equity.py rename to qsim/simulator/sources.py index 790046d8..1c59240e 100644 --- a/qsim/data/equity.py +++ b/qsim/simulator/sources.py @@ -1,31 +1,39 @@ +""" +Provides data source handlers that can push messages to a qsim.simulator.DataFeed +""" import datetime import zmq import json -import pytz -import copy import multiprocessing -import logging import random import qsim.util as qutil class DataSource(object): + """ + 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). + """ def __init__(self, feed, source_id): self.source_id = source_id - self.logger = qutil.logger self.feed = feed self.sync = qutil.FeedSync(self.feed, str(source_id)) self.data_address = self.feed.data_address - self.logger.info("data address is {ds}".format(ds=self.feed.data_address)) + qutil.logger.info("data address is {ds}".format(ds=self.feed.data_address)) self.cur_event = None def start(self): + """Launch the datasource in a separate process.""" self.proc = multiprocessing.Process(target=self.run) self.proc.start() def open(self): - self.logger.info("starting data source:{source_id} on {addr}".format(source_id=self.source_id, addr=self.feed.data_address)) + """create zmq context and socket""" + qutil.logger.info("starting data source:{source_id} on {addr}" + .format(source_id=self.source_id, addr=self.feed.data_address)) + self.context = zmq.Context() #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 @@ -35,29 +43,40 @@ class DataSource(object): #signal we are ready self.sync.confirm() - def run(self): + def run(self): + """Fully execute this datasource.""" try: self.open() self.send_all() self.close() except Exception as err: - self.logger.exception("Unexpected failure running datasource - {name}.".format(name=self.source_id)) + qutil.logger.exception("Unexpected failure running datasource - {name}." + .format(name=self.source_id)) def send(self, event): + """ + event is expected to be a dict + sets source_id and type properties in the dict + sends to the data_socket. + """ event['s'] = self.source_id event['type'] = 'event' self.data_socket.send(json.dumps(event)) def close(self): + """ + Close the zmq context and sockets. + """ done_msg = {} done_msg['type'] = 'DONE' done_msg['s'] = self.source_id self.data_socket.send(json.dumps(done_msg)) self.data_socket.close() self.context.term() - self.logger.info("finished processing data source") + qutil.logger.info("finished processing data source") class RandomEquityTrades(DataSource): + """Generates a random stream of trades for testing.""" def __init__(self, sid, feed, source_id, count): DataSource.__init__(self, feed, source_id) @@ -67,11 +86,14 @@ class RandomEquityTrades(DataSource): def send_all(self): trade_start = datetime.datetime.now() minute = datetime.timedelta(minutes=1) - price = random.uniform(5.0,50.0) + price = random.uniform(5.0, 50.0) for i in range(self.count): - price = price + random.uniform(-0.05,0.05) - event = {'sid':self.sid, 'dt':qutil.format_date(trade_start + (minute * i)),'price':price, 'volume':random.randrange(100,10000,100)} + price = price + random.uniform(-0.05, 0.05) + event = {'sid':self.sid, + 'dt':qutil.format_date(trade_start + (minute * i)), + 'price':price, + 'volume':random.randrange(100,10000,100)} self.send(event)