the code is a mess right now, but i need to save progress off of my lappy. figured out that creating a context in the init wouldn't work with sending an object to multiprocessing. probably due to serialization issues for the context

README.md

@@ -10,7 +10,7 @@ Navigate to your mongodb installation and start your db server:
 Install the necessary python libraries:
 
 ```
-	easy_install tornado pymongo
+	easy_install tornado pymongo pytz
 ```
 
 Create a development database with sample data, will create one qbt user:
@@ -75,3 +75,13 @@ QBT requires a running mongodb instance with a few collections:
 ## Authenticating
 
 ## Requesting a Backtest
+
+#Data Sources
+The Backtest can handle multiple concurrent data sources. QBT will start a subprocess to run each datasource, and merge all events from all sources into a single serial feed. 
+
+Data sources have events with very different frequencies. For example, liquid stocks will trade many times per minute, while illiquid stocks may trade just once a day. In order to serialize events from all sources into a single feed, qbt loads events from all sources into memory, then sorts. The communication happens like this:
+1. QBT requests the next event from each data source, ignoring date (i.e. just next in sequence for all)
+2. Using the earliest date from all the events from all sources, QBT then asks for "next after <date>" from all sources. 
+3. All datasources send all events in their history from before <date>, moving their internal pointer forward to the next unsent event.
+4. QBT merges all events in memory
+5. goto 1!

backtest/util.py

@@ -32,4 +32,18 @@ class DocWrap():
         if self.store.has_key(attrname):
             return self.store[attrname]
         else:
-            raise AttributeError("No attribute named {name}".format(name=attrname))
+            raise AttributeError("No attribute named {name}".format(name=attrname))
+            
+def parse_date(dt_str):
+    """parse strings according to the same format as generated by format_date"""
+    if(dt_str == None):
+        return None
+    dt = datetime.datetime.strptime(dt_str(".")[0], '%Y/%m/%d-%H:%M:%S').replace(microsecond=int(dt_str(".")[1]+"000")).replace(tzinfo = pytz.utc)
+    return dt
+    
+def format_date(dt):
+    """Format the date into a date with millesecond resolution and string/alphabetical sorting that is equivalent to datetime sorting"""
+    if(dt == None):
+        return None
+    dt_str = dt.strftime('%Y/%m/%d-%H:%M:%S') + "." + str(dt.microsecond / 1000)
+    return dt_str

data/sources/equity.py

@@ -0,0 +1,95 @@
+import datetime
+import zmq
+import pymongo
+import pymongo.json_util
+import json
+import pytz
+import copy
+from pymongo import ASCENDING, DESCENDING
+
+from backtest.util import *
+
+
+class EquityMinuteTrades(object):
+    
+    def __init__(self, sid, db, data_socket, control_socket, 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))
+        
+        
+    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 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, '')
+        
+        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
+        
+        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
+            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))
+   
+       
+        
+        

dev_setup.py

@@ -0,0 +1,54 @@
+import tornado.auth
+import tornado.httpserver
+import tornado.ioloop
+from tornado.options import define, options
+import tornado.web
+import pymongo
+import bson
+import hashlib
+import base64
+import uuid
+import os
+import logging
+import datetime
+import random
+
+import qbt_server
+
+MINUTE_COUNT=390
+
+define("user_email", default="qbt@quantopian.com", help="email address for qbt user")
+define("password", default="foobar", help="password for qbt user")
+
+def db_main():
+    tornado.options.parse_command_line()
+    connection, db = qbt_server.connect_db()
+    
+    #create a user for testing
+    salt, encrypted_password = qbt_server.encrypt_password(None, options.password)
+    
+    if not db.users.find_one({'email':options.user_email}):
+        db.users.insert({'email':options.user_email, 'encrypted_password':encrypted_password, 'salt':salt})
+    
+    #create one mythical company
+    if not db.company_info.find_one({'sid':133}):
+        db.company_info.insert({'sid':133, "exchange" : "NEW YORK STOCK EXCHANGE", "symbol" : "JHF", "first date" : "01/04/1993", "last date" : "10/01/2008", "sid" : 133, "industry code" : "130A", "company name" : "JACK INC"})    
+    
+    #create one mythical company
+    if not db.company_info.find_one({'sid':134}):
+        db.company_info.insert({'sid':134, "exchange" : "NEW YORK STOCK EXCHANGE", "symbol" : "RCF", "first date" : "01/04/1993", "last date" : "10/01/2008", "sid" : 134, "industry code" : "130A", "company name" : "ROCCO INC"})    
+        
+    #create minute equity data collection and populate with a day of random data
+    prices = {133:25.0,134:45.0} #sid, initial price.
+    if not db.equity.trades.minute.find().count() == MINUTE_COUNT * len(prices):
+        db.equity.trades.minute.drop()
+        trade_start = datetime.datetime.now()
+        minute = datetime.timedelta(minutes=1)
+
+        for i in range(MINUTE_COUNT):
+            for sid,price in prices.iteritems():
+                price = price + random.uniform(-0.05,0.05)
+                db.equity.trades.minute.insert({'sid':sid, 'dt':trade_start + (minute * i),'price':price, 'volume':random.randrange(100,10000,100)})
+
+if __name__ == "__main__":
+    db_main()

qbt.py

@@ -1,10 +1,13 @@
 """
-qbt runs backtests using multiple processes and zeromq messaging. 
+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.PULL. Port = port_start + 1
+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 
@@ -14,6 +17,7 @@ zmq sockets for internal processing:
     - 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:
@@ -26,10 +30,98 @@ zmq sockets for backtest clients:
            
     
 """
