From abce64914c126585306d659326c826ea50d506ea Mon Sep 17 00:00:00 2001 From: fawce Date: Sat, 14 Jan 2012 01:46:59 -0500 Subject: [PATCH] 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 | 12 ++- qbt_test_client.py => backtest/__init__.py | 0 backtest/util.py | 16 +++- data/__init__.py | 0 data/sources/__init__.py | 0 data/sources/equity.py | 95 +++++++++++++++++++ dev_setup.py | 54 +++++++++++ qbt.py | 104 +++++++++++++++++++-- qbt_client.py | 20 ++++ qbt_data_bootstrap.py | 30 ------ qbt_server.py | 16 +++- test.py | 19 ++++ test2.py | 11 +++ test3.py | 12 +++ 14 files changed, 349 insertions(+), 40 deletions(-) rename qbt_test_client.py => backtest/__init__.py (100%) create mode 100644 data/__init__.py create mode 100644 data/sources/__init__.py create mode 100644 data/sources/equity.py create mode 100644 dev_setup.py create mode 100644 qbt_client.py delete mode 100644 qbt_data_bootstrap.py create mode 100644 test.py create mode 100644 test2.py create mode 100644 test3.py diff --git a/README.md b/README.md index bb2f92a4..39ebcc45 100644 --- a/README.md +++ b/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 " from all sources. +3. All datasources send all events in their history from before , moving their internal pointer forward to the next unsent event. +4. QBT merges all events in memory +5. goto 1! diff --git a/qbt_test_client.py b/backtest/__init__.py similarity index 100% rename from qbt_test_client.py rename to backtest/__init__.py diff --git a/backtest/util.py b/backtest/util.py index 2d3f7990..592f7203 100644 --- a/backtest/util.py +++ b/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)) \ No newline at end of file + 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 \ No newline at end of file diff --git a/data/__init__.py b/data/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/data/sources/__init__.py b/data/sources/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/data/sources/equity.py b/data/sources/equity.py new file mode 100644 index 00000000..9b8c1090 --- /dev/null +++ b/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)) + + + + \ No newline at end of file diff --git a/dev_setup.py b/dev_setup.py new file mode 100644 index 00000000..77a7df84 --- /dev/null +++ b/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() diff --git a/qbt.py b/qbt.py index 30f6e3d9..3d5de978 100644 --- a/qbt.py +++ b/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): - \ No newline at end of file + 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'] + \ No newline at end of file diff --git a/qbt_client.py b/qbt_client.py new file mode 100644 index 00000000..a0e522b5 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/qbt_data_bootstrap.py b/qbt_data_bootstrap.py deleted file mode 100644 index 3f0365bf..00000000 --- a/qbt_data_bootstrap.py +++ /dev/null @@ -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() diff --git a/qbt_server.py b/qbt_server.py index d1d9c98f..13482fc8 100644 --- a/qbt_server.py +++ b/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()) diff --git a/test.py b/test.py new file mode 100644 index 00000000..bb5ce9c5 --- /dev/null +++ b/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() + \ No newline at end of file diff --git a/test2.py b/test2.py new file mode 100644 index 00000000..9e378aa4 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/test3.py b/test3.py new file mode 100644 index 00000000..7f89c9c1 --- /dev/null +++ b/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() \ No newline at end of file