diff --git a/dataloader.py b/dataloader.py deleted file mode 100644 index 5bd0c0d6..00000000 --- a/dataloader.py +++ /dev/null @@ -1,39 +0,0 @@ -import datetime -import sys -import zipline.util as qutil -from zipline.finance.data import DataLoader - -def print_usage(): - print """ - Usage is: - python loaddata.py (pt | lt | lh | ld | ei | bm | si | help) - - pt - purge trade collection from the db - lt - load trades (minute bars) to the db - lh - load trades (hour bars) to the db - ld - load trades (daily close) to the db - ei - ensure all indexes on all collections in tick and algo db - tr - load treasury rates - bm - load benchmark data - si - load security info (sid, symbol, qualifier) - help - display this message - """ - - -if __name__ == "__main__": - - if len(sys.argv) == 2: - qutil.configure_logging() - operation = sys.argv[1] - if(operation not in['pt','lt','lh','ld','ei','si', 'tr','bm'] or operation == 'help'): - print_usage() - else: - ts = datetime.datetime.now().strftime("%Y%m%d%H%M%S") - pidfile = "/tmp/loaddata-{stamp}.pid".format(stamp=ts) - daemon = DataLoader(pidfile,operation) - qutil.LOGGER.info("DataLoader starting.") - daemon.run() - sys.exit(0) - else: - print_usage() - sys.exit(2) diff --git a/etc/jenkins.sh b/etc/jenkins.sh index fc6e218d..5b304092 100755 --- a/etc/jenkins.sh +++ b/etc/jenkins.sh @@ -25,16 +25,9 @@ workon zipline # Show what we have installed pip freeze -#copy the host_settings file into place -cp /mnt/jenkins/zipline_host_settings.py ./host_settings.py - #documentation output paver apidocs html -#load treasury data -python dataloader.py tr -#load benchmark data -python dataloader.py bm #run all the tests in test. see setup.cfg for flags. nosetests diff --git a/zipline/daemon.py b/zipline/daemon.py deleted file mode 100644 index 82f3135a..00000000 --- a/zipline/daemon.py +++ /dev/null @@ -1,143 +0,0 @@ -""" -Daemon class, based on the excellent article: -http://www.jejik.com/articles/2007/02/a_simple_unix_linux_daemon_in_python/ -""" - -import sys, os, time, atexit -from signal import SIGTERM, SIGINT - -class Daemon: - """ - A generic daemon class. - - Usage: subclass the Daemon class and override the run() method - """ - def __init__(self, pidfile, stdin='/dev/null', stdout='/dev/null', stderr='/dev/null'): - self.stdin = stdin - self.stdout = stdout - self.stderr = stderr - self.pidfile = pidfile - - def daemonize(self): - """ - do the UNIX double-fork magic, see Stevens' "Advanced - Programming in the UNIX Environment" for details (ISBN 0201563177) - http://www.erlenstar.demon.co.uk/unix/faq_2.html#SEC16 - """ - try: - pid = os.fork() - if pid > 0: - # exit first parent - sys.exit(0) - except OSError, e: - sys.stderr.write("fork #1 failed: %d (%s)\n" % (e.errno, e.strerror)) - sys.exit(1) - - # decouple from parent environment - os.chdir("/") - os.setsid() - os.umask(0) - - # do second fork - try: - pid = os.fork() - if pid > 0: - # exit from second parent - sys.exit(0) - except OSError, e: - sys.stderr.write("fork #2 failed: %d (%s)\n" % (e.errno, e.strerror)) - sys.exit(1) - - # redirect standard file descriptors - sys.stdout.flush() - sys.stderr.flush() - si = file(self.stdin, 'r') - so = file(self.stdout, 'a+') - se = file(self.stderr, 'a+', 0) - os.dup2(si.fileno(), sys.stdin.fileno()) - os.dup2(so.fileno(), sys.stdout.fileno()) - os.dup2(se.fileno(), sys.stderr.fileno()) - - # write pidfile - atexit.register(self.delpid) - pid = str(os.getpid()) - file(self.pidfile,'w+').write("%s\n" % pid) - - def delpid(self): - os.remove(self.pidfile) - - def start(self): - """ - Start the daemon - """ - # Check for a pidfile to see if the daemon already runs - try: - pf = file(self.pidfile,'r') - pid = int(pf.read().strip()) - pf.close() - except IOError: - pid = None - - if pid: - message = "pidfile %s already exist. Daemon already running?\n" - sys.stderr.write(message % self.pidfile) - sys.exit(1) - - # Start the daemon - self.daemonize() - try: - signal.signal(signal.SIGINT, self.handle_kill) - except Exception, err: - print "Problem with sigint signup " + str(err) - self.run() - - def stop(self): - """ - Stop the daemon - """ - # Get the pid from the pidfile - try: - pf = file(self.pidfile,'r') - pid = int(pf.read().strip()) - pf.close() - except IOError: - pid = None - - if not pid: - message = "pidfile %s does not exist. Daemon not running?\n" - sys.stderr.write(message % self.pidfile) - return # not an error in a restart - - # First signal the process that we need to interrupt, so it can do things like close child procs - try: - os.kill(pid, SIGINT) - time.sleep(2.0) #Give the process some time to kill... - except OSError, err: - print "Error trying to sigint the process" + str(err) - - # Try killing the daemon process - try: - while 1: - os.kill(pid, SIGTERM) - time.sleep(0.1) - except OSError, err: - err = str(err) - if err.find("No such process") > 0: - if os.path.exists(self.pidfile): - os.remove(self.pidfile) - else: - print str(err) - sys.exit(1) - - def restart(self): - """ - Restart the daemon - """ - self.stop() - self.start() - - def run(self): - """ - You should override this method when you subclass Daemon. It will be called after the process has been - daemonized by start() or restart(). - """ diff --git a/zipline/db.py b/zipline/db.py deleted file mode 100644 index 37b78548..00000000 --- a/zipline/db.py +++ /dev/null @@ -1,76 +0,0 @@ -import atexit -import pymongo -import zipline.util as qutil - -class MongoOptions(object): - - def __init__(self, host, port, dbname, user, password): - self.mongodb_host = host - self.mongodb_port = port - self.mongodb_dbname = dbname - self.mongodb_user = user - self.mongodb_password = password - -class NoDatabase(Exception): - def __repr__(self): - return 'The database has not been set up yet.' - -def setup_db(credentials): - """ - Setup the database. Has global side effects. - """ - qutil.LOGGER.info(dir(DbConnection)) - if not DbConnection.initd: - connector = connect_db(credentials) - DbConnection.set(*connector) - -def connect_db(options): - """ - Connect to pymongo, return a connection and database instance - as a tuple. - """ - - connection = pymongo.Connection(options.mongodb_host, options.mongodb_port) - - db = connection[options.mongodb_dbname] - db.authenticate(options.mongodb_user, options.mongodb_password) - - def _gc_connection(): # pragma: no cover - connection.close() - - atexit.register(_gc_connection) - return connection, db - -class DbConnection(object): - """ - Hold the shared state of the database connection. - """ - - initd = False - __shared = {} - - def __init__(self): - self.__dict__ = self.__shared - - @staticmethod - def set(conn, db): - DbConnection.__shared['conn'] = conn - DbConnection.__shared['db'] = db - DbConnection.initd = True - - @staticmethod - def get(): - return ( - DbConnection.__shared['conn'], - DbConnection.__shared['db'] - ) - - def __getattr__(self, key): - if not DbConnection.__shared.get('initd'): - raise NoDatabase() - else: - return DbConnection.__shared.get(key) - - def destory(self): # pragma: no cover - DbConnection.__shared['initd'] = False - self.conn.close() diff --git a/zipline/finance/data.py b/zipline/finance/data.py deleted file mode 100644 index 4cc31afb..00000000 --- a/zipline/finance/data.py +++ /dev/null @@ -1,498 +0,0 @@ -import sys -import logging -import datetime -import sys -import os -import pymongo -import csv -import re -import copy -import datetime -import time -import pytz -import shutil -import urllib -import subprocess -from pymongo import ASCENDING, DESCENDING -from zipline.daemon import Daemon -import zipline.util as qutil -import zipline.db as db -import host_settings - -class FinancialDataLoader(): - """ - Load trade and quote data from tickdata extracts into the db. - Dates and times in the extracts must be in GMT. - - All data extract files are expected to be in $HOME/fdl/. The expected directory layout is:: - /benchmark.csv -- this will be created from yahoo data each time load_bench_marks is run - /interest_rates.csv -- - """ - BATCH_SIZE = 100 - - def __init__(self): - self.conn, self.db = db.DbConnection.get() - self.data_file_path = os.environ['HOME'] + "/fdl/" - subprocess.call("mkdir {data_dir}".format(data_dir=self.data_file_path), shell=True) - self.last_bm_close = None - - def load_bench_marks(self): - """Fetches the S&P end of day pricing history from yahoo, loads it to db.bench_marks""" - start = time.time() - start_date = datetime.datetime(year=1950, month=1, day=3) - end_date = datetime.datetime.utcnow() - file_path = os.path.join(self.data_file_path, "benchmark.csv") - fp = open(file_path + ".tmp", "wb") - - #create benchmark files - #^GSPC 19500103 - query = {} - query['s'] = "^GSPC" #the s&p 500 - query['d'] = end_date.month - 1 # end_date month, zero indexed - query['e'] = end_date.day # end_date day str(int(todate[6:8])) #day - query['f'] = end_date.year #end_date year str(int(todate[0:4])) - query['g'] = "d" #daily frequency - query['a'] = start_date.month - 1 #start_date month, zero indexed - query['b'] = start_date.day #start_date day - query['c'] = start_date.year #start_date year - - #print query - params = urllib.urlencode(query) - params += "&ignore=.csv" - - url = "http://ichart.yahoo.com/table.csv?%s" % params - qutil.LOGGER.info("fetching {url}".format(url=url)) - f = urllib.urlopen(url) - fp.write(f.read()) - fp.close() - qutil.LOGGER.info("fetched {url} Reversing.".format(url=url)) - - tmp_file = file_path + ".tmp" - reversed_tmp_file = file_path + ".rev" - - rcode = subprocess.call("tac {oldfile} > {newfile}".format(oldfile=tmp_file, newfile=reversed_tmp_file), shell=True) - #on mac, there is no tac command, so use tail -r (which isn't available on debian) - if rcode != 0: - rcode = subprocess.call("tail -r {oldfile} > {newfile}".format(oldfile=tmp_file, newfile=reversed_tmp_file), shell=True) - - #tail -1 benchmark.csv.rev > benchmark.csv - subprocess.call("echo \"date,open,high,low,close,volume,adj_close\" > {result}".format(newfile=reversed_tmp_file, result=file_path), shell=True) - #sed '$d' < ~/fdl/benchmark.csv.rev >> ~/fdl/benchmark.csv - subprocess.call("sed '$d' < {newfile} >> {result}".format(newfile=reversed_tmp_file, result=file_path), shell=True) - #clean up working files - subprocess.call("rm {file}".format(file=tmp_file), shell=True) - subprocess.call("rm {file}".format(file=reversed_tmp_file), shell=True) - - #load the records into mongodb - self.db.bench_marks.drop() - qutil.LOGGER.info("processing benchmark info") - self.parse_file(self.db.bench_marks, - self.bench_mark_cb, - file_path, - ['date','open','high','low','close','volume','adj_close'], - None, - 0) - qutil.LOGGER.info("benchmark info complete") - total = time.time() - start - qutil.LOGGER.info("%d seconds to load benchmark history" % total) - - def load_treasuries(self): - """fetches data from the treasury.gov yield curve website, and populates the treasury_curves table. - - to explore data available from the treasury: - http://www.treasury.gov/resource-center/data-chart-center/interest-rates/Pages/TextView.aspx?data=yield - - to fetch xml of all daily yield curves: - http://data.treasury.gov/feed.svc/DailyTreasuryYieldCurveRateData - """ - - from xml.dom.minidom import parse - self.db.treasury_curves.drop() - path = os.path.join(self.data_file_path + "all_treasury_rates.xml") - #download all data to local filesystem - subprocess.call("curl http://data.treasury.gov/feed.svc/DailyTreasuryYieldCurveRateData > {path}".format(path=path), shell=True) - dom = parse(path) - - - entries = dom.getElementsByTagName("entry") - for entry in entries: - curve = {} - curve['tid'] = self.get_node_value(entry, "d:Id") - - curve['date'] = self.get_treasury_date(self.get_node_value(entry, "d:NEW_DATE")) - curve['1month'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_1MONTH")) - curve['3month'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_3MONTH")) - curve['6month'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_6MONTH")) - curve['1year'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_1YEAR")) - curve['2year'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_2YEAR")) - curve['3year'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_3YEAR")) - curve['5year'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_5YEAR")) - curve['7year'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_7YEAR")) - curve['10year'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_10YEAR")) - curve['20year'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_20YEAR")) - curve['30year'] = self.get_treasury_rate(self.get_node_value(entry, "d:BC_30YEAR")) - self.db.treasury_curves.insert(curve, True) - - def get_treasury_date(self, dstring): - return datetime.datetime.strptime(dstring.split("T")[0], '%Y-%m-%d') - - def get_treasury_rate(self, string_val): - val = self.guarded_conversion(float, string_val, None) - if val != None: - val = round(val / 100.0, 4) - return val - def get_node_value(self, entry_node, tag_name): - return self.get_xml_text(entry_node.getElementsByTagName(tag_name)[0].childNodes) - - def get_xml_text(self, nodelist): - rc = [] - for node in nodelist: - if node.nodeType == node.TEXT_NODE: - rc.append(node.data) - - return ''.join(rc) - - def purge_quotes(self): - self.db.equity.quotes.drop() - - def purge_trades(self): - self.db.equity.trades.drop() - - def load_quotes(self): - start = time.time() - qutil.LOGGER.info("processing equity quotes") - self.load_events(self.db.equity.quotes, - self.quoteRowCallback, - self.data_file_path + "2008/Quotes/DATA", - ['trade_date', 'trade_time','exchange_code','bid_price','ask_price', 'bid_size','ask_size']) - qutil.LOGGER.info("quotes complete") - total = time.time() - start - qutil.LOGGER.info("%d seconds to update equity quotes" % total) - - - def load_trades(self): - start = time.time() - qutil.LOGGER.info("processing equity minute bars") - self.load_events(self.db.equity.trades.minute, - self.trade_cb, - os.path.join(self.data_file_path, "2008/Trades/MINUTE_DATA"), - ['trade_date','trade_time','price', 'volume']) - qutil.LOGGER.info("minute trades complete") - total = time.time() - start - qutil.LOGGER.info("%d seconds to recreate equity trades" % total) - - def load_hourly_trades(self): - start = time.time() - qutil.LOGGER.info("processing equity hour bars") - self.load_events(self.db.equity.trades.hourly, - self.trade_cb, - os.path.join(self.data_file_path, "2008/Trades/HOURLY_DATA"), - ['trade_date','trade_time','price','volume']) - qutil.LOGGER.info("hourly trades complete") - total = time.time() - start - qutil.LOGGER.info("%d seconds to recreate equity trades" % total) - - - def load_daily_close(self): - start = time.time() - qutil.LOGGER.info("processing equity daily close") - self.load_events(self.db.equity.trades.daily, - self.trade_cb, - os.path.join(self.data_file_path, "2008/Trades/DAILY_DATA"), - ['trade_date','price', 'volume']) - qutil.LOGGER.info("daily close complete") - total = time.time() - start - qutil.LOGGER.info("%d seconds to recreate equity trades" % total) - - def ensure_indexes(self): - - #ensure indexes on minute trades - qutil.LOGGER.info("ensuring (+datetime, +sid) index on trades.minute") - self.db.equity.trades.minute.ensure_index([("dt",ASCENDING),("sid",ASCENDING)],background=True) - qutil.LOGGER.info("(+datetime, +sid) index on trades.minute ready") - - #ensure indexes for hourly trades - qutil.LOGGER.info("ensuring (sid, +datetime) index on trades.hourly") - self.db.equity.trades.hourly.ensure_index([("dt",ASCENDING),("sid",ASCENDING)],background=True) - qutil.LOGGER.info("(sid, +datetime) index on trades.hourly ready") - - #ensure indexes for daily trades - qutil.LOGGER.info("ensuring (+datetime,+sid) index on trades.daily") - self.db.equity.trades.daily.ensure_index([("dt",ASCENDING),("sid",ASCENDING)],background=True) - qutil.LOGGER.info("(+datetime,+sid) index on trades.daily ready") - - #ensure indexes for orders and transactions - qutil.LOGGER.info("ensuring (+backtestid) index on orders") - self.db.orders.ensure_index([("back_test_run_id",ASCENDING)],background=True) - qutil.LOGGER.info("(+backtestid) index on orders ready") - - qutil.LOGGER.info("ensuring (+backtestid, +datetime) index on orders") - self.db.orders.ensure_index([("back_test_run_id",ASCENDING),("dt",ASCENDING)],background=True) - qutil.LOGGER.info("(+backtestid, +datetime) index on orders ready") - - qutil.LOGGER.info("ensuring (+backtestid) index on orders") - self.db.transactions.ensure_index([("back_test_run_id",ASCENDING)],background=True) - qutil.LOGGER.info("(+backtestid) index on orders ready") - - qutil.LOGGER.info("ensuring (+backtestid) index on transactions") - self.db.transactions.ensure_index([("back_test_run_id",ASCENDING),("dt",ASCENDING)],background=True) - qutil.LOGGER.info("(+backtestid) index on transactions ready") - - #indexes for benchmarks and treasuries - qutil.LOGGER.info("ensuring (+date) index on treasury_curves") - self.db.treasury_curves.ensure_index([("date",ASCENDING)],background=True) - qutil.LOGGER.info(" (+date) index on treasury_curves ready") - - qutil.LOGGER.info("ensuring (-date) index on treasury_curves") - self.db.treasury_curves.ensure_index([("date",DESCENDING)],background=True) - qutil.LOGGER.info(" (-date) index on treasury_curves ready") - - qutil.LOGGER.info("ensuring (+date) index on bench_marks") - self.db.bench_marks.ensure_index([("date",ASCENDING)],background=True) - qutil.LOGGER.info(" (+date) index on bench_marks ready") - - qutil.LOGGER.info("ensuring (+symbol, +date) index on bench_marks") - self.db.bench_marks.ensure_index([("symbol",ASCENDING),("date",ASCENDING)],background=True) - qutil.LOGGER.info(" (+symbol, +date) index on bench_marks ready") - - def load_security_info(self): - start = time.time() - qutil.LOGGER.info("processing company info") - - sourceFile = os.path.join(self.data_file_path, "2008/Trades/MINUTE_DATA/CompanyInfo/CompanyInfo.asc") - self.db.securities.drop() - self.parse_file(self.db.securities, - self.security_cb, - sourceFile, - ['symbol','file name','company name','CUSIP','exchange','industry code','first date','last date','company id'], - None, - 0) - qutil.LOGGER.info("company info complete") - total = time.time() - start - qutil.LOGGER.info("%d seconds to recreate equity trades" % total) - - - - def load_events(self, collection, rowCallBack, dataDirectory, csvFields): - id_counter = 0 - listing = os.listdir(dataDirectory) - processedDir = os.path.join(dataDirectory,"processed") - if not os.path.exists(processedDir): - os.mkdir(processedDir) - for curFile in listing: - if os.path.isdir(os.path.join(dataDirectory,curFile)): - continue - start = time.time() - if id_counter == 0: #this is the first file we are processing, so we want to ensure we don't duplicate records - minDateTime = self.get_latest_entry_for_sid(self.get_sid_from_filename(curFile),collection) - else: - minDateTime = None #this isn't the first file, so don't bother querying - rowCount, totalCount = self.parse_file(collection, rowCallBack, os.path.join(dataDirectory,curFile), csvFields, minDateTime, id_counter) - id_counter = id_counter + rowCount - parseTime = time.time() - start - qutil.LOGGER.info("{time} seconds to parse and load {rowCount} records of {totalCount} from {file}. {rate} records/second". - format(time = parseTime, rowCount=rowCount, totalCount=totalCount, file=curFile, rate = rowCount/parseTime)) - #we successfully processed the file without an exception, move it to the processed folder - #qutil.LOGGER.info("moving data file to {newpath}".format(newpath=os.path.join(processedDir,curFile))) - shutil.move(os.path.join(dataDirectory,curFile),os.path.join(processedDir,curFile)) - - def parse_file(self, collection, rowCallBack, curFile, pFieldnames, minDateTime, id_counter): - """Parses the given file into the collection. Returns tuple of the rows committed, rows in csvfile""" - - qutil.LOGGER.debug("processing {fn}".format(fn=curFile)) - cur_id = id_counter - rowCount = 0 - csvRowCount = 0 - with open(curFile, 'rb') as f: - reader = csv.DictReader(f,fieldnames=pFieldnames) - header = False - - if csv.Sniffer().has_header(f.read(1024)): - header = True - f.seek(0) - - if header: - reader.next() - try: - rows = [] - for row in reader: - #row['_id'] = cur_id - cur_id = cur_id + 1 - csvRowCount += 1 - utcDT, dt = self.get_event_datetime(row) - #only add rows that are after the mindate for the current sid. - if(minDateTime != None and dt <= minDateTime): - continue - if(dt != None): - row['dt'] = dt - if('company id' not in pFieldnames): - company_id = self.get_sid_from_filename(curFile) - if(company_id): - row['sid'] = int(company_id) - if not rowCallBack(curFile, row): - continue - rows.append(row) - rowCount+=1 - if(len(rows) >= self.BATCH_SIZE): - collection.insert(rows, safe=True) - rows = [] - if(len(rows) > 0): - collection.insert(rows, safe=True) - rows = None - except csv.Error, e: - sys.exit('file %s, line %d: %s' % (curFile, reader.line_num, e)) - return rowCount, csvRowCount - - def trade_cb(self, curFile, row): - row['price'] = self.guarded_conversion(float,row['price']) - row['volume'] = self.guarded_conversion(self.safe_int,row['volume']) - return True - - def bench_mark_cb(self, curFile, row): - row['symbol'] = "GSPC" - row['volume'] = self.guarded_conversion(int,row['volume']) - row['open'] = self.guarded_conversion(float,row['open']) - row['high'] = self.guarded_conversion(float,row['high']) - row['low'] = self.guarded_conversion(float,row['low']) - row['close'] = self.guarded_conversion(float,row['close']) - row['adj_close'] = self.guarded_conversion(float,row['adj_close']) - row['date'] = datetime.datetime.strptime(row['date'], '%Y-%m-%d') - if self.last_bm_close == None: - row['returns'] = (row['close'] - row['open'])/row['open'] - else: - row['returns'] = (row['close'] - self.last_bm_close) / self.last_bm_close - self.last_bm_close = row['close'] - return True - - def security_cb(self, curFile, row): - """source columns: ['symbol','file name','company name','CUSIP','exchange','industry code','first date','last date','company id']""" - row['sid'] = self.guarded_conversion(int,row['company id']) - del(row['company id']) - row['start_date'] = self.guarded_conversion(self.date_conversion, row['first date']) - del(row['first date']) - row['end_date'] = self.guarded_conversion(self.date_conversion, row['last date']) - del(row['last date']) - row['symbol'] = self.verify_symbol_in_filename(row['symbol'], row['file name']) - del(row['file name']) - row['company_name'] = row['company name'] - del(row['company name']) - return True - - def guarded_conversion(self, conversion, strVal, default = None): - if(strVal == None or strVal == ""): - return default - return conversion(strVal) - - def safe_int(self,str): - """casts the string to a float to handle the occassionaly decimal point in int fields from data providers.""" - f = float(str) - i = int(f) - return i - - def date_conversion(self, dateStr): - dt = datetime.datetime.strptime(dateStr, '%m/%d/%Y') - dt = dt.replace (tzinfo = pytz.utc) - return dt - - def verify_symbol_in_filename(self, symbol, file_name): - if(symbol == file_name): - return symbol - - parts = file_name.split('_') - if(len(parts) == 2): - return file_name - else: - raise Exception("found a mismatch between symbol and filename, but no underscore.") - - def get_event_datetime(self, row): - """python 2.5 doesn't support %f for setting the microseconds, so this override is necessary. - a significant side effect - the trade date and trade time elements are removed from this dictionary. done to - avoid storing the source fields in the db. - """ - if row.has_key('trade_date') and row.has_key('trade_time'): - value = row['trade_date'] + "-" + row['trade_time'] - dt = datetime.datetime.strptime(value.split(".")[0], '%m/%d/%Y-%H:%M:%S') - dt = dt.replace(microsecond=int(value.split(".")[1]+"000")) - del row['trade_date'] - del row['trade_time'] - elif row.has_key('trade_date'): - dt = datetime.datetime.strptime(row['trade_date'],'%m/%d/%Y') - del row['trade_date'] - else: - return None, None - - utcDT = quantoenv.getUTCFromExchangeTime(dt) #store everything in UTC - return utcDT, dt - - def get_sid_from_filename(self, filename): - - regexp = r"(?P[0-9]+)([.]csv)" - result = re.search(regexp,filename) - if(result): - companyID = int(result.group('company_id')) - return companyID - else: - return None - - def get_latest_entry_for_sid(self, sid, collection): - """checks given collection for the most recent record for the given sid.""" - results = collection.find(fields=["dt"], - spec={"sid":sid}, - sort=[("dt",DESCENDING)], - limit=1, - as_class=quantoenv.DocWrap) - - if(results.count() > 0): - return results[0].dt - else: - return datetime.datetime.min - - - -class DataLoader(Daemon): - """A daemon process that manages the data in the finance database.""" - - def __init__(self, pidfile, operation): - self.operation = operation - self.pidfile = pidfile - self.stdin = '/dev/null' - self.stdout = '/dev/null' - self.stderr = '/dev/null' - - def run(self): - qutil.LOGGER.info("running operation: {op}".format(op=self.operation)) - try: - fdl = FinancialDataLoader() - if(self.operation == 'pt'): - qutil.LOGGER.info("Purging trades from database!") - fdl.purge_trades() - elif(self.operation == 'ei'): - qutil.LOGGER.info("Ensuring indexes.") - fdl.ensure_indexes() - elif(self.operation == 'lt'): - qutil.LOGGER.info("Loading trades into database.") - fdl.loadTrades() - elif(self.operation == 'lh'): - qutil.LOGGER.info("Loading trades into database.") - fdl.load_hourly_trades() - elif(self.operation == 'ld'): - qutil.LOGGER.info("Loading trades into database.") - fdl.load_daily_close() - elif(self.operation == 'si'): - qutil.LOGGER.info("Loading security info into database.") - fdl.load_security_info() - elif(self.operation == 'tr'): - qutil.LOGGER.info("Loading US Treasury rates into database.") - fdl.load_treasuries() - elif(self.operation == 'bm'): - qutil.LOGGER.info("loading benchmark data into database.") - fdl.load_bench_marks() - else: - qutil.LOGGER.warning("Unknown command for load data: {op}.".format(op=self.operation)) - qutil.LOGGER.info("Finished.") - except: - qutil.LOGGER.exception("exiting load_data due to unexpected exception.") - finally: - logging.shutdown() - - diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py new file mode 100644 index 00000000..8f3d9aa1 --- /dev/null +++ b/zipline/finance/performance.py @@ -0,0 +1,197 @@ +import datetime +import pytz +import math + +from zmq.core.poll import select + +import zipline.messaging as qmsg +import zipline.util as qutil +import zipline.protocol as zp +import zipline.finance.risk as risk + +class PortfolioClient(qmsg.Component): + + def __init__(self, period_start, period_end, capital_base, trading_environment): + qmsg.Component.__init__(self) + self.trading_day = datetime.timedelta(hours=6, minutes=30) + self.calendar_day = datetime.timedelta(hours=24) + self.period_start = period_start + self.period_end = period_end + self.market_open = self.period_start + self.market_close = self.market_open + self.trading_day + self.progress = 0.0 + self.total_days = (self.period_end - self.period_start).days + self.day_count = 0 + self.cumulative_capital_used= 0.0 + self.max_capital_used = 0.0 + self.capital_base = capital_base + self.trading_environment = trading_environment + self.returns = [] + self.cumulative_performance = PerformancePeriod(self.period_start, self.period_end, {}, 0, capital_base = capital_base) + self.todays_performance = PerformancePeriod(self.market_open, self.market_close, {}, 0, capital_base = capital_base) + + @property + def get_id(self): + return str(zp.FINANCE_COMPONENT.PORTFOLIO_CLIENT) + + def open(self): + self.result_feed = self.connect_result() + + def do_work(self): + #next feed event + socks = dict(self.poll.poll(self.heartbeat_timeout)) + + if self.result_feed in socks and socks[self.result_feed] == self.zmq.POLLIN: + msg = self.result_feed.recv() + + if msg == str(zp.CONTROL_PROTOCOL.DONE): + self.handle_simulation_end() + qutil.LOGGER.info("Portfolio Client is DONE!") + self.signal_done() + return + + event = zp.MERGE_UNFRAME(msg) + + if(event.dt >= self.market_close): + self.handle_market_close() + + if event.TRANSACTION: + self.cumulative_performance.execute_transaction(event.TRANSACTION) + self.todays_performance.execute_transaction(event.TRANSACTION) + + #we're adding a 10% cushion to the capital used, and then rounding to the nearest 5k + self.cumulative_capital_used += event.TRANSACTION.price * event.TRANSACTION.amount + if(math.fabs(self.cumulative_capital_used) > self.max_capital_used): + self.max_capital_used = math.fabs(self.cumulative_capital_used) + self.max_capital_used = self.round_to_nearest(1.1 * self.max_capital_used, base=5000) + self.max_leverage = self.max_capital_used/self.capital_base + + #update last sale + self.cumulative_performance.update_last_sale(event) + self.todays_performance.update_last_sale(event) + + #calculate performance as of last trade + self.cumulative_performance.calculate_performance() + self.todays_performance.calculate_performance() + + + + def handle_market_close(self): + self.market_open = self.market_open + self.calendar_day + while not self.trading_environment.is_trading_day(self.market_open): + if self.market_open > self.trading_environment.trading_days[-1]: + raise Exception("Attempting to backtest beyond available history.") + self.market_open = self.market_open + self.calendar_day + self.market_close = self.market_open + self.trading_day + self.day_count += 1.0 + self.progress = self.day_count / self.total_days + #add the return results from today to the list of daily return objects. + todays_date = self.todays_performance.period_end.replace(hour=0, minute=0, second=0) + todays_return_obj = risk.daily_return(todays_date, self.todays_performance.returns) + self.returns.append(todays_return_obj) + + #calculate risk metrics for cumulative performance + self.cur_period_metrics = risk.RiskMetrics(start_date=self.cumulative_performance.period_start, + end_date=self.cumulative_performance.period_end.replace(hour=0, minute=0, second=0), + returns=self.returns, + trading_environment=self.trading_environment) + + ###################################################################################################### + #######TODO: report/relay metrics out to qexec -- values come from self.cur_period_metrics ########### + #######TODO: report/relay position data out to qexec -- values come from self.cumulative_performance # + ###################################################################################################### + + #roll over positions to current day. + self.todays_performance = PerformancePeriod(self.market_open, + self.market_close, + self.todays_performance.positions, + self.todays_performance.ending_value, + self.capital_base) + + def handle_simulation_end(self): + self.risk_report = risk.RiskReport(self.returns, self.trading_environment) + ###################################################################################################### + #######TODO: report/relay metrics out to qexec -- values come from self.risk_report ########### + ###################################################################################################### + + def round_to_nearest(self, x, base=5): + return int(base * round(float(x)/base)) + +class Position(): + sid = None + amount = None + cost_basis = None + last_sale = None + last_date = None + + def __init__(self, sid): + self.sid = sid + self.amount = 0 + self.cost_basis = 0.0 ##per share + + def update(self, txn): + if(self.sid != txn.sid): + raise NameError('attempt to update position with transaction in different sid') + #throw exception + + if(self.amount + txn.amount == 0): #we're covering a short or closing a position + self.cost_basis = 0.0 + self.amount = 0 + else: + self.cost_basis = (self.cost_basis*self.amount + (txn.amount*txn.price))/(self.amount + txn.amount) + self.amount = self.amount + txn.amount + + def currentValue(self): + return self.amount * self.last_sale + + + def __repr__(self): + return "sid: {sid}, amount: {amount}, cost_basis: {cost_basis}, last_sale: {last_sale}".format( + sid=self.sid, amount=self.amount, cost_basis=self.cost_basis, last_sale=self.last_sale) + +class PerformancePeriod(): + + def __init__(self, period_start, period_end, initial_positions, initial_value, capital_base = None): + self.ending_value = 0.0 + self.period_capital_used = 0.0 + self.period_start = period_start + self.period_end = period_end + self.positions = initial_positions #sid => position object + self.starting_value = initial_value + if(capital_base != None): + self.capital_base = capital_base + else: + self.capital_base = 0 + + def calculate_performance(self): + self.ending_value = self.calculate_positions_value() + self.pnl = (self.ending_value - self.starting_value) - self.period_capital_used + if(self.capital_base != 0): + self.returns = self.pnl / self.starting_value + else: + self.returns = 0.0 + + def execute_transaction(self, txn): + if(txn.dt > self.period_end): + raise Exception("transaction dated {dt} attempted for period ending {ending}". + format(dt=txn.dt, ending=self.period_end)) + if(not self.positions.has_key(txn.sid)): + self.positions[txn.sid] = Position(txn.sid) + self.positions[txn.sid].update(txn) + self.period_capital_used += -1 * txn.price * txn.amount + + + def calculate_positions_value(self): + mktValue = 0.0 + for key,pos in self.positions.iteritems(): + mktValue += pos.currentValue() + return mktValue + + def update_last_sale(self, event): + if self.positions.has_key(event.sid): + self.positions[event.sid].last_sale = event.price + self.positions[event.sid].last_date = event.dt + + + + \ No newline at end of file diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index a13e98bb..2d6195c0 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -4,7 +4,6 @@ import pytz import numpy as np import numpy.linalg as la import zipline.util as qutil -import zipline.db as db import zipline.protocol as zp from pymongo import ASCENDING, DESCENDING @@ -13,10 +12,17 @@ class daily_return(): def __init__(self, date, returns): self.date = date self.returns = returns + + def __repr__(self): + return str(self.date) + " - " + str(self.returns) -class periodmetrics(): - def __init__(self, start_date, end_date, returns, benchmark_returns): - self.db = db.DbConnection.get()[1] +class RiskMetrics(): + def __init__(self, start_date, end_date, returns, benchmark_returns, treasury_curves, trading_calendar): + """ + :param treasury_curves: {datetime in utc -> {duration label -> interest rate}} + """ + + self.treasury_curves = treasury_curves self.start_date = start_date self.end_date = end_date self.trading_calendar = trading_calendar @@ -134,13 +140,18 @@ class periodmetrics(): else: self.treasury_duration = '30year' - treasuryQS = self.db.treasury_curves.find( - spec={"date" : {"$lte" : self.end_date}}, - sort=[("date",DESCENDING)], - limit=3, - slave_ok=True) - - for curve in treasuryQS: + + one_day = datetime.timedelta(days=1) + + curve = None + #in case end date is not a trading day, search for the next market day for an interest rate + for i in range(7): + if(self.treasury_curves.has_key(self.end_date + i * one_day)): + #qutil.LOGGER.info(self.treasury_curves[self.end_date + i * one_day]) + curve = self.treasury_curves[self.end_date + i * one_day] + break + + if curve: self.treasury_curve = curve rate = self.treasury_curve[self.treasury_duration] #1month note data begins in 8/2001, so we can use 3month instead. @@ -149,17 +160,18 @@ class periodmetrics(): if rate != None: return rate * (td.days + 1) / 365 - raise Exception("no rate for end date = {dt} and term = {term}, from {curve}. Using zero.".format(dt=self.end_date, - term=self.treasury_duration, - curve=self.treasury_curve['date'])) + raise Exception("no rate for end date = {dt} and term = {term}. Using zero.".format(dt=self.end_date, + term=self.treasury_duration)) -class riskmetrics(): +class RiskReport(): - def __init__(self, algorithm_returns): + def __init__(self, algorithm_returns, benchmark_returns, treasury_curves, trading_calendar): """algorithm_returns needs to be a list of daily_return objects sorted in date ascending order""" - self.db = db.DbConnection.get()[1] + self.algorithm_returns = algorithm_returns self.bm_returns = [x for x in benchmark_returns if x.date >= self.algorithm_returns[0].date and x.date <= self.algorithm_returns[-1].date] + self.treasury_curves = treasury_curves + self.trading_calendar = trading_calendar qutil.LOGGER.debug("#### {start} thru {end} with {count} trading_days of {total} possible".format(start=self.algorithm_returns[0].date, end=self.algorithm_returns[-1].date, @@ -191,23 +203,16 @@ class riskmetrics(): if(cur_end > the_end): break #qutil.LOGGER.debug("start: {start}, end: {end}".format(start=cur_start, end=cur_end)) - cur_period_metrics = periodmetrics(start_date=cur_start, end_date=cur_end, returns=self.algorithm_returns, benchmark_returns=self.bm_returns) + cur_period_metrics = RiskMetrics(start_date=cur_start, + end_date=cur_end, + returns=self.algorithm_returns, + benchmark_returns=self.bm_returns, + treasury_curves=self.treasury_curves, + trading_calendar=self.trading_calendar) ends.append(cur_period_metrics) cur_start = advance_by_months(cur_start, 1) return ends - - def store_to_db(self, back_test_run_id): - col = self.db.risk_metrics - for period in self.month_periods: - for metric in ["algorithm_period_returns", "benchmark_period_returns", "excess_return", "trading_days", "benchmark_volatility", "algorithm_volatility", "sharpe", "beta", "alpha", "max_drawdown"]: - record = {'back_test_run_id':back_test_run_id} - record['ending_on'] = period.end_date - record['metric_name'] = metric - for dur in ["month", "three_month", "six_month", "year", "three_year", "five_year"]: - record[dur] = self.find_metric_by_end(period.end_date, dur, metric) - #qutil.LOGGER.debug("storing {val} for {metric} and {dur}".format(val=record[dur], metric=metric, dur=dur)) - col.insert(record, safe=True) def find_metric_by_end(self, end_date, duration, metric): col = getattr(self, duration + "_periods") @@ -231,11 +236,12 @@ def advance_by_months(dt, jump_in_months): r = dt.replace(year = dt.year + years, month = month) return r -class TradingCalendar(object): +class TradingEnvironment(object): - def __init__(self, benchmark_returns): + def __init__(self, benchmark_returns, treasury_curves): self.trading_days = [] self.trading_day_map = {} + self.treasury_curves = treasury_curves for bm in benchmark_returns: self.trading_days.append(bm.date) self.trading_day_map[bm.date] = bm @@ -253,21 +259,4 @@ class TradingCalendar(object): return self.trading_day_map[date].returns else: return 0.0 - - -def get_benchmark_data(): - bmQS = db.DbConnection.get()[1].bench_marks.find( - spec={"symbol" : "GSPC"}, - sort=[("date",ASCENDING)], - slave_ok=True, - as_class=zp.namedict) - bm_returns = [] - for bm in bmQS: - bm_r = daily_return(date=bm.date.replace(tzinfo=pytz.utc), returns=bm.returns) - bm_returns.append(bm_r) - - cal = TradingCalendar(bm_returns) - return bm_returns, cal - -benchmark_returns, trading_calendar = get_benchmark_data() diff --git a/zipline/test/factory.py b/zipline/test/factory.py index 2194aed1..68ed322b 100644 --- a/zipline/test/factory.py +++ b/zipline/test/factory.py @@ -1,9 +1,26 @@ import datetime import pytz +import msgpack +import random import zipline.util as qutil import zipline.finance.risk as risk import zipline.protocol as zp +def load_market_data(): + fp_bm = open("./zipline/test/benchmark.msgpack", "rb") + bm_map = msgpack.loads(fp_bm.read()) + bm_returns = [] + for epoch, returns in bm_map.iteritems(): + bm_returns.append(risk.daily_return(date=datetime.datetime.fromtimestamp(epoch).replace(hour=0, minute=0, second=0, tzinfo=pytz.utc), returns=returns)) + bm_returns = sorted(bm_returns, key=lambda(x): x.date) + fp_tr = open("./zipline/test/treasury_curves.msgpack", "rb") + tr_map = msgpack.loads(fp_tr.read()) + tr_curves = {} + for epoch, curve in tr_map.iteritems(): + tr_curves[datetime.datetime.fromtimestamp(epoch).replace(hour=0, minute=0, second=0, tzinfo=pytz.utc)] = curve + + return bm_returns, tr_curves + def create_trade(sid, price, amount, datetime): row = { @@ -16,14 +33,14 @@ def create_trade(sid, price, amount, datetime): } return row -def create_trade_history(sid, prices, amounts, start_time, interval): +def create_trade_history(sid, prices, amounts, start_time, interval, trading_calendar): i = 0 trades = [] current = start_time.replace(tzinfo = pytz.utc) for price, amount in zip(prices, amounts): - if(risk.trading_calendar.is_trading_day(current)): + if(trading_calendar.is_trading_day(current)): trade = create_trade(sid, price, amount, current) trades.append(trade) @@ -38,13 +55,13 @@ def createTxn(sid, price, amount, datetime, btrid=None): price=price, transaction_cost=-1*price*amount) return txn -def createTxnHistory(sid, priceList, amtList, startTime, interval): +def create_transaction_history(sid, priceList, amtList, startTime, interval, trading_calendar): txns = [] current = startTime for price, amount in zip(priceList, amtList): - if risk.trading_calendar.is_trading_day(current): + if trading_calendar.is_trading_day(current): txns.append(createTxn(sid, price, amount, current)) current = current + interval @@ -52,3 +69,47 @@ def createTxnHistory(sid, priceList, amtList, startTime, interval): current = current + datetime.timedelta(days=1) return txns + + +def create_returns(daycount, start, trading_calendar): + i = 0 + test_range = [] + current = start.replace(tzinfo=pytz.utc) + one_day = datetime.timedelta(days = 1) + while i < daycount: + i += 1 + r = risk.daily_return(current, random.random()) + test_range.append(r) + current = current + one_day + return [ x for x in test_range if(trading_calendar.is_trading_day(x.date)) ] + + +def create_returns_from_range(start, end, trading_calendar): + current = start.replace(tzinfo=pytz.utc) + end = end.replace(tzinfo=pytz.utc) + one_day = datetime.timedelta(days = 1) + test_range = [] + i = 0 + while current <= end: + current = current + one_day + if(not trading_calendar.is_trading_day(current)): + continue + r = risk.daily_return(current, random.random()) + i += 1 + test_range.append(r) + + return test_range + +def create_returns_from_list(returns, start, trading_calendar): + current = start.replace(tzinfo=pytz.utc) + one_day = datetime.timedelta(days = 1) + test_range = [] + i = 0 + while len(test_range) < len(returns): + if(trading_calendar.is_trading_day(current)): + r = risk.daily_return(current, returns[i]) + i += 1 + test_range.append(r) + current = current + one_day + return sorted(test_range, key=lambda(x):x.date) + diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 0544f5c6..829f5fa2 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -1,15 +1,14 @@ """Tests for the zipline.finance package""" import mock import pytz -import host_settings from unittest2 import TestCase from datetime import datetime, timedelta import zipline.test.factory as factory import zipline.util as qutil -import zipline.db as db import zipline.finance.risk as risk import zipline.protocol as zp +import zipline.finance.performance as perf from zipline.test.client import TestTradingClient from zipline.sources import SpecificEquityTrades @@ -22,6 +21,14 @@ class FinanceTestCase(TestCase): def setUp(self): qutil.configure_logging() + self.benchmark_returns, self.treasury_curves = \ + factory.load_market_data() + + self.trading_environment = risk.TradingEnvironment( + self.benchmark_returns, + self.treasury_curves + ) + def test_trade_feed_protocol(self): @@ -33,7 +40,14 @@ class FinanceTestCase(TestCase): start_date = datetime.strptime("02/15/2012","%m/%d/%Y") one_day_td = timedelta(days=1) - trades = factory.create_trade_history( sid, price, volume, start_date, one_day_td ) + trades = factory.create_trade_history( + sid, + price, + volume, + start_date, + one_day_td, + self.trading_environment + ) for trade in trades: #simulate data source sending frame @@ -112,14 +126,6 @@ class FinanceTestCase(TestCase): self.assertEqual(recovered_tx.sid, 133) self.assertEqual(recovered_tx.amount, 100) - def test_trading_calendar(self): - known_trading_day = datetime.strptime("02/24/2012","%m/%d/%Y") - known_holiday = datetime.strptime("02/20/2012", "%m/%d/%Y") #president's day - saturday = datetime.strptime("02/25/2012", "%m/%d/%Y") - self.assertTrue(risk.trading_calendar.is_trading_day(known_trading_day)) - self.assertFalse(risk.trading_calendar.is_trading_day(known_holiday)) - self.assertFalse(risk.trading_calendar.is_trading_day(saturday)) - def test_orders(self): # Just verify sending and receiving orders. @@ -156,7 +162,14 @@ class FinanceTestCase(TestCase): start_date = datetime.strptime("02/1/2012","%m/%d/%Y") trade_time_increment = timedelta(days=1) - trade_history = factory.create_trade_history( sid, price, volume, start_date, trade_time_increment ) + trade_history = factory.create_trade_history( + sid, + price, + volume, + start_date, + trade_time_increment, + self.trading_environment + ) set1 = SpecificEquityTrades("flat-133", trade_history) @@ -180,3 +193,83 @@ class FinanceTestCase(TestCase): self.assertEqual(sim.feed.pending_messages(), 0, \ "The feed should be drained of all messages, found {n} remaining." \ .format(n=sim.feed.pending_messages())) + + + def test_performance(self): + + # verify order -> transaction -> portfolio position. + # -------------- + + # Allocate sockets for the simulator components + allocator = AddressAllocator(8) + sockets = allocator.lease(8) + + addresses = { + 'sync_address' : sockets[0], + 'data_address' : sockets[1], + 'feed_address' : sockets[2], + 'merge_address' : sockets[3], + 'result_address' : sockets[4], + 'order_address' : sockets[5] + } + + con = Controller( + sockets[6], + sockets[7], + logging = qutil.LOGGER + ) + + sim = Simulator(addresses) + + # Simulation Components + # --------------------- + + # TODO: Perhaps something more self-documenting for variables names? + sid = 133 + price = [10.1] * 16 + volume = [100] * 16 + start_date = datetime.strptime("02/1/2012","%m/%d/%Y") + trade_time_increment = timedelta(days=1) + + trade_history = factory.create_trade_history( + sid, + price, + volume, + start_date, + trade_time_increment, + self.trading_environment ) + + set1 = SpecificEquityTrades("flat-133", trade_history) + + #client sill send 10 orders for 100 shares of 133 + client = TestTradingClient(133, 100, 10) + ts = datetime.strptime("02/1/2012","%m/%d/%Y") + ts = ts.replace(tzinfo=pytz.utc) + + order_source = OrderDataSource(ts) + transaction_sim = TransactionSimulator() + portfolio_client = perf.PortfolioClient( + trade_history[0]['dt'], + trade_history[-1]['dt'], + 1000000.0, + self.trading_environment) + + sim.register_components([ + client, + order_source, + transaction_sim, + set1, + portfolio_client, + ]) + sim.register_controller( con ) + + # Simulation + # ---------- + sim_context = sim.simulate() + sim_context.join() + + + # TODO: Make more assertions about the final state of the components. + self.assertEqual(sim.feed.pending_messages(), 0, \ + "The feed should be drained of all messages, found {n} remaining." \ + .format(n=sim.feed.pending_messages())) \ No newline at end of file diff --git a/zipline/test/test_risk.py b/zipline/test/test_risk.py new file mode 100644 index 00000000..81aa0de9 --- /dev/null +++ b/zipline/test/test_risk.py @@ -0,0 +1,282 @@ +import unittest +import copy +import datetime +import calendar +import pytz +import zipline.finance.risk as risk +import zipline.test.factory as factory +import zipline.util as qutil + +class Risk(unittest.TestCase): + + def setUp(self): + qutil.configure_logging() + self.benchmark_returns, self.treasury_curves = \ + factory.load_market_data() + + self.trading_calendar = risk.TradingEnvironment( + self.benchmark_returns, + self.treasury_curves + ) + + self.onesec = datetime.timedelta(seconds=1) + self.oneday = datetime.timedelta(days=1) + self.tradingday = datetime.timedelta(hours=6, minutes=30) + self.dt = datetime.datetime.utcnow() + start_date = datetime.datetime(year=2006, month=1, day=1, tzinfo=pytz.utc) + self.algo_returns_06 = factory.create_returns_from_list(RETURNS, start_date, self.trading_calendar) + end_date = datetime.datetime(year=2006, month=12, day=31, tzinfo=pytz.utc) + self.metrics_06 = risk.RiskReport(self.algo_returns_06, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + + def tearDown(self): + return + + def test_factory(self): + returns = [0.1] * 100 + start_date = datetime.datetime(year=2006, month=1, day=1, tzinfo=pytz.utc) + r_objects = factory.create_returns_from_list(returns, start_date, self.trading_calendar) + self.assertTrue(r_objects[-1].date <= datetime.datetime(year=2006, month=12, day=31, tzinfo=pytz.utc)) + + def test_drawdown(self): + start_date = datetime.datetime(year=2006, month=1, day=1) + returns = factory.create_returns_from_list([1.0,-0.5,0.8,.17,1.0,-0.1,-0.45], start_date, self.trading_calendar) + #200, 100, 180, 210.6, 421.2, 379.8, 208.494 + metrics = risk.RiskMetrics(returns[0].date, returns[-1].date, returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + self.assertEqual(metrics.max_drawdown, 0.505) + + def test_benchmark_returns_06(self): + start_date = datetime.datetime(year=2006, month=1, day=1) + end_date = datetime.datetime(year=2006, month=12, day=31) + returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar) + metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + self.assertEqual([round(x.benchmark_period_returns, 4) for x in metrics.month_periods], + [0.0255,0.0005,0.0111,0.0122,-0.0309,0.0001,0.0051,0.0213,0.0246,0.0315,0.0165,0.0126]) + self.assertEqual([round(x.benchmark_period_returns, 4) for x in metrics.three_month_periods], + [0.0373,0.0239,-0.0083,-0.0191,-0.0259,0.0266,0.0517,0.0793,0.0743,0.0617]) + self.assertEqual([round(x.benchmark_period_returns, 4) for x in metrics.six_month_periods], + [0.0176,-0.0027,0.0181,0.0316,0.0514,0.1028,0.1166]) + self.assertEqual([round(x.benchmark_period_returns,4) for x in metrics.year_periods],[0.1362]) + + def test_trading_days_06(self): + start_date = datetime.datetime(year=2006, month=1, day=1) + end_date = datetime.datetime(year=2006, month=12, day=31) + returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar) + metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + self.assertEqual([x.trading_days for x in metrics.year_periods],[251]) + self.assertEqual([x.trading_days for x in metrics.month_periods],[20,19,23,19,22,22,20,23,20,22,21,20]) + + def test_benchmark_volatility_06(self): + start_date = datetime.datetime(year=2006, month=1, day=1) + end_date = datetime.datetime(year=2006, month=12, day=31) + returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar) + metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.month_periods], + [0.031,0.026,0.024,0.025,0.037,0.047,0.039,0.022,0.023,0.021,0.025,0.019]) + self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.three_month_periods], + [0.047,0.042,0.050,0.064,0.070,0.064,0.049,0.037,0.039,0.037]) + self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.six_month_periods], + [0.079,0.082,0.081,0.081,0.08,0.074,0.061]) + self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.year_periods],[0.100]) + + def test_algorithm_returns_06(self): + self.assertEqual([round(x.algorithm_period_returns, 3) for x in self.metrics_06.month_periods],[0.101,-0.062,-0.041,0.092,0.135,-0.25,0.076,-0.003,-0.024,0.072,0.063,-0.071]) + self.assertEqual([round(x.algorithm_period_returns, 3) for x in self.metrics_06.three_month_periods],[-0.009,-0.017,0.188,-0.071,-0.085,-0.196,0.047,0.043,0.112,0.058]) + self.assertEqual([round(x.algorithm_period_returns, 3) for x in self.metrics_06.six_month_periods],[-0.08,-0.101,-0.044,-0.027,-0.045,-0.106,0.108]) + self.assertEqual([round(x.algorithm_period_returns, 3) for x in self.metrics_06.year_periods],[0.02]) + + def test_algorithm_volatility_06(self): + self.assertEqual([round(x.algorithm_volatility, 3) for x in self.metrics_06.month_periods],[0.137,0.12,0.13,0.142,0.128,0.14,0.141,0.118,0.143,0.144,0.117,0.135]) + self.assertEqual([round(x.algorithm_volatility, 3) for x in self.metrics_06.three_month_periods],[0.222,0.224,0.229,0.243,0.243,0.235,0.23,0.231,0.231,0.227]) + self.assertEqual([round(x.algorithm_volatility, 3) for x in self.metrics_06.six_month_periods],[0.328,0.329,0.329,0.333,0.334,0.329,0.321]) + self.assertEqual([round(x.algorithm_volatility, 3) for x in self.metrics_06.year_periods],[0.458]) + + def test_algorithm_sharpe_06(self): + self.assertEqual([round(x.sharpe, 3) for x in self.metrics_06.month_periods],[0.711,-0.541,-0.348,0.625,1.017,-1.809,0.508,-0.062,-0.193,0.467,0.502,-0.557]) + self.assertEqual([round(x.sharpe, 3) for x in self.metrics_06.three_month_periods],[-0.094,-0.129,0.769,-0.342,-0.402,-0.888,0.153,0.131,0.432,0.2]) + self.assertEqual([round(x.sharpe, 3) for x in self.metrics_06.six_month_periods],[-0.322,-0.383,-0.213,-0.156,-0.213,-0.398,0.257]) + self.assertEqual([round(x.sharpe, 3) for x in self.metrics_06.year_periods],[-0.066]) + + def dtest_algorithm_beta_06(self): + self.assertEqual([round(x.beta, 3) for x in self.metrics_06.month_periods],[0.553,0.583,-2.168,-0.548,1.463,-0.322,-1.38,1.473,-1.315,-0.7,0.352,-2.002]) + self.assertEqual([round(x.beta, 3) for x in self.metrics_06.three_month_periods],[-0.075,-0.637,0.124,0.186,-0.204,-0.497,-0.867,-0.173,-0.499,-0.563]) + self.assertEqual([round(x.beta, 3) for x in self.metrics_06.six_month_periods],[-0.075,-0.637,0.124,0.186,-0.204,-0.497,-0.867,-0.173,-0.499,-0.563]) + self.assertEqual([round(x.beta, 3) for x in self.metrics_06.year_periods],[-0.219]) + + def dtest_algorithm_alpha_06(self): + self.assertEqual([round(x.alpha, 3) for x in self.metrics_06.month_periods],[0.085,-0.063,-0.03,0.093,0.182,-0.255,0.073,-0.032,0,0.086,0.054,-0.058]) + self.assertEqual([round(x.alpha, 3) for x in self.metrics_06.three_month_periods],[-0.051,-0.021,0.179,-0.077,-0.106,-0.202,0.069,0.042,0.13,0.073]) + self.assertEqual([round(x.alpha, 3) for x in self.metrics_06.six_month_periods],[-0.105,-0.135,-0.072,-0.051,-0.066,-0.094,0.152]) + self.assertEqual([round(x.alpha, 3) for x in self.metrics_06.year_periods],[-0.011]) + + #FIXME: Covariance is not matching excel precisely enough to run the test. Month 4 seems to be the problem. Variance is disabled + #just to avoid distraction - it is much closer than covariance and can probably pass with 6 significant digits instead of 7. + #re-enable variance, alpha, and beta tests once this is resolved + def dtest_algorithm_covariance_06(self): + metric = self.metrics_06.month_periods[3] + print repr(metric) + print "----" + self.assertEqual([round(x.algorithm_covariance, 7) for x in self.metrics_06.month_periods],[0.0000289,0.0000222,-0.0000554,-0.0000192,0.0000954,-0.0000333,-0.0001111,0.0000322,-0.0000349,-0.0000143,0.0000108,-0.0000386]) + self.assertEqual([round(x.algorithm_covariance, 7) for x in self.metrics_06.three_month_periods],[-0.0000026,-0.0000189,0.0000049,0.0000121,-0.0000158,-0.000031,-0.0000336,-0.0000036,-0.0000119,-0.0000122]) + self.assertEqual([round(x.algorithm_covariance, 7) for x in self.metrics_06.six_month_periods],[0.000005,-0.0000172,-0.0000142,-0.0000102,-0.0000089,-0.0000207,-0.0000229]) + self.assertEqual([round(x.algorithm_covariance, 7) for x in self.metrics_06.year_periods],[-8.75273E-06]) + + def dtest_benchmark_variance_06(self): + self.assertEqual([round(x.benchmark_variance, 7) for x in self.metrics_06.month_periods],[0.0000496,0.000036,0.0000244,0.0000332,0.0000623,0.0000989,0.0000765,0.0000209,0.0000252,0.0000194,0.0000292,0.0000183]) + self.assertEqual([round(x.benchmark_variance, 7) for x in self.metrics_06.three_month_periods],[0.0000351,0.0000298,0.0000395,0.0000648,0.0000773,0.0000625,0.0000387,0.0000211,0.0000238,0.0000217]) + self.assertEqual([round(x.benchmark_variance, 7) for x in self.metrics_06.six_month_periods],[0.0000499,0.0000538,0.0000508,0.0000517,0.0000492,0.0000432,0.00003]) + self.assertEqual([round(x.benchmark_variance, 7) for x in self.metrics_06.year_periods],[0.0000399]) + + + def test_benchmark_returns_08(self): + start_date = datetime.datetime(year=2008, month=1, day=1) + end_date = datetime.datetime(year=2008, month=12, day=31) + returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar) + metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + self.assertEqual([round(x.benchmark_period_returns, 3) for x in metrics.month_periods], + [-0.061,-0.035,-0.006,0.048,0.011,-0.086,-0.01,0.012,-0.091,-0.169,-0.075,0.008]) + self.assertEqual([round(x.benchmark_period_returns, 3) for x in metrics.three_month_periods], + [-0.099,0.005,0.052,-0.032,-0.085,-0.084,-0.089,-0.236,-0.301,-0.226]) + self.assertEqual([round(x.benchmark_period_returns, 3) for x in metrics.six_month_periods], + [-0.128,-0.081,-0.036,-0.118,-0.301,-0.360,-0.294]) + self.assertEqual([round(x.benchmark_period_returns,3) for x in metrics.year_periods],[-0.385]) + + def test_trading_days_08(self): + start_date = datetime.datetime(year=2008, month=1, day=1) + end_date = datetime.datetime(year=2008, month=12, day=31) + returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar) + metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + self.assertEqual([x.trading_days for x in metrics.year_periods],[253]) + self.assertEqual([x.trading_days for x in metrics.month_periods],[21,20,20,22,21,21,22,21,21,23,19,22]) + + def test_benchmark_volatility_08(self): + start_date = datetime.datetime(year=2008, month=1, day=1) + end_date = datetime.datetime(year=2008, month=12, day=31) + returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar) + metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.month_periods], + [0.07,0.058,0.082,0.054,0.041,0.057,0.068,0.06,0.157,0.244,0.195,0.145]) + self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.three_month_periods], + [0.120,0.113,0.105,0.09,0.098,0.107,0.179,0.293,0.344,0.340]) + self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.six_month_periods], + [0.15,0.149,0.15,0.2,0.308,0.36,0.383]) + #TODO: ugly, but I can't get the rounded float to match. maybe we need a different test that checks the difference between the numbers + self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.year_periods],[0.41099999999999998]) + + def test_treasury_returns_06(self): + start_date = datetime.datetime(year=2006, month=1, day=1) + end_date = datetime.datetime(year=2006, month=12, day=31) + returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar) + metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.month_periods], + [0.0037,0.0034,0.0039,0.0038,0.0040,0.0037,0.0043,0.0043,0.0038,0.0044,0.0043,0.0041]) + self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.three_month_periods], + [0.0114,0.0118,0.0122,0.0125,0.0129,0.0127,0.0123,0.0128,0.0125,0.0128]) + self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.six_month_periods], + [0.0260,0.0257,0.0258,0.0252,0.0259,0.0256,0.0258]) + self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.year_periods], + [0.0500]) + + def test_benchmarkrange(self): + self.check_year_range(datetime.datetime(year=2008,month=1,day=1), 2) + + def test_partial_month(self): + start_date = datetime.datetime(year=1991, month=1, day=1) + returns = factory.create_returns(365 * 5 + 2, start_date, self.trading_calendar) #1992 and 1996 were leap years + returns = returns[:-10] #truncate the returns series to end mid-month + metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + total_months = 60 + self.check_metrics(metrics, total_months, start_date) + + def check_year_range(self, start_date, years): + if(start_date.month <= 2): + ld = calendar.leapdays(start_date.year, start_date.year + years) + else: + #because we may catch the leap of the last year, and i think this func is [start,end) + ld = calendar.leapdays(start_date.year, start_date.year + years + 1) + returns = factory.create_returns(365 * years + ld, start_date, self.trading_calendar) + metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar) + total_months = years * 12 + self.check_metrics(metrics, total_months, start_date) + + def check_metrics(self, metrics, total_months, start_date): + self.assert_range_length(metrics.month_periods, total_months, 1, start_date) + self.assert_range_length(metrics.three_month_periods, total_months, 3, start_date) + self.assert_range_length(metrics.six_month_periods, total_months, 6, start_date) + self.assert_range_length(metrics.year_periods, total_months, 12, start_date) + self.assert_range_length(metrics.three_year_periods, total_months, 36, start_date) + self.assert_range_length(metrics.five_year_periods, total_months, 60, start_date) + + def assert_last_day(self, period_end): + #30 days has september, april, june and november + if(period_end.month in [9,4,6,11]): + self.assertEqual(period_end.day, 30) + #all the rest have 31, except for february + elif(period_end.month != 2): + self.assertEqual(period_end.day, 31) + else: + if calendar.isleap(period_end.year): + self.assertEqual(period_end.day, 29) + else: + self.assertEqual(period_end.day, 28) + + def assert_month(self, start_month, actual_end_month): + if start_month == 1: + expected_end_month = 12 + else: + expected_end_month = start_month - 1 + + self.assertEqual(expected_end_month, actual_end_month) + + def assert_range_length(self, col, total_months, period_length, start_date): + if(period_length > total_months): + self.assertEqual(len(col), 0) + else: + self.assertEqual(len(col), total_months - (period_length - 1), "mismatch for total months - expected:{total_months}/actual:{actual}, period:{period_length}, start:{start_date}, calculated end:{end}".format( + total_months=total_months, + period_length=period_length, + start_date=start_date, + end=col[-1].end_date, + actual=len(col) + )) + self.assert_month(start_date.month, col[-1].end_date.month) + self.assert_last_day(col[-1].end_date) + + +RETURNS = [ + 0.0093, -0.0193, 0.0351, 0.0396, 0.0338, -0.0211, 0.0389, + 0.0326, -0.0137, -0.0411, -0.0032, 0.0149, 0.0133, 0.0348, + 0.042 , -0.0455, 0.0262, -0.0461, 0.0021, -0.0273, -0.0429, + 0.0427, -0.0104, 0.0346, -0.0311, 0.0003, 0.0211, 0.0248, + -0.0215, 0.004 , 0.0267, 0.0029, -0.0369, 0.0057, 0.0298, + -0.0179, -0.0361, -0.0401, -0.0123, -0.005 , 0.0203, -0.041 , + 0.0011, 0.0118, 0.0103, -0.0184, -0.0437, 0.0411, -0.0242, + -0.0054, -0.0039, -0.0273, -0.0075, 0.0064, -0.0376, 0.0424, + 0.0399, 0.019 , 0.0236, -0.0284, -0.0341, 0.0266, 0.05 , + 0.0069, -0.0442, -0.016 , 0.0173, 0.0348, -0.0404, -0.0068, + -0.0376, 0.0356, 0.0043, -0.0481, -0.0134, 0.0257, 0.0442, + 0.0234, 0.0394, 0.0376, -0.0147, -0.0098, 0.0474, -0.0102, + 0.0138, 0.0286, 0.0347, 0.0279, -0.0067, 0.0462, -0.0432, + 0.0247, 0.0174, -0.0305, -0.0317, -0.0068, 0.0264, -0.0257, + -0.0328, 0.0092, 0.0288, -0.002 , 0.0288, 0.028 , -0.0093, + 0.0178, -0.0365, -0.0086, -0.0133, -0.0309, 0.0473, -0.0149, + 0.0378, -0.0316, -0.0292, -0.0453, -0.0451, 0.0093, 0.0397, + -0.0361, -0.0168, -0.0494, -0.0143, -0.0405, -0.0349, 0.0069, + 0.0378, -0.0233, -0.0492, 0.018 , -0.0386, 0.0339, 0.0119, + 0.0454, 0.0118, -0.011 , -0.0254, 0.0266, -0.0366, -0.0211, + 0.0399, 0.0307, 0.035 , -0.0402, 0.0304, -0.0031, 0.0256, + 0.0134, -0.0019, -0.0235, -0.0058, -0.0117, 0.0051, -0.0451, + -0.0466, -0.0124, 0.0283, -0.0499, 0.0318, -0.0028, 0.0203, + 0.005 , 0.0085, 0.0048, 0.0277, 0.0159, -0.0149, 0.035 , + 0.0404, -0.01 , 0.0377, 0.0302, 0.0046, -0.0328, -0.0469, + 0.0071, -0.0382, -0.0214, 0.0429, 0.0145, -0.0279, -0.0172, + 0.0423, 0.041 , -0.0183, 0.0137, -0.0412, -0.0348, 0.0302, + 0.0248, 0.0051, -0.0298, -0.0103, -0.0333, -0.0399, 0.0485, + -0.0166, 0.0384, 0.0259, -0.0163, 0.0357, 0.0308, -0.0386, + 0.0481, -0.0446, -0.0282, -0.0037, 0.0202, 0.0216, 0.0113, + 0.0194, 0.0392, 0.0016, 0.0268, -0.0155, -0.027 , 0.02 , + 0.0216, -0.0009, 0.022 , 0. , 0.041 , 0.0133, -0.0382, + 0.0495, -0.0221, -0.0329, -0.0033, -0.0089, -0.0129, -0.0252, + 0.048 , -0.0307, -0.0357, 0.0033, -0.0412, -0.0407, 0.0455, + 0.0159, -0.0051, -0.0274, -0.0213, 0.0361, 0.0051, -0.0378, + 0.0084, 0.0066, -0.0103, -0.0037, 0.0478, -0.0278 +] \ No newline at end of file