mirror of
https://github.com/wassname/catalyst.git
synced 2026-08-07 11:20:19 +08:00
reorganizing modules and scripts
This commit is contained in:
fawce
@@ -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()
|
||||
-34
@@ -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
|
||||
@@ -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";
|
||||
@@ -1,8 +0,0 @@
|
||||
tornado
|
||||
|
||||
#data source related
|
||||
pymongo==2.1.1
|
||||
|
||||
#zeromq related
|
||||
pyzmq==2.1.11
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -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'
|
||||
@@ -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)
|
||||
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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 <a href='/backtest'>backtest</a>.")
|
||||
|
||||
class LoginHandler(BaseHandler):
|
||||
def get(self):
|
||||
self.write('<html><body><form action="/login" method="post">'
|
||||
'Name: <input type="text" name="user_name">'
|
||||
'pass: <input type="password" name="password">'
|
||||
'<input type="submit" value="Sign in">'
|
||||
'</form></body></html>')
|
||||
|
||||
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('<html><body><form action="/backtest" method="post">'
|
||||
'<input type="submit" value="Launch">'
|
||||
'</form></body></html>')
|
||||
@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()
|
||||
@@ -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.")
|
||||
|
||||
@@ -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))
|
||||
|
||||
Reference in New Issue
Block a user