synchronized data sources send events to feed.

data/feed.py

@@ -0,0 +1,67 @@
+
+from data.sources.equity import *
+
+class DataFeed(object):
+    
+    def __init__(self, db, logger):
+        self.logger = logger
+        self.db = db
+        self.data_workers = {}
+        self.data_address = "tcp://127.0.0.1:{port}".format(port=10101)
+        self.sync_address = "tcp://127.0.0.1:{port}".format(port=10102)
+        
+    def start_data_workers(self):
+        """Start a sub-process for each datasource."""
+        
+        # Socket to receive signals
+        syncservice = self.context.socket(zmq.REP)
+        syncservice.bind(self.sync_address)
+        
+        
+        emt1 = EquityMinuteTrades(133, self.db, self.data_address, self.sync_address, 1, self.logger)
+        self.data_workers[1] = emt1
+        emt1.start()
+        emt2 = EquityMinuteTrades(134, self.db, self.data_address, self.sync_address, 2, self.logger)
+        self.data_workers[2] = emt1
+        emt2.start()
+        
+        workers = 0
+        while workers < len(self.data_workers):
+            # wait for synchronization request
+            msg = syncservice.recv()
+            # send synchronization reply
+            syncservice.send('')
+            workers += 1
+        
+        syncservice.close()
+        
+        self.logger.info("{count} ds processes launched".format(count=workers))
+       
+    def run(self):   
+        # Prepare our context and sockets
+        self.context = zmq.Context()
+        
+        #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 
+        #see: http://zguide.zeromq.org/py:taskwork2
+        self.data_socket = self.context.socket(zmq.PULL)
+        self.data_socket.bind(self.data_address)
+        
+        counter = 0
+        
+        self.start_data_workers()
+        
+        while True:
+            message = self.data_socket.recv()
+            event = json.loads(message)
+            #self.logger.info(message)
+            counter = counter + 1
+            if(event['type'] == "DONE"):
+                source = event['s']
+                if(self.data_workers.has_key(source)):
+                    del(self.data_workers[source])
+                if(len(self.data_workers) == 0):
+                    break
+                
+        self.logger.info("Collected {n} messages".format(n=counter))
+        self.data_socket.close()
+        self.context.term()

data/sources/equity.py

@@ -5,6 +5,7 @@ import pymongo.json_util
 import json
 import pytz
 import copy
+import multiprocessing
 from pymongo import ASCENDING, DESCENDING
 
 from backtest.util import *
@@ -12,83 +13,69 @@ from backtest.util import *
 
 class EquityMinuteTrades(object):
     
-    def __init__(self, sid, db, data_socket, control_socket, source_id, logger):
+    def __init__(self, sid, db, data_address, sync_address, source_id, logger):
         self.sid = sid
         self.db = db
         self.source_id      = source_id
         self.logger         = logger
-        self.control_socket = control_socket
-        self.data_socket = data_socket
-        self.logger.info("data socket is {ds}".format(ds=data_socket))
-        self.logger.info("control socket is {cs}".format(cs=control_socket))
+        self.data_address   = data_address
+        self.sync_address   = sync_address
+        self.logger.info("data address is {ds}".format(ds=data_address))
         
+        self.cur_event = None
         
-    def run2(self):
-        self.context = zmq.Context()
-        
-        #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 
-        #self.data_source = self.context.socket(zmq.PUSH)
-        #self.data_source.connect(data_socket)
-        
-        #create the control subscription
-        self.qbt_control = self.context.socket(zmq.SUB)
-        self.qbt_control.connect(self.control_socket)
-        self.qbt_control.setsockopt(zmq.SUBSCRIBE, '')
-        
-        while True:
-            self.logger.info("about to call receive")
-            try:
-                message = self.qbt_control.recv()
-                self.logger.info("received message: {msg}".format(msg=message))
-            except zmq.ZMQError as err:
-                if err.errno != zmq.EAGAIN:
-                    raise err
-            
-            
-        
+    def start(self):
+        self.proc = multiprocessing.Process(target=self.run)
+        self.proc.start()
+               
     def run(self):