+import copy
 import multiprocessing
 import zmq
 
+from backtest.util import *
+from data.sources.equity import *
+
+
+
+DATA_SINK_PORT      = 10000
+CONTROLLER_PORT     = 10099
+DATA_FEED_PORT      = 10002
+
 class Backtest(object):
     
-    def __init__(self, port_start=10000):
-        
+    def __init__(self, db, logger):
+        self.logger = logger
+        self.db = db
+        
+        
+    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)
+        #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()
+        self.logger.info("ds processes launched")
+       
+    def run(self):   
+        # Prepare our context and sockets
+        self.context = zmq.Context()
+        
+        #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 
+        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)
+        
+        
+        #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.last_event_time = None
+        self.data_workers = []
+        
+        last_dt = "."
+        event_q = []
+        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
+                
+            #no events in q at this point means we've processed all the data in all the sources!
+            #if(len(event_q) == 0):
+            #    return
+                
+            #event_q = sorted(event_q, key=lambda event: event['dt'])    
+            # we have the most recent message from each source, so we can only 
+            # be sure the first uninterrupted streak of messages from the same source 
+            # in the queue are most recent.
+            #while len(event_q) > 0:
+            #    cur_event = event_q.pop(0)
+            #    self.data_feed.send(json.dumps(cur_event))
+            #    last_dt = event_q[0]['dt']
+      

qbt_client.py

@@ -0,0 +1,20 @@
+import copy
+import multiprocessing
+import zmq
+
+from backtest.util import *
+
+class BacktestClient(object):
+    
+    def __init__(self,feed_port, logger):
+        self.context = zmq.Context()
+        
+        self.data_feed = self.context.socket(zmq.PULL)
+        self.data_feed.connect("tcp://localhost:{port}".format(port=feed_port))
+        self.logger = logger
+        
+    def run(self):
+        self.logger.info("running the client")
+        while True:
+            msg = self.data_feed.recv()
+            self.logger.info(msg)

qbt_data_bootstrap.py

@@ -1,30 +0,0 @@
-import tornado.auth
-import tornado.httpserver
-import tornado.ioloop
-from tornado.options import define, options
-import tornado.web
-import pymongo
-import bson
-import hashlib
-import base64
-import uuid
-import os
-import logging
-
-import qbt_server
-
-define("user_email", default="qbt@quantopian.com", help="email address for qbt user")
-define("password", default="foobar", help="password for qbt user")
-
-def db_main():
-    tornado.options.parse_command_line()
-    connection, db = qbt_server.connect_db()
-    
-    #create a user for testing
-    salt, encrypted_password = qbt_server.encrypt_password(None, options.password)
-    
-    if not db.users.find_one({'email':options.user_email}):
-        db.users.insert({'email':options.user_email, 'encrypted_password':encrypted_password, 'salt':salt})
-    
-if __name__ == "__main__":
-    db_main()

qbt_server.py

@@ -11,7 +11,10 @@ import uuid
 import os
 import logging
 import datetime
-import multiprocessing 
+import multiprocessing
+
+from qbt import *
+from qbt_client import * 
 
 logger = logging.getLogger()
 
@@ -56,7 +59,7 @@ class Application(tornado.web.Application):
             cookie_secret=base64.b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes),
             login_url="/login",
             #autoescape=None,
-            debug=True,
+            #debug=True,
         )
         tornado.web.Application.__init__(self, handlers, **settings)
 
@@ -70,6 +73,7 @@ class BaseHandler(tornado.web.RequestHandler):
         return self.application.db
 
     def get_current_user(self):
+        return "fawce"
         user_id = self.get_secure_cookie(u"user_id")
         logger.info("looking up user with id: {id}".format(id=user_id))
         if not user_id: return None
@@ -123,8 +127,16 @@ class BacktestHandler(BaseHandler):
     @tornado.web.authenticated
     def post(self):
         
+        bt = Backtest(self.db, logger)
+        #btc = BacktestClient(DATA_FEED_PORT, logger)
+        bt_proc = multiprocessing.Process(target=bt.run)
+        #btc_proc = multiprocessing.Process(target=btc.run)
+        bt_proc.start()
+        #btc_proc.start()
 
 
+
+    
 def main():
     tornado.options.parse_command_line()
     http_server = tornado.httpserver.HTTPServer(Application())

test.py

@@ -0,0 +1,19 @@
+import zmq
+from data.sources.equity import *
+
+def main():
+    context = zmq.Context()
+    controller = context.socket(zmq.SUB)
+    controller.connect("tcp://127.0.0.1:10099")    
+    controller.setsockopt(zmq.SUBSCRIBE, '')
+    while True:
+        try:
+            message = controller.recv()
+            print message
+        except zmq.ZMQError as err:
+            if err.errno != zmq.EAGAIN:
+                raise err
+     
+if __name__ == "__main__":
+    main()              
+     

test2.py

@@ -0,0 +1,11 @@
+import zmq
+
+def main():
+    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() 

test3.py

@@ -0,0 +1,12 @@
+import multiprocessing
+import test
+import test2
+
+def main():
+    proc1 = multiprocessing.Process(target=test2.main)
+    proc1.start()
+    proc2 = multiprocessing.Process(target=test.main)
+    proc2.start()
+    
+if __name__ == "__main__":
+    main()