wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-10 22:32:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #11 from quantopian/portfolio

Browse Source 
Portfolio
This commit is contained in:
fawce
2012-03-07 15:26:05 -08:00
parent 31cd993442 2ba7d526a6
commit 3c0abbf946
10 changed files with 687 additions and 828 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dataloader.py
-39
View File
@@ -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)
etc/jenkins.sh
-7
View File
@@ -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
zipline/daemon.py
-143
View File
@@ -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().
"""
zipline/db.py
-76
View File
@@ -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()
zipline/finance/data.py
-498
View File
@@ -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<company_id>[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()
zipline/finance/performance.py
+197
View File
@@ -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
zipline/finance/risk.py
+38 -49
View File
@@ -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()
zipline/test/factory.py
+65 -4
View File
@@ -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)
zipline/test/test_finance.py
+105 -12
View File
@@ -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()))
zipline/test/test_risk.py
+282
View File
@@ -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
]