From 4365852b92ac5557c95303261f0ad56f7f46fb36 Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 7 Feb 2012 22:16:17 -0500 Subject: [PATCH] reorganizing modules and scripts --- dev_setup.py | 54 ------- jenkins.sh | 34 ----- ordered_pip.sh | 11 -- requirements.txt | 8 - requirements_dev.txt | 14 -- requirements_sci.txt | 18 --- simulator/__init__.py | 0 simulator/backtest/__init__.py | 0 simulator/backtest/util.py | 162 --------------------- simulator/config.py | 43 ------ simulator/data/__init__.py | 0 simulator/data/feed.py | 117 --------------- simulator/data/sources/__init__.py | 0 simulator/data/sources/equity.py | 110 -------------- simulator/host_settings.py | 14 -- simulator/qbt.py | 67 --------- simulator/qbt_client.py | 55 ------- simulator/qbt_server.py | 148 ------------------- test/__init__.py | 0 test/test_messaging.py | 65 --------- transforms/__init__.py | 0 transforms/transforms.py | 225 ----------------------------- 22 files changed, 1145 deletions(-) delete mode 100644 dev_setup.py delete mode 100755 jenkins.sh delete mode 100755 ordered_pip.sh delete mode 100644 requirements.txt delete mode 100644 requirements_dev.txt delete mode 100644 requirements_sci.txt delete mode 100644 simulator/__init__.py delete mode 100644 simulator/backtest/__init__.py delete mode 100644 simulator/backtest/util.py delete mode 100644 simulator/config.py delete mode 100644 simulator/data/__init__.py delete mode 100644 simulator/data/feed.py delete mode 100644 simulator/data/sources/__init__.py delete mode 100644 simulator/data/sources/equity.py delete mode 100644 simulator/host_settings.py delete mode 100644 simulator/qbt.py delete mode 100644 simulator/qbt_client.py delete mode 100644 simulator/qbt_server.py delete mode 100644 test/__init__.py delete mode 100644 test/test_messaging.py delete mode 100644 transforms/__init__.py delete mode 100644 transforms/transforms.py diff --git a/dev_setup.py b/dev_setup.py deleted file mode 100644 index 29c085b6..00000000 --- a/dev_setup.py +++ /dev/null @@ -1,54 +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 datetime -import random - -import simulator.qbt_server as 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/jenkins.sh b/jenkins.sh deleted file mode 100755 index d47db64f..00000000 --- a/jenkins.sh +++ /dev/null @@ -1,34 +0,0 @@ -#!/bin/bash - -#setup virtualenvironment -export VIRTUALENVWRAPPER_PYTHON=/usr/bin/python2.7 -if [ ! -d $HOME/.venvs ]; then - mkdir $HOME/.venvs -fi -export WORKON_HOME=$HOME/.venvs -source /usr/local/bin/virtualenvwrapper.sh - -#create the scientific python virtualenv and copy to provide qsim base -mkvirtualenv --no-site-packages scientific_base -workon scientific_base -./ordered_pip.sh requirements_sci.txt -deactivate -#re-base qsim -#rmvirtualenv qsim -cpvirtualenv scientific_base qsim - -workon qsim -./ordered_pip.sh requirements.txt -./ordered_pip.sh requirements_dev.txt - -#setup the local mongodb -python dev_setup.py - -#run all the tests in test -nosetests --with-xcoverage --with-xunit --cover-erase --cover-package=simulator,transforms -pylint -f parseable . | tee pylint.out - -#run sloccount analysis -sloccount --wide --details ./ > sloccount.sc - -deactivate diff --git a/ordered_pip.sh b/ordered_pip.sh deleted file mode 100755 index ca3fa440..00000000 --- a/ordered_pip.sh +++ /dev/null @@ -1,11 +0,0 @@ -#!/bin/bash - -echo $hash -while read line -do - if [[ $line != \#* ]] ; then - #echo $line - pip install $line - fi -done < $1 -echo "Final line count is: $a"; diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 83b1ea93..00000000 --- a/requirements.txt +++ /dev/null @@ -1,8 +0,0 @@ -tornado - -#data source related -pymongo==2.1.1 - -#zeromq related -pyzmq==2.1.11 - diff --git a/requirements_dev.txt b/requirements_dev.txt deleted file mode 100644 index a22d34ae..00000000 --- a/requirements_dev.txt +++ /dev/null @@ -1,14 +0,0 @@ -# For debugger -fancycompleter==0.2 -pyrepl==0.8.2 -Pygments==1.4 -pdbpp==0.7.2 - -ipython==0.12 - -# For unit tests -nose==1.1.2 -unittest2==0.5.1 -requests==0.10.1 -nosexcover -pylint diff --git a/requirements_sci.txt b/requirements_sci.txt deleted file mode 100644 index 4ed7d007..00000000 --- a/requirements_sci.txt +++ /dev/null @@ -1,18 +0,0 @@ -#date related -pytz==2011n -python-dateutil==1.5 - -#core scientific python -numpy==1.6.1 -scipy==0.10.0 -matplotlib==1.1.0 -#http://sourceforge.net/projects/matplotlib/files/matplotlib/matplotlib-1.1.0/matplotlib-1.1.0.tar.gz -numexpr==2.0.1 -Cython==0.15.1 -tables==2.3.1 -scikits.statsmodels==0.3.1 -pandas==0.7.0rc1 - -#zeromq related -pyzmq==2.1.11 - diff --git a/simulator/__init__.py b/simulator/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/simulator/backtest/__init__.py b/simulator/backtest/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/simulator/backtest/util.py b/simulator/backtest/util.py deleted file mode 100644 index f4ec3652..00000000 --- a/simulator/backtest/util.py +++ /dev/null @@ -1,162 +0,0 @@ -""" -Small classes to assist with db access, timezone calculations, and so on. -""" - -import datetime -import pytz -import json -import logging -import uuid -import zmq - -class DocWrap(): - """ - Provides attribute access style on top of dictionary results from pymongo. - Allows you to access result['field'] as result.field. - Aliases result['_id'] to result.id. - - """ - def __init__(self, store=None): - if(store == None): - self.store = {} - else: - self.store = store.copy() - if(self.store.has_key('_id')): - self.store['id'] = self.store['_id'] - del(self.store['_id']) - - def __setitem__(self,key,value): - if(key == '_id'): - self.store['id'] = value - else: - self.store[key] = value - - def __getitem__(self, key): - if self.store.has_key(key): - return self.store[key] - - def __getattr__(self,attrname): - if self.store.has_key(attrname): - return self.store[attrname] - else: - 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 - parts = dt_str.split(".") - dt = datetime.datetime.strptime(parts[0], '%Y/%m/%d-%H:%M:%S').replace(microsecond=int(parts[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 - - -class ParallelBuffer(object): - """ holds several queues of events by key, allows retrieval in date order or by merging""" - def __init__(self, key_list): - self.out_socket = None - self.sent_count = 0 - self.received_count = 0 - self.draining = False - self.data_buffer = {} - for key in key_list: - self.data_buffer[key] = [] - - def __len__(self): - return len(self.data_buffer) - - def append(self, key, value): - self.data_buffer[key].append(value) - self.received_count += 1 - - def next(self): - if(not(self.is_full() or self.draining)): - return - - cur = None - earliest = None - for source, events in self.data_buffer.iteritems(): - if len(events) == 0: - continue - cur = events - if(earliest == None) or (cur[0]['dt'] <= earliest[0]['dt']): - earliest = cur - - if(earliest != None): - return earliest.pop(0) - - def is_full(self): - for source, events in self.data_buffer.iteritems(): - if (len(events) == 0): - return False - return True - - def pending_messages(self): - total = 0 - for source, events in self.data_buffer.iteritems(): - total += len(events) - return total - - def drain(self): - self.draining = True - while(self.pending_messages() > 0): - self.send_next() - - def send_next(self): - if(not(self.is_full() or self.draining)): - return - - event = self.next() - if(event != None): - self.out_socket.send(json.dumps(event)) - self.sent_count += 1 - - -class MergedParallelBuffer(ParallelBuffer): - - def __init__(self, keys): - ParallelBuffer.__init__(self, keys) - self.feed = [] - self.data_buffer["feed"] = self.feed - - def next(self): - if(not(self.is_full() or self.draining)): - return - - result = self.feed.pop(0) - for source, events in self.data_buffer.iteritems(): - if(source == "feed"): - continue - if(len(events) > 0): - cur = events.pop(0) - result[source] = cur['value'] - return result - - -class FeedSync(object): - - def __init__(self, feed, name): - self.feed = feed - self.id = "{name}-{id}".format(name=name, id=uuid.uuid1()) - self.feed.register_sync(self.id) - self.logger = logging.getLogger() - #self.logger.info("registered {id} with feed".format(id=self.id)) - - def confirm(self): - context = zmq.Context() - #synchronize with feed - sync_socket = context.socket(zmq.REQ) - sync_socket.connect(self.feed.sync_address) - # send a synchronization request to the feed - sync_socket.send(self.id) - # wait for synchronization reply from the feed - sync_socket.recv() - sync_socket.close() - context.term() - self.logger.info("sync'd feed from {id}".format(id = self.id)) \ No newline at end of file diff --git a/simulator/config.py b/simulator/config.py deleted file mode 100644 index 584e41b3..00000000 --- a/simulator/config.py +++ /dev/null @@ -1,43 +0,0 @@ -import json - -class Config(object): - """ Name/Value configuration object with type-safe accessors and json serialization/deserialization.""" - - def __init__(self, props): - self.store = props - - def __setitem__(self,key,value): - self.store[key] = value - - def __getitem__(self, key): - if self.store.has_key(key): - return self.store[key] - - def __getattr__(self,attrname): - if self.store.has_key(attrname): - return self.store[attrname] - else: - raise AttributeError("No attribute named {name}".format(name=attrname)) - - def get_integer(self, name, default=0): - return self.get_value(name, default, type(1)) - - def get_string(self, name, default=''): - return self.get_value(name, default, type('')) - - def get_float(self, name, default=0.0): - return self.get_value(name, default, type(1.0)) - - def get_value(self, name, default, expected_type): - if(self.store.has_key(name)): - val = self.store[name] - if isinstance(val, expected_type): - return val - else: - return default - - def to_json(self): - return json.dumps(self.store) - - def from_json(self, json_string): - self.store = json.loads(json_string) \ No newline at end of file diff --git a/simulator/data/__init__.py b/simulator/data/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/simulator/data/feed.py b/simulator/data/feed.py deleted file mode 100644 index 1f84d4ea..00000000 --- a/simulator/data/feed.py +++ /dev/null @@ -1,117 +0,0 @@ - -from simulator.data.sources.equity import * -from simulator.backtest.util import * -import time -import logging - -class DataFeed(object): - - def __init__(self, config): - self.logger = logging.getLogger() - - self.data_address = "tcp://127.0.0.1:{port}".format(port=10101) - self.sync_address = "tcp://127.0.0.1:{port}".format(port=10102) - self.feed_address = "tcp://127.0.0.1:{port}".format(port=10103) - - self.client_register = {} - - self.data_workers = {} - self.config = config - for name, info in config.iteritems(): - if(info['class'] == "EquityMinuteTrades"): - emt = EquityMinuteTrades(info['sid'], self, name) - self.data_workers[name] = emt - elif(info['class'] == "RandomEquityTrades"): - ret = RandomEquityTrades(info['sid'], self, name, info['count']) - self.data_workers[name] = ret - - self.data_buffer = ParallelBuffer(self.data_workers.keys()) - - def start_data_workers(self): - """Start a sub-process for each datasource.""" - for source_id, source in self.data_workers.iteritems(): - self.logger.info("starting {id}".format(id=source_id)) - source.start() - self.logger.info("ds processes launched") - - def register_sync(self, sync_id): - self.client_register[sync_id] = "UNCONFIRMED" - - def registration_complete(self): - for sync_id, status in self.client_register.iteritems(): - if status == "UNCONFIRMED": - return False - - return True - - def sync_clients(self): - # Socket to receive signals - self.logger.info("waiting for all datasources and clients to be ready") - self.syncservice = self.context.socket(zmq.REP) - self.syncservice.bind(self.sync_address) - - while not self.registration_complete(): - # wait for synchronization request - msg = self.syncservice.recv() - self.client_register[msg] = "CONFIRMED" - #self.logger.info("confirmed {id}".format(id=msg)) - # send synchronization reply - self.syncservice.send('CONFIRMED') - - self.syncservice.close() - self.logger.info("sync'd all datasources and clients") - - def run(self): - # Prepare our context and sockets - self.context = zmq.Context() - - ds_finished_counter = 0 - - #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) - - #create the feed - self.feed_socket = self.context.socket(zmq.PUB) - self.feed_socket.bind(self.feed_address) - - self.data_buffer.out_socket = self.feed_socket - - #start the data source workers - self.start_data_workers() - - #wait for all feed subscribers - self.sync_clients() - - self.logger.info("entering feed loop on {addr}".format(addr=self.data_address)) - - while True: - message = self.data_socket.recv() - event = json.loads(message) - if(event["type"] == "DONE"): - ds_finished_counter += 1 - if(len(self.data_workers) == ds_finished_counter): - break - else: - self.data_buffer.append(event[u's'], event) - self.data_buffer.send_next() - - - #drain any remaining messages in the buffer - self.data_buffer.drain() - - #send the DONE message - self.feed_socket.send("DONE") - self.logger.info("received {n} messages, sent {m} messages".format(n=self.data_buffer.received_count, m=self.data_buffer.sent_count)) - self.data_socket.close() - self.feed_socket.close() - self.context.term() - - - - - - - - diff --git a/simulator/data/sources/__init__.py b/simulator/data/sources/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/simulator/data/sources/equity.py b/simulator/data/sources/equity.py deleted file mode 100644 index a3ef1d62..00000000 --- a/simulator/data/sources/equity.py +++ /dev/null @@ -1,110 +0,0 @@ -import datetime -import zmq -import pymongo -import pymongo.json_util -import json -import pytz -import copy -import multiprocessing -import logging -import random -from pymongo import ASCENDING, DESCENDING - -from simulator.backtest.util import * - -from simulator.qbt_server import * #connect_db - -class DataSource(object): - def __init__(self, feed, source_id): - self.source_id = source_id - self.logger = logging.getLogger() - self.feed = feed - self.sync = 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)) - self.cur_event = None - - def start(self): - 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)) - self.context = zmq.Context() - - #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 - self.data_socket = self.context.socket(zmq.PUSH) - self.data_socket.connect(self.data_address) - - #signal we are ready - self.sync.confirm() - - def run(self): - try: - self.open() - self.send_all() - self.close() - except Exception as err: - self.logger.error(err, "Unexpected failure running datasource - {name}.".format(name=name)) - - def send(self, event): - event['s'] = self.source_id - event['type'] = 'event' - self.data_socket.send(json.dumps(event)) - - def close(self): - 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") - -class EquityMinuteTrades(DataSource): - - def __init__(self, sid, feed, source_id): - self.sid = sid - self.connection, self.db = connect_db() - DataSource.__init__(self, feed, source_id) - - - def send_all(self): - 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())) - - 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 - del(event['_id']) - self.send(event) - - - -class RandomEquityTrades(DataSource): - - def __init__(self, sid, feed, source_id, count): - DataSource.__init__(self, feed, source_id) - self.count = count - self.sid = sid - - def send_all(self): - trade_start = datetime.datetime.now() - minute = datetime.timedelta(minutes=1) - 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':format_date(trade_start + (minute * i)),'price':price, 'volume':random.randrange(100,10000,100)} - self.send(event) - - - - - diff --git a/simulator/host_settings.py b/simulator/host_settings.py deleted file mode 100644 index 23c9efdc..00000000 --- a/simulator/host_settings.py +++ /dev/null @@ -1,14 +0,0 @@ -class Config(object): - def __init__(self, dct): - self.__dict__.update(dct) - -mongo_conn_args = Config({ - 'mongodb_host' : 'claire.mongohq.com', - 'mongodb_port' : 10087, - 'mongodb_dbname' : 'quantodata-staging', - 'mongodb_user' : 'quantopian', - 'mongodb_password' : 'quantopian', -}) - -root_url = 'http://localhost:8000' -ws_url = 'ws://localhost:8001' diff --git a/simulator/qbt.py b/simulator/qbt.py deleted file mode 100644 index 1835b931..00000000 --- a/simulator/qbt.py +++ /dev/null @@ -1,67 +0,0 @@ -""" -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.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 - aggregators, merging agg results, and transmitting consolidated stream to event feed. - - agg source, ZMQ.PUSH. Port = port_start + 2 - - agg sink, ZMQ.PULL. Port = port_start + 3 - - 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: -================================= - - orders sink, ZMQ.RESP. Port = port_start + 5 - - backtest will connect (can you bind?) to this socket and await orders from the client. Order data will be processed against the streaming datafeed. - - event feed, ZMQ.RESP. Port = port_start + 6 - - backtest will bind to this socket and respond to requests from client for more data. Response data will be the queue of events that - transpired since the last request. - - -""" -import copy -import multiprocessing -import zmq - -from backtest.util import * -from data.sources.equity import * - - - -CONTROLLER_PORT = 9000 -DATA_SINK_PORT = 10000 -DATA_FEED_PORT = 30000 - -class Backtest(object): - - def __init__(self, db, logger): - self.logger = logger - self.db = db - self.feed = DataFeed(db, logger) - - def start_feed(self): - proc1 = multiprocessing.Process(target=feed.run) - proc1.start() - - def run(self): - # Prepare our context and sockets - self.context = zmq.Context() - - #create the feed sink. - self.feed_address = self.feed.data_address - self.feed_socket = self.context.connect(self.feed_address) - self.feed_socket.connect(zmq.PULL) - - - \ No newline at end of file diff --git a/simulator/qbt_client.py b/simulator/qbt_client.py deleted file mode 100644 index eda31fb6..00000000 --- a/simulator/qbt_client.py +++ /dev/null @@ -1,55 +0,0 @@ -import copy -import multiprocessing -import zmq -import logging -import json - -from backtest.util import * - -class TestClient(object): - - def __init__(self,feed, address, bind=False): - self.logger = logging.getLogger() - self.feed = feed - self.address = address - self.sync = FeedSync(feed, "testclient") - self.bind = bind - self.received_count = 0 - - def run(self): - - self.logger.info("running the client") - self.context = zmq.Context() - - self.data_feed = self.context.socket(zmq.PULL) - - if(self.bind): - self.logger.info("binding to {address}".format(address=self.address)) - self.data_feed.bind(self.address) - else: - self.logger.info("connecting to {address}".format(address=self.address)) - self.data_feed.connect(self.address) - - self.sync.confirm() - - self.logger.info("Starting the client loop") - - prev_dt = None - while True: - msg = self.data_feed.recv() - if(msg == "DONE"): - self.logger.info("DONE!") - break - self.received_count += 1 - event = json.loads(msg) - if(prev_dt != None): - if(not event['dt'] >= prev_dt): - raise Exception("message arrived out of order: {date} after {prev}".format(date=event['dt'], prev=prev_dt)) - - prev_dt = event['dt'] - if(self.received_count % 100 == 0): - self.logger.info("received {n} messages".format(n=self.received_count)) - - self.logger.info("received {n} messages".format(n=self.received_count)) - self.data_feed.close() - self.context.term() \ No newline at end of file diff --git a/simulator/qbt_server.py b/simulator/qbt_server.py deleted file mode 100644 index 13482fc8..00000000 --- a/simulator/qbt_server.py +++ /dev/null @@ -1,148 +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 datetime -import multiprocessing - -from qbt import * -from qbt_client import * - -logger = logging.getLogger() - -define("port", default=8888, help="run the qbt on the given port", type=int) -define("mongodb_host", default="127.0.0.1", help="mongodb host address") -define("mongodb_port", default=27017, help="connect to the mongodb on the given port", type=int) -define("mongodb_dbname", default="qbt", help="database name") -define("mongodb_user", default="qbt", help="database user") -define("mongodb_password", default="qbt", help="database password") - -HASH_ALGO = 'sha256' - -def connect_db(): - connection = pymongo.Connection(options.mongodb_host, options.mongodb_port) - db = connection[options.mongodb_dbname] - db.authenticate(options.mongodb_user, options.mongodb_password) - return connection, db - -def encrypt_password(salt, password): - if(salt == None): - h1 = hashlib.new(HASH_ALGO) - h1.update(str(datetime.datetime.utcnow())+"--"+password) - salt = h1.hexdigest() - - h2 = hashlib.new(HASH_ALGO) - h2.update(salt+"--"+password) - encrypted_password = h2.hexdigest() - - return salt, encrypted_password - -class Application(tornado.web.Application): - def __init__(self): - handlers = [ - (r"/", MainHandler), - (r"/login", LoginHandler), - (r"/backtest", BacktestHandler) - ] - settings = dict( - template_path=os.path.join(os.path.dirname(__file__), "templates"), - static_path=os.path.join(os.path.dirname(__file__), "static"), - xsrf_cookies=False, - cookie_secret=base64.b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes), - login_url="/login", - #autoescape=None, - #debug=True, - ) - tornado.web.Application.__init__(self, handlers, **settings) - - # Have one global connection to the blog DB across all handlers - self.connection, self.db = connect_db() - - -class BaseHandler(tornado.web.RequestHandler): - @property - def db(self): - 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 - #get user record by id - users = self.db.users.find(spec={"_id":bson.ObjectId(user_id)}, limit=1) - if(users.count > 0): - return users[0] - return None - -class MainHandler(BaseHandler): - @tornado.web.authenticated - def get(self): - self.write("Hello, world. Try launching a backtest.") - -class LoginHandler(BaseHandler): - def get(self): - self.write('
' - 'Name: ' - 'pass: ' - '' - '
') - - def post(self): - self.authenticate(self.get_argument("user_name"),self.get_argument("password")) - self.redirect("/") - - def authenticate(self, username, password): - h = hashlib.new(HASH_ALGO) - #find user record by username. - users = self.db.users.find(spec={"email":username}, limit=1) - if(users.count > 0): - user_record = users[0] - else: - logger.debug("no user with name: {username}", username=username) - return - - - #calculate password hash - salt, encrypted_password = encrypt_password(user_record['salt'], password) - if (user_record['encrypted_password'] == encrypted_password): - #we have a match, so set the secure cookie to the salt - logger.debug("setting user_id cookie to {id}".format(id=user_record['_id'])) - self.set_secure_cookie(u"user_id", unicode(user_record['_id'])) - -class BacktestHandler(BaseHandler): - @tornado.web.authenticated - def get(self): - self.write('
' - '' - '
') - @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()) - http_server.listen(options.port) - tornado.ioloop.IOLoop.instance().start() - - -if __name__ == "__main__": - main() diff --git a/test/__init__.py b/test/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/test/test_messaging.py b/test/test_messaging.py deleted file mode 100644 index 12d547b6..00000000 --- a/test/test_messaging.py +++ /dev/null @@ -1,65 +0,0 @@ -import unittest2 as unittest -import zmq -import logging -import tornado -from simulator.data.sources.equity import * -from simulator.data.feed import * -from transforms.transforms import MergedTransformsFeed, MovingAverage - -from simulator.qbt_client import TestClient - - -class MessagingTestCase(unittest.TestCase): - - def setUp(self): - self.total_data_count = 800 - self.feed_config = {'emt1':{'sid':133, 'class':'RandomEquityTrades', 'count':400}, - 'emt2':{'sid':134, 'class':'RandomEquityTrades', 'count':400}} - self.feed = DataFeed(self.feed_config) - self.feed_proc = multiprocessing.Process(target=self.feed.run) - - self.config = {} - self.config['name'] = '**merged feed**' - self.config['transforms'] = [{'name':'mavg1', 'class':'MovingAverage', 'hours':1},{'name':'mavg2', 'class':'MovingAverage', 'hours':2}] - - def test_client(self): - #subscribe a client to the transformed feed - client = TestClient(self.feed, self.feed.feed_address) - - feed_proc = multiprocessing.Process(target=self.feed.run) - feed_proc.start() - - - client.run() - self.assertEqual(self.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages, found {n} remaining.".format(n=self.feed.data_buffer.pending_messages())) - self.assertEqual(self.total_data_count, client.received_count, "The client should have received ({n}) the same number of messages as the feed sent ({m}).".format(n=client.received_count, m=self.total_data_count)) - - - def dtest_moving_average_to_client(self): - mavg = MovingAverage(self.feed, self.config['transforms'][0]) - mavg_proc = multiprocessing.Process(target=mavg.run) - mavg_proc.start() - - client = TestClient(self.feed, mavg.result_address, bind=True) - - feed_proc = multiprocessing.Process(target=self.feed.run) - feed_proc.start() - - client.run() - self.assertEqual(self.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages.") - self.assertEqual(self.total_data_count, client.received_count, "The client should have received the same number of messages as the feed sent.") - - def dtest_merged_to_client(self): - merger = MergedTransformsFeed(self.feed, self.config) - merger_proc = multiprocessing.Process(target=merger.run) - merger_proc.start() - - client = TestClient(self.feed, merger.result_address) - - feed_proc = multiprocessing.Process(target=self.feed.run) - feed_proc.start() - - client.run() - self.assertEqual(self.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages.") - self.assertEqual(self.total_data_count, client.received_count, "The client should have received the same number of messages as the feed sent.") - diff --git a/transforms/__init__.py b/transforms/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/transforms/transforms.py b/transforms/transforms.py deleted file mode 100644 index b578fd16..00000000 --- a/transforms/transforms.py +++ /dev/null @@ -1,225 +0,0 @@ -import zmq -import logging -import datetime -import json -import copy -import multiprocessing -from simulator.backtest.util import * -import simulator.config as config -class Transform(object): - """Parent class for feed transforms. Subclass to create a new derived value from the combined feed.""" - - def __init__(self, feed, config_dict, result_address): - """ - feed_address - zmq socket address, Transform will CONNECT a PULL socket and receive messages until "DONE" is received. - result_address - zmq socket address, Transform will CONNECT a PUSH socket and send messaes until feed_socket receives "DONE" - sync_address - zmq socket address, Transform will CONNECT a REQ socket and send/receive one message before entering feed loop - config - must be a dict that can be wrapped in a config.Config object with at least an entry for 'name':string value - server - if True, transform will bind to the result address (and act as a server), if False it will connect. The - the last transform in a series should be server=True so that clients can connect. - """ - self.logger = logging.getLogger() - self.feed = feed - self.feed_address = feed.feed_address - self.result_address = result_address - self.config = config.Config(config_dict) - self.name = self.config.get_string('name') - self.sync = FeedSync(feed, self.name) - self.state = {} - self.state['name'] = self.name - self.received_count = 0 - self.sent_count = 0 - - def run(self): - self.open() - self.process_all() - self.close() - - def open(self): - self.context = zmq.Context() - - self.logger.info("starting {name} transform".format(name = self.name)) - #create the feed SUB. - self.feed_socket = self.context.socket(zmq.SUB) - self.feed_socket.connect(self.feed_address) - self.feed_socket.setsockopt(zmq.SUBSCRIBE,'') - - #create the result PUSH - self.result_socket = self.context.socket(zmq.PUSH) - self.result_socket.connect(self.result_address) - - def process_all(self): - self.logger.info("starting {name} event loop".format(name = self.name)) - self.sync.confirm() - - while True: - message = self.feed_socket.recv() - if(message == "DONE"): - self.logger.info("{name} received the Done message from the feed".format(name=self.name)) - self.result_socket.send("DONE") - break; - self.received_count += 1 - event = json.loads(message) - cur_state = self.transform(event) - cur_state['dt'] = event['dt'] - cur_state['name'] = self.name - self.result_socket.send(json.dumps(cur_state)) - self.sent_count += 1 - - def close(self): - self.logger.info("Transform {name} recieved {r} and sent {s}".format(name=self.name, r=self.received_count, s=self.sent_count)) - - self.feed_socket.close() - self.result_socket.close() - self.context.term() - - def transform(self, event): - return {} - - -class MovingAverage(Transform): - - def __init__(self, feed, props, result_address): - Transform.__init__(self, feed, props, result_address) - self.events = [] - - self.window = datetime.timedelta(days = self.config.get_integer('days'), - seconds = self.config.get_integer('seconds'), - microseconds = self.config.get_integer('microseconds'), - milliseconds = self.config.get_integer('milliseconds'), - minutes = self.config.get_integer('minutes'), - hours = self.config.get_integer('hours'), - weeks = self.config.get_integer('weeks')) - - - - - def transform(self, event): - self.events.append(event) - - #filter the event list to the window length. - self.events = [x for x in self.events if (parse_date(x['dt']) - parse_date(event['dt'])) <= self.window] - - if(len(self.events) == 0): - return 0.0 - - total = 0.0 - for event in self.events: - total += event['price'] - - self.average = total/len(self.events) - - self.state['value'] = self.average - - return self.state - - -class MergedTransformsFeed(Transform): - """ Merge data feed and array of transform feeds into a single result vector. - PULL from feed - PULL from child transforms - PUSH merged message to client - - """ - - def __init__(self, feed, props): - """ - config - must have an entry for 'transforms':array of dicts, which are convertedto configs. - """ - Transform.__init__(self, feed, props, "tcp://127.0.0.1:20202") - self.transform_address = "tcp://127.0.0.1:{port}".format(port=10104) - self.transform_socket = None - self.create_transforms(self.config.transforms) - - - def create_transforms(self, configs): - self.transforms = {} - for props in configs: - class_name = props['class'] - if(class_name == 'MovingAverage'): - mavg = MovingAverage(self.feed, props, self.transform_address) - self.transforms[mavg.name] = mavg - - keys = copy.copy(self.transforms.keys()) - keys.append("feed") #for the raw feed - self.data_buffer = MergedParallelBuffer(keys) - - self.buffers = {} - for name, transform in self.transforms.iteritems(): - self.buffers[name] = [] - - def open(self): - self.context = zmq.Context() - - self.logger.info("starting {name} transform".format(name = self.name)) - #create the feed SUB. - self.feed_socket = self.context.socket(zmq.SUB) - self.feed_socket.connect(self.feed_address) - self.feed_socket.setsockopt(zmq.SUBSCRIBE,'') - - #create the result PUSH - self.result_socket = self.context.socket(zmq.PUSH) - self.result_socket.bind(self.result_address) - - #create the transform PULL. - self.transform_socket = self.context.socket(zmq.PULL) - self.transform_socket.bind(self.transform_address) - self.data_buffer.out_socket = self.result_socket - - # Initialize poll set - self.poller = zmq.Poller() - self.poller.register(self.feed_socket, zmq.POLLIN) - self.poller.register(self.transform_socket, zmq.POLLIN) - - for name, transform in self.transforms.iteritems(): - self.logger.info("starting {name}".format(name=name)) - proc = multiprocessing.Process(target=transform.run) - proc.start() - - self.sync.confirm() - - def close(self): - self.transform_socket.close() - Transform.close(self) - - def process_all(self): - - done_count = 0 - while True: - socks = dict(self.poller.poll()) - - if self.feed_socket in socks and socks[self.feed_socket] == zmq.POLLIN: - message = self.feed_socket.recv() - if(message == "DONE"): - self.logger.info("finished receiving feed to merge") - done_count += 1 - else: - self.received_count += 1 - event = json.loads(message) - self.data_buffer.append("feed",event) - - if self.transform_socket in socks and socks[self.transform_socket] == zmq.POLLIN: - t_message = self.transform_socket.recv() - if(t_message == "DONE"): - self.logger.info("finished receiving a transform to merge") - done_count += 1 - else: - self.received_count += 1 - t_event = json.loads(t_message) - self.data_buffer.append(t_event['name'], t_event) - - if(done_count >= len(self.data_buffer)): - break #done! - - self.data_buffer.send_next() - - self.logger.info("Transform {name} received {r} and sent {s}".format(name=self.name, r=self.data_buffer.received_count, s=self.data_buffer.sent_count)) - self.logger.info("about to drain {n} messages from merger's buffer".format(n=self.data_buffer.pending_messages())) - - #drain any remaining messages in the buffer - self.data_buffer.drain() - - #signal to client that we're done - self.result_socket.send("DONE") - self.logger.info("Transform {name} received {r} and sent {s}".format(name=self.name, r=self.data_buffer.received_count, s=self.data_buffer.sent_count)) -