From ddaefb9e1bf6c68973961a85f4096798b9829be1 Mon Sep 17 00:00:00 2001 From: fawce Date: Sun, 15 Jan 2012 14:53:33 -0500 Subject: [PATCH] synchronized data sources send events to feed. --- data/feed.py | 67 +++++++++++++++++++++++++++++++ data/sources/equity.py | 91 ++++++++++++++++++------------------------ qbt.py | 71 ++++++++++++++++---------------- test.py | 26 +++++++++++- 4 files changed, 165 insertions(+), 90 deletions(-) create mode 100644 data/feed.py diff --git a/data/feed.py b/data/feed.py new file mode 100644 index 00000000..22a2910f --- /dev/null +++ b/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() \ No newline at end of file diff --git a/data/sources/equity.py b/data/sources/equity.py index 9b8c1090..87cfb00e 100644 --- a/data/sources/equity.py +++ b/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") + + + diff --git a/qbt.py b/qbt.py index 3d5de978..9baa616d 100644 --- a/qbt.py +++ b/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): diff --git a/test.py b/test.py index bb5ce9c5..a85c345d 100644 --- a/test.py +++ b/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() \ No newline at end of file