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)