delinting sources.py

docs/qsim.data.rst

@@ -1,11 +0,0 @@
-data Package
-============
-
-:mod:`equity` Module
---------------------
-
-.. automodule:: qsim.data.equity
-    :members:
-    :undoc-members:
-    :show-inheritance:
-

nosetests.xml

@@ -1 +1 @@
-<?xml version="1.0" encoding="UTF-8"?><testsuite name="nosetests" tests="2" errors="0" failures="0" skip="0"><testcase classname="qsim.test.test_messaging.MessagingTestCase" name="test_client" time="0.606" /><testcase classname="qsim.test.test_messaging.MessagingTestCase" name="test_moving_average_to_client" time="161.303" /></testsuite>
+<?xml version="1.0" encoding="UTF-8"?><testsuite name="nosetests" tests="2" errors="0" failures="0" skip="0"><testcase classname="qsim.test.test_messaging.MessagingTestCase" name="test_client" time="0.520" /><testcase classname="qsim.test.test_messaging.MessagingTestCase" name="test_moving_average_to_client" time="50.254" /></testsuite>

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())

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
 
 
 

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)