-        
         self.logger.info("starting data source:{sid}".format(sid=self.sid))
         self.context = zmq.Context()
         
         #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 
-        self.data_source = self.context.socket(zmq.PUSH)
-        self.data_source.connect(self.data_socket)
-        
-        #create the control subscription
-        self.qbt_control = self.context.socket(zmq.SUB)
-        self.qbt_control.connect(self.control_socket)
-        self.qbt_control.setsockopt(zmq.SUBSCRIBE, '')
+        self.data_socket = self.context.socket(zmq.PUSH)
+        self.data_socket.connect(self.data_address)
         
         eventQS = self.db.equity.trades.minute.find(fields=["sid","price","volume","dt"],
                                      spec={"sid":self.sid},
                                      sort=[("dt",ASCENDING)],
                                      slave_ok=True)
         self.logger.info("found {count} events".format(count=eventQS.count()))
-        control_dt_str = None
+        
+        #synchronize with feed
+        syncclient = self.context.socket(zmq.REQ)
+        syncclient.connect(self.sync_address)
+    
+        # send a synchronization request
+        syncclient.send('')
+        # wait for synchronization reply
+        syncclient.recv()
+        
+        syncclient.close()
         
         for doc in eventQS:
             doc_dt = doc['dt'].replace(tzinfo = pytz.utc)
             doc_dt_str = format_date(doc_dt)
             event = copy.copy(doc)
             event['dt'] = doc_dt_str
-            event['s'] = self.source_id #s is for source
+            event['s'] = self.source_id 
+            event['type'] = 'event'
             del(event['_id'])
             
-            #wait for a control message if our current event is ahead of the control time
-            #if our current event is in the past wrt to control time, keep sending messages.
-            if(control_dt_str == None or event['dt'] > control_dt_str):
-                try:               
-                    self.logger.info("about to call receive")
-                    control_dt_str = self.qbt_control.recv()
-                    self.logger.info("received message: {msg}".format(msg=control_dt_str))
-                except zmq.ZMQError:
-                    self.logger.info("we got an error on receive")
-                    continue
-                    
-            #send this event to qbt
-            self.logger.info("sending {event}".format(event=event))
-            self.data_source.send(json.dumps(event))
+            #send this event to feed address
+            #self.logger.info("sending {event}".format(event=event))
+            self.data_socket.send(json.dumps(event))
+        
+        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")
+        
+        
+            
    
        
         

qbt.py

@@ -39,24 +39,26 @@ from data.sources.equity import *
 
 
 
+CONTROLLER_PORT     =  9000
 DATA_SINK_PORT      = 10000
-CONTROLLER_PORT     = 10099
-DATA_FEED_PORT      = 10002
+DATA_FEED_PORT      = 30000
 
 class Backtest(object):
     
     def __init__(self, db, logger):
         self.logger = logger
         self.db = db
+        self.data_workers = {}
         
         
     def start_data_workers(self):
         """Start a sub-process for each datasource."""
         
-        emt1 = EquityMinuteTrades(133, self.db, self.data_socket, self.controller_socket, 1, self.logger)
+        emt1 = EquityMinuteTrades(133, self.db, DATA_SINK_PORT, 1, self.logger)
+        data_workers[1] = emt1
+        emt1.start()
         #emt2 = EquityMinuteTrades(134, self.db, self.data_socket, self.controller_socket, 2, self.logger)
-        multiprocessing.Process(target=emt1.run).start()
-        #multiprocessing.Process(target=emt2.run).start()
+        #emt2.start()
         self.logger.info("ds processes launched")
        
     def run(self):   
@@ -64,53 +66,50 @@ class Backtest(object):
         self.context = zmq.Context()
         
         #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 
+        #see: http://zguide.zeromq.org/py:taskwork2
         self.data_socket = "tcp://127.0.0.1:{port}".format(port=DATA_SINK_PORT)
         self.data_sink = self.context.socket(zmq.PULL)
