From e7cf34d3c926795ccaef7a435b4799f00c090021 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Thu, 15 Mar 2012 16:44:15 -0400 Subject: [PATCH 1/4] Interstitical risk.py --- zipline/finance/risk.py | 348 +++++++++++++++++++++++----------------- 1 file changed, 200 insertions(+), 148 deletions(-) diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index a3f123f9..1a149f3a 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -1,70 +1,13 @@ -import datetime -import math -import pytz -import numpy as np -import numpy.linalg as la -import zipline.util as qutil -import zipline.protocol as zp -from pymongo import ASCENDING, DESCENDING +""" -class DailyReturn(): - - def __init__(self, date, returns): - self.date = date - self.returns = returns - - def to_dict(self): - d = { - 'dt': self.date, - 'returns': self.returns - } - - return d - - def __repr__(self): - return str(self.date) + " - " + str(self.returns) - -class RiskMetrics(): - def __init__(self, start_date, end_date, returns, trading_environment): - - self.treasury_curves = trading_environment.treasury_curves - self.start_date = start_date - self.end_date = end_date - self.trading_environment = trading_environment - self.algorithm_period_returns, self.algorithm_returns = self.calculate_period_returns(returns) - benchmark_returns = [x for x in self.trading_environment.benchmark_returns if x.date >= returns[0].date and x.date <= returns[-1].date] - - self.benchmark_period_returns, self.benchmark_returns = self.calculate_period_returns(benchmark_returns) - if(len(self.benchmark_returns) != len(self.algorithm_returns)): - message = "Mismatch between benchmark_returns ({bm_count}) and \ - algorithm_returns ({algo_count}) in range {start} : {end}" - message.format( - bm_count=len(self.benchmark_returns), - algo_count=len(self.algorithm_returns), - start=start_date, - end=end_date - ) - - raise Exception(messge) +Risk Report +=========== - self.trading_days = len(self.benchmark_returns) - self.benchmark_volatility = self.calculate_volatility(self.benchmark_returns) - self.algorithm_volatility = self.calculate_volatility(self.algorithm_returns) - self.treasury_period_return = self.choose_treasury() - self.sharpe = self.calculate_sharpe() - self.beta, self.algorithm_covariance, self.benchmark_variance, \ - self.condition_number, self.eigen_values = self.calculate_beta() - self.alpha = self.calculate_alpha() - self.excess_return = self.algorithm_period_returns - self.treasury_period_return - self.max_drawdown = self.calculate_max_drawdown() - - def to_dict(self): - """ +-----------------+----------------------------------------------------+ | key | value | +=================+====================================================+ - | trading_days | The number of trading days between self.start_date | - | | and self.end_date | + | trading_days | The number of trading days between self.start_date | + | | and self.end_date | +-----------------+----------------------------------------------------+ | benchmark_volat\| The volatility of the benchmark between | | ility | self.start_date and self.end_date. | @@ -81,7 +24,7 @@ class RiskMetrics(): +-----------------+----------------------------------------------------+ | beta | The _algorithm_ beta to the benchmark. | +-----------------+----------------------------------------------------+ - | alpha | The _algorithm_ alpha to the benchmark. | + | alpha | The _algorithm_ alpha to the benchmark. | +-----------------+----------------------------------------------------+ | excess_return | The excess return of the algorithm over the | | | benchmark. | @@ -90,6 +33,96 @@ class RiskMetrics(): | | for the portfolio returns between self.start_date | | | and self.end_date. | +-----------------+----------------------------------------------------+ + +""" + +import datetime +import math +import pytz +import numpy as np +import numpy.linalg as la +import zipline.util as qutil +import zipline.protocol as zp + + +def advance_by_months(dt, jump_in_months): + month = dt.month + jump_in_months + years = month / 12 + month = month % 12 + + # no remainder means that we are landing in december. + # modulo is, in a way, a zero indexed circular array. + # this is a way of converting to 1 indexed months. + # (in our modulo index, december is zeroth) + if(month == 0): + month = 12 + years = years - 1 + + return dt.replace(year = dt.year + years, month = month) + + +class DailyReturn(): + + def __init__(self, date, returns): + self.date = date + self.returns = returns + + def to_dict(self): + return { + 'dt' : self.date, + 'returns' : self.returns + } + + def __repr__(self): + return str(self.date) + " - " + str(self.returns) + + +class RiskMetrics(): + def __init__(self, start_date, end_date, returns, trading_environment): + + self.treasury_curves = trading_environment.treasury_curves + self.start_date = start_date + self.end_date = end_date + self.trading_environment = trading_environment + self.algorithm_period_returns, self.algorithm_returns = \ + self.calculate_period_returns(returns) + + benchmark_returns = [ + x for x in self.trading_environment.benchmark_returns + if x.date >= returns[0].date and x.date <= returns[-1].date + ] + + self.benchmark_period_returns, self.benchmark_returns = \ + self.calculate_period_returns(benchmark_returns) + + if(len(self.benchmark_returns) != len(self.algorithm_returns)): + message = "Mismatch between benchmark_returns ({bm_count}) and \ + algorithm_returns ({algo_count}) in range {start} : {end}" + message.format( + bm_count=len(self.benchmark_returns), + algo_count=len(self.algorithm_returns), + start=start_date, + end=end_date + ) + + # TODO: vestigal? + #raise Exception(messge) + + self.trading_days = len(self.benchmark_returns) + self.benchmark_volatility = self.calculate_volatility(self.benchmark_returns) + self.algorithm_volatility = self.calculate_volatility(self.algorithm_returns) + self.treasury_period_return = self.choose_treasury() + self.sharpe = self.calculate_sharpe() + self.beta, self.algorithm_covariance, self.benchmark_variance, \ + self.condition_number, self.eigen_values = self.calculate_beta() + self.alpha = self.calculate_alpha() + self.excess_return = self.algorithm_period_returns - self.treasury_period_return + self.max_drawdown = self.calculate_max_drawdown() + + def to_dict(self): + """ + Creates a dictionary representing the state of the risk report. + Returns a dict object of the form: """ return { 'trading_days' : self.trading_days, @@ -105,50 +138,74 @@ class RiskMetrics(): def __repr__(self): statements = [] - for metric in [ - "algorithm_period_returns", - "benchmark_period_returns", - "excess_return", - "trading_days", - "benchmark_volatility", - "algorithm_volatility", - "sharpe", - "algorithm_covariance", - "benchmark_variance", - "beta", - "alpha", - "max_drawdown", - "algorithm_returns", - "benchmark_returns", - "condition_number", + metrics = [ + "algorithm_period_returns" , + "benchmark_period_returns" , + "excess_return" , + "trading_days" , + "benchmark_volatility" , + "algorithm_volatility" , + "sharpe" , + "algorithm_covariance" , + "benchmark_variance" , + "beta" , + "alpha" , + "max_drawdown" , + "algorithm_returns" , + "benchmark_returns" , + "condition_number" , "eigen_values" - ]: + ] + + for metric in metrics: value = getattr(self, metric) statements.append("{m}:{v}".format(m=metric, v=value)) - + return '\n'.join(statements) - + def calculate_period_returns(self, daily_returns): + #TODO: replace this with pandas. - returns = [x.returns for x in daily_returns if x.date >= self.start_date and x.date <= self.end_date and self.trading_environment.is_trading_day(x.date)] + returns = [ + x.returns for x in daily_returns + if x.date >= self.start_date and + x.date <= self.end_date and + self.trading_environment.is_trading_day(x.date) + ] + period_returns = 1.0 + for r in returns: period_returns = period_returns * (1.0 + r) + period_returns = period_returns - 1.0 return period_returns, returns - + def calculate_volatility(self, daily_returns): #qutil.LOGGER.debug("trading days {td}".format(td=self.trading_days)) return np.std(daily_returns, ddof=1) * math.sqrt(self.trading_days) - + def calculate_sharpe(self): - return (self.algorithm_period_returns - self.treasury_period_return) / self.algorithm_volatility - + """ + http://en.wikipedia.org/wiki/Sharpe_ratio + """ + return ( (self.algorithm_period_returns - self.treasury_period_return) / + self.algorithm_volatility ) + def calculate_beta(self): - #it doesn't make much sense to calculate beta for less than two days, + """ + + .. math:: + \beta_a = \frac {\mathrm{Cov}(r_a,r_p)}{\mathrm{Var}(r_p)} + + http://en.wikipedia.org/wiki/Beta_(finance) + """ + + #it doesn't make much sense to calculate beta for less than two days, #so return none. if len(self.algorithm_returns) < 2: return 0.0, 0.0, 0.0, 0.0, [] + returns_matrix = np.vstack([self.algorithm_returns, self.benchmark_returns]) C = np.cov(returns_matrix) eigen_values = la.eigvals(C) @@ -156,12 +213,21 @@ class RiskMetrics(): algorithm_covariance = C[0][1] benchmark_variance = C[1][1] beta = C[0][1] / C[1][1] - - return beta, algorithm_covariance, benchmark_variance, condition_number, eigen_values - + + return ( + beta, + algorithm_covariance, + benchmark_variance, + condition_number, + eigen_values + ) + def calculate_alpha(self): + """ + http://en.wikipedia.org/wiki/Alpha_(investment) + """ return self.algorithm_period_returns - (self.treasury_period_return + self.beta * (self.benchmark_period_returns - self.treasury_period_return)) - + def calculate_max_drawdown(self): compounded_returns = [] cur_return = 0.0 @@ -173,23 +239,23 @@ class RiskMetrics(): qutil.LOGGER.warn("negative 100 percent return, zeroing the returns") cur_return = 0.0 compounded_returns.append(cur_return) - + cur_max = None max_drawdown = None for cur in compounded_returns: if cur_max == None or cur > cur_max: cur_max = cur - + drawdown = (cur - cur_max) if max_drawdown == None or drawdown < max_drawdown: max_drawdown = drawdown - + if max_drawdown == None: return 0.0 - + return 1.0 - math.exp(max_drawdown) - - + + def choose_treasury(self): td = self.end_date - self.start_date if td.days <= 31: @@ -212,18 +278,18 @@ class RiskMetrics(): self.treasury_duration = '10year' else: self.treasury_duration = '30year' - - + + 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): + for i in range(7): if(self.treasury_curves.has_key(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] @@ -236,25 +302,27 @@ class RiskMetrics(): message = "no rate for end date = {dt} and term = {term}. Using zero." message = message.format(dt=self.end_date,term=self.treasury_duration) raise Exception(message) - + + class RiskReport(): - + def __init__(self, algorithm_returns, trading_environment): - """ algorithm_returns needs to be a list of daily_return objects - sorted in date ascending order """ - + algorithm_returns needs to be a list of daily_return objects + sorted in date ascending order + """ + self.algorithm_returns = algorithm_returns self.trading_environment = trading_environment start_date = self.algorithm_returns[0].date end_date = self.algorithm_returns[-1].date - + self.month_periods = self.periodsInRange(1, start_date, end_date) self.three_month_periods = self.periodsInRange(3, start_date, end_date) self.six_month_periods = self.periodsInRange(6, start_date, end_date) self.year_periods = self.periodsInRange(12, start_date, end_date) - + def to_dict(self): """ RiskMetrics are calculated for rolling windows in four lengths:: @@ -262,28 +330,27 @@ class RiskReport(): - 3_month - 6_month - 12_month - + The return value of this funciton is a dictionary keyed by the above list of durations. The value of each entry is a list of RiskMetric - dicts of the same duration as denoted by the top_level key. - + dicts of the same duration as denoted by the top_level key. + See :py:meth:`RiskMetrics.to_dict` for the detailed list of fields - provided for each period. + provided for each period. """ - d = { + return { '1_month' : [x.to_dict() for x in self.month_periods], '3_month' : [x.to_dict() for x in self.three_year_periods], - '6_month' : [x.to_dict() for x in self.six_month_periods], + '6_month' : [x.to_dict() for x in self.six_month_periods], '12_month' : [x.to_dict() for x in self.month_periods] } - - return d - + def periodsInRange(self, months_per, start, end): one_day = datetime.timedelta(days = 1) ends = [] cur_start = start.replace(day=1) - #ensure that we have an end at the end of a calendar month, in case + + #ensure that we have an end at the end of a calendar month, in case #the return series ends mid-month... the_end = advance_by_months(end.replace(day=1),1) - one_day while True: @@ -291,39 +358,24 @@ class RiskReport(): if(cur_end > the_end): break cur_period_metrics = RiskMetrics( - start_date=cur_start, - end_date=cur_end, - returns=self.algorithm_returns, + start_date=cur_start, + end_date=cur_end, + returns=self.algorithm_returns, trading_environment=self.trading_environment ) - + ends.append(cur_period_metrics) cur_start = advance_by_months(cur_start, 1) - + return ends - + def find_metric_by_end(self, end_date, duration, metric): col = getattr(self, duration + "_periods") col = [getattr(x, metric) for x in col if x.end_date == end_date] if len(col) == 1: return col[0] return None - -def advance_by_months(dt, jump_in_months): - month = dt.month + jump_in_months - years = month / 12 - month = month % 12 - # no remainder means that we are landing in december. - # modulo is, in a way, a zero indexed circular array. - # this is a way of converting to 1 indexed months. - # (in our modulo index, december is zeroth) - if(month == 0): - month = 12 - years = years - 1 - - r = dt.replace(year = dt.year + years, month = month) - return r class TradingEnvironment(object): @@ -335,23 +387,23 @@ class TradingEnvironment(object): for bm in benchmark_returns: self.trading_days.append(bm.date) self.trading_day_map[bm.date] = bm - + def normalize_date(self, test_date): return datetime.datetime( - year=test_date.year, - month=test_date.month, - day=test_date.day, + year=test_date.year, + month=test_date.month, + day=test_date.day, tzinfo=pytz.utc ) - + def is_trading_day(self, test_date): dt = self.normalize_date(test_date) return self.trading_day_map.has_key(dt) - + def get_benchmark_daily_return(self, test_date): date = self.normalize_date(test_date) if self.trading_day_map.has_key(date): return self.trading_day_map[date].returns else: return 0.0 - + From 931f10fe3203293ecb17b77b546129c474e94b92 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Thu, 15 Mar 2012 18:33:34 -0400 Subject: [PATCH 2/4] Misc fixes, disabled Risk reporting until PR. --- zipline/finance/performance.py | 15 +++++++-------- zipline/protocol_utils.py | 5 +++++ zipline/test/client.py | 19 +++++++++++++++++++ zipline/test/test_messaging.py | 9 +++++---- zipline/zmq_utils.py | 4 ++-- 5 files changed, 38 insertions(+), 14 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 21557dc6..6039d584 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -290,17 +290,16 @@ class PerformanceTracker(): ) def handle_simulation_end(self): - assert False - self.risk_report = risk.RiskReport( - self.returns, - self.trading_environment - ) + #self.risk_report = risk.RiskReport( + #self.returns, + #self.trading_environment + #) # Output Results - if self.result_stream: - # TODO: proper framing - self.result_stream.send_pyobj(self.risk_report.to_dict()) + #if self.result_stream: + ## TODO: proper framing + #self.result_stream.send_pyobj(self.risk_report.to_dict()) self.result_stream.send_pyobj(None) diff --git a/zipline/protocol_utils.py b/zipline/protocol_utils.py index 0e6cb858..4b38fdda 100644 --- a/zipline/protocol_utils.py +++ b/zipline/protocol_utils.py @@ -1,6 +1,8 @@ import copy from ctypes import Structure, c_ubyte +from pandas import Series + def Enum(*options): """ Fast enums are very important when we want really tight zmq @@ -85,3 +87,6 @@ class namedict(object): def has_attr(self, name): return self.__dict__.has_key(name) + + def as_series(self): + return Series(self.__dict__, self.__dict__.keys()) diff --git a/zipline/test/client.py b/zipline/test/client.py index f9ddf443..0167bf7b 100644 --- a/zipline/test/client.py +++ b/zipline/test/client.py @@ -1,3 +1,5 @@ +from gevent_zeromq import zmq + import zipline.util as qutil import zipline.messaging as qmsg import zipline.protocol as zp @@ -14,6 +16,12 @@ class TestClient(qmsg.Component): self.received_count = 0 self.prev_dt = None + self.result_streams = [] + + # Maximum outgoing result streams, really shouldn't ever + # need more than 1. + self.max_outgoing = 5 + @property def get_id(self): return "TEST_CLIENT" @@ -25,6 +33,17 @@ class TestClient(qmsg.Component): def open(self): self.data_feed = self.connect_result() + def result_stream(self, zmq_socket, context=None): + """ + Asynchronously grab a socket to stream results out on. + """ + ctx = context or zmq.Context.instance() + sock = ctx.socket(zmq.PULL) + sock.bind(zmq_socket) + + # Add + self.result_streams.append( sock ) + def do_work(self): socks = dict(self.poll.poll(self.heartbeat_timeout)) diff --git a/zipline/test/test_messaging.py b/zipline/test/test_messaging.py index c901f984..0dde6799 100644 --- a/zipline/test/test_messaging.py +++ b/zipline/test/test_messaging.py @@ -13,9 +13,7 @@ from zipline.test.transform import DivideByZeroTransform from nose.tools import timed -# Should not inherit form TestCase since test runners will pick -# it up as a test. Its a Mixin of sorts at this point. -class SimulatorTestCase(object): +class BaseSimulator(object): # Leased sockets is a defaultdict keyed by the test case. # This lets you debug the sockets being allocated in the @@ -25,7 +23,6 @@ class SimulatorTestCase(object): # 'test_orders' : ['tcp : //127.0.0.1 : 1000', ... ], # 'test_performance' : ['tcp : //127.0.0.1 : 1025', ... ], # } - leased_sockets = defaultdict(list) def setUp(self): @@ -79,6 +76,10 @@ class SimulatorTestCase(object): def unallocate_sockets(self): self.allocator.reaquire(*self.leased_sockets[self.id()]) +# Should not inherit form TestCase since test runners will pick +# it up as a test. Its a Mixin of sorts at this point. +class SimulatorTestCase(BaseSimulator): + # ------- # Cases # ------- diff --git a/zipline/zmq_utils.py b/zipline/zmq_utils.py index e7c09047..e13a5e34 100644 --- a/zipline/zmq_utils.py +++ b/zipline/zmq_utils.py @@ -5,11 +5,11 @@ def ZmqConsole(sock_typ, socket_addr, sock_conn=None, context=None): context = context or zmq.Context.instance() socket = context.socket(zmq.PULL) - socket.connect('tcp://127.0.0.1:3141') + socket.bind(socket_addr) def console(): while True: - msg = socket.recv() + msg = socket.recv_pyobj() print msg import pdb; pdb.set_trace() From 89485c6ae372d63dd7137a8f12813390e1dc749f Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 16 Mar 2012 16:42:39 -0400 Subject: [PATCH 3/4] Hack to ignore dataframes in process_event, code seems to be written against namedicts. --- zipline/finance/performance.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 9a46f6bb..bf9276c4 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -209,6 +209,15 @@ class PerformanceTracker(): def process_event(self, event): self.event_count += 1 + # TODO: This seems to get both dataframes and namedicts? + # I can't find a corresponding test for this + # function so not sure if this is the desired behavior or + # something is going wrong upstream... + + # hack for now + if isinstance(event, pandas.DataFrame): + return + if(event.dt >= self.market_close): self.handle_market_close() From ce16122be3e9c95e1c361d154a32b6632a7de3a3 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 16 Mar 2012 18:18:42 -0400 Subject: [PATCH 4/4] Remove dropout on dataframe. --- zipline/finance/performance.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index bf9276c4..9a46f6bb 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -209,15 +209,6 @@ class PerformanceTracker(): def process_event(self, event): self.event_count += 1 - # TODO: This seems to get both dataframes and namedicts? - # I can't find a corresponding test for this - # function so not sure if this is the desired behavior or - # something is going wrong upstream... - - # hack for now - if isinstance(event, pandas.DataFrame): - return - if(event.dt >= self.market_close): self.handle_market_close()