-        ##TODO: findout out why the * is necessary. localhost causes "not supported" exceptions.
-        #see: http://zguide.zeromq.org/py:tasksink2
-        #see: http://zguide.zeromq.org/py:taskwork2
         self.data_sink.bind(self.data_socket)
         
         #create the controller publishing socket.
         self.controller_socket = "tcp://127.0.0.1:{port}".format(port=CONTROLLER_PORT) 
-        self.controller = self.context.socket(zmq.PUB)
-        self.controller.bind(self.controller_socket)
+        #self.controller = self.context.socket(zmq.PUB)
+        #self.controller.bind(self.controller_socket)
         
         
         #create the merged dataset feed socket
-        #self.data_feed = self.context.socket(zmq.PUSH)
-        #self.data_feed.bind("tcp://127.0.0.1:{port}".format(port=DATA_FEED_PORT))
+        self.data_feed = self.context.socket(zmq.PUSH)
+        self.data_feed.bind("tcp://127.0.0.1:{port}".format(port=DATA_FEED_PORT))
         
         self.last_event_time = None
         self.data_workers = []
         
-        last_dt = "."
+        last_dt = None
         event_q = []
+        #data workers start in independent processes, connecting to the data_socket
         self.start_data_workers()
         while True:
-            #ask all data sources for next event in their sequence, which happened on or before last_dt.
-            #last_dt of None is interpreted as next, without regard to date.
-            self.logger.info("qbt sending dt message")
-            self.controller.send(last_dt)
-            #self.data_feed.send("2011/04/11-22:30:10.100")
-            #self.logger.info("qbt message sent")
-            
-            while True:
-                try:
-                    #self.logger.info("about to receive")
-                    message = self.data_sink.recv(zmq.NOBLOCK)
-                    event = json.loads(message)
-                    last_dt = event['dt']
-                    self.logger.info("got message: {msg} with dt : {dt}".format(msg=event, dt=last_dt))
-                    event_q.append(event)
-                except zmq.ZMQError as err:
-                    
-                    #EAGAIN indicates recv doesn't have messages now, and datasources are sending 
-                    #so only throw if we have an error other than EAGAIN.
-                    if err.errno != zmq.EAGAIN:
-                         raise err #real error, throw back to caller
-                    #self.logger.info("we received an error")
-                    break
+            try:
+                #self.logger.info("about to receive")
+                message = self.data_sink.recv()
+                event = json.loads(message)
+                last_dt = event['dt']
+                
+                event_q.append(event)
+                event_q = sorted(event_q, key=lambda event: event['dt'])
+                cur_event = event_q.pop(0)
+                self.data_feed.send(json.dumps(cur_event))
+                
+                #this signals loop at datasource should proceed... all sources will send all events as fast as possible.
+                self.data_sink.send(str(last_dt))
+            except zmq.ZMQError as err:
+                
+                #EAGAIN indicates recv doesn't have messages now, and datasources are sending 
+                #so only throw if we have an error other than EAGAIN.
+                if err.errno != zmq.EAGAIN:
+                     raise err #real error, throw back to caller
+                #self.logger.info("we received an error")
+                break
                 
             #no events in q at this point means we've processed all the data in all the sources!
             #if(len(event_q) == 0):

test.py

@@ -1,7 +1,19 @@
 import zmq
+import logging
+import tornado
 from data.sources.equity import *
+from data.feed import *
+from qbt_server import connect_db
 
-def main():
+def datafeed():
+    connection, db = connect_db()
+    logger = logging.getLogger()
+    feed = DataFeed(db, logger)
+    logger.info("starting the feed")
+    feed.run()
+    
+    
+def sub():
     context = zmq.Context()
     controller = context.socket(zmq.SUB)
     controller.connect("tcp://127.0.0.1:10099")    
@@ -13,7 +25,17 @@ def main():
         except zmq.ZMQError as err:
             if err.errno != zmq.EAGAIN:
                 raise err
+                
+def pub():
+    context = zmq.Context()
+    controller = context.socket(zmq.PUB)
+    controller.bind("tcp://127.0.0.1:10099")    
+    while True:
+        controller.send("HELLO3")
+        
+
      
 if __name__ == "__main__":
-    main()              
+    tornado.options.parse_command_line()
+    datafeed()