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

MAINT: Break period and cumulative risk metrics into submodules.

Browse Source 
In anticipation of changing the sharpe, beta, et al. calculations
dependent on whether the period returns or the overall returns
are being calculated.
This commit is contained in:
Eddie Hebert
2013-08-06 17:49:19 -04:00
parent 66e7f48cdd
commit 5b2a23ddd0
5 changed files with 776 additions and 650 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/finance/risk/__init__.py
+3 -5
View File
@@ -13,11 +13,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from . risk import (
RiskReport,
RiskMetricsPeriod,
RiskMetricsCumulative,
)
from . report import RiskReport
from . period import RiskMetricsPeriod
from . cumulative import RiskMetricsCumulative
__all__ = [
zipline/finance/risk/cumulative.py
+346
View File
@@ -0,0 +1,346 @@
#
# Copyright 2013 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logbook
import math
import numpy as np
import numpy.linalg as la
import zipline.finance.trading as trading
import pandas as pd
from . risk import (
alpha,
check_entry,
choose_treasury,
information_ratio,
sharpe_ratio,
sortino_ratio,
)
log = logbook.Logger('Risk Cumulative')
class RiskMetricsCumulative(object):
"""
:Usage:
Instantiate RiskMetricsCumulative once.
Call update() method on each dt to update the metrics.
"""
def __init__(self, sim_params):
self.treasury_curves = trading.environment.treasury_curves
self.start_date = sim_params.period_start.replace(
hour=0, minute=0, second=0, microsecond=0
)
self.end_date = sim_params.period_end.replace(
hour=0, minute=0, second=0, microsecond=0
)
all_trading_days = trading.environment.trading_days
mask = ((all_trading_days >= self.start_date) &
(all_trading_days <= self.end_date))
self.trading_days = all_trading_days[mask]
if sim_params.period_end not in self.trading_days:
last_day = pd.tseries.index.DatetimeIndex(
[sim_params.period_end]
)
self.trading_days = self.trading_days.append(last_day)
self.sim_params = sim_params
if sim_params.emission_rate == 'daily':
self.initialize_daily_indices()
elif sim_params.emission_rate == 'minute':
self.initialize_minute_indices(sim_params)
self.algorithm_returns = None
self.benchmark_returns = None
self.compounded_log_returns = []
self.moving_avg = []
self.algorithm_volatility = []
self.benchmark_volatility = []
self.algorithm_period_returns = []
self.benchmark_period_returns = []
self.algorithm_covariance = None
self.benchmark_variance = None
self.condition_number = None
self.eigen_values = None
self.sharpe = []
self.sortino = []
self.information = []
self.beta = []
self.alpha = []
self.max_drawdown = 0
self.current_max = -np.inf
self.excess_returns = []
self.daily_treasury = {}
def initialize_minute_indices(self, sim_params):
self.algorithm_returns_cont = pd.Series(index=pd.date_range(
sim_params.first_open, sim_params.last_close,
freq="Min"))
self.benchmark_returns_cont = pd.Series(index=pd.date_range(
sim_params.first_open, sim_params.last_close,
freq="Min"))
def initialize_daily_indices(self):
self.algorithm_returns_cont = pd.Series(index=self.trading_days)
self.benchmark_returns_cont = pd.Series(index=self.trading_days)
@property
def last_return_date(self):
return self.algorithm_returns.index[-1]
def update(self, dt, algorithm_returns, benchmark_returns):
self.algorithm_returns_cont[dt] = algorithm_returns
self.algorithm_returns = self.algorithm_returns_cont.valid()
self.benchmark_returns_cont[dt] = benchmark_returns
self.benchmark_returns = self.benchmark_returns_cont.valid()
self.num_trading_days = len(self.algorithm_returns)
self.update_compounded_log_returns()
self.algorithm_period_returns.append(
self.calculate_period_returns(self.algorithm_returns))
self.benchmark_period_returns.append(
self.calculate_period_returns(self.benchmark_returns))
if not self.algorithm_returns.index.equals(
self.benchmark_returns.index
):
message = "Mismatch between benchmark_returns ({bm_count}) and \
algorithm_returns ({algo_count}) in range {start} : {end} on {dt}"
message = message.format(
bm_count=len(self.benchmark_returns),
algo_count=len(self.algorithm_returns),
start=self.start_date,
end=self.end_date,
dt=dt
)
raise Exception(message)
self.update_current_max()
self.benchmark_volatility.append(
self.calculate_volatility(self.benchmark_returns))
self.algorithm_volatility.append(
self.calculate_volatility(self.algorithm_returns))
# caching the treasury rates for the minutely case is a
# big speedup, because it avoids searching the treasury
# curves on every minute.
treasury_end = self.algorithm_returns.index[-1].replace(
hour=0, minute=0)
if treasury_end not in self.daily_treasury:
treasury_period_return = choose_treasury(
self.treasury_curves,
self.start_date,
self.algorithm_returns.index[-1]
)
self.daily_treasury[treasury_end] =\
treasury_period_return
self.treasury_period_return = \
self.daily_treasury[treasury_end]
self.excess_returns.append(
self.algorithm_period_returns[-1] - self.treasury_period_return)
self.beta.append(self.calculate_beta()[0])
self.alpha.append(self.calculate_alpha())
self.sharpe.append(self.calculate_sharpe())
self.sortino.append(self.calculate_sortino())
self.information.append(self.calculate_information())
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:
"""
period_label = self.last_return_date.strftime("%Y-%m")
rval = {
'trading_days': len(self.algorithm_returns.valid()),
'benchmark_volatility': self.benchmark_volatility[-1],
'algo_volatility': self.algorithm_volatility[-1],
'treasury_period_return': self.treasury_period_return,
'algorithm_period_return': self.algorithm_period_returns[-1],
'benchmark_period_return': self.benchmark_period_returns[-1],
'beta': self.beta[-1],
'alpha': self.alpha[-1],
'excess_return': self.excess_returns[-1],
'max_drawdown': self.max_drawdown,
'period_label': period_label
}
rval['sharpe'] = self.sharpe[-1]
rval['sortino'] = self.sortino[-1]
rval['information'] = self.information[-1]
return {k: None
if check_entry(k, v)
else v for k, v in rval.iteritems()}
def __repr__(self):
statements = []
metrics = [
"algorithm_period_returns",
"benchmark_period_returns",
"excess_returns",
"trading_days",
"benchmark_volatility",
"algorithm_volatility",
"sharpe",
"sortino",
"information",
"algorithm_covariance",
"benchmark_variance",
"beta",
"alpha",
"max_drawdown",
"algorithm_returns",
"benchmark_returns",
"condition_number",
"eigen_values"
]
for metric in metrics:
value = getattr(self, metric)
if isinstance(value, list):
if len(value) == 0:
value = np.nan
else:
value = value[-1]
statements.append("{m}:{v}".format(m=metric, v=value))
return '\n'.join(statements)
def update_compounded_log_returns(self):
if len(self.algorithm_returns) == 0:
return
try:
compound = math.log(1 + self.algorithm_returns[
self.algorithm_returns.last_valid_index()])
except ValueError:
compound = 0.0
# BUG? Shouldn't this be set to log(1.0 + 0) ?
if len(self.compounded_log_returns) == 0:
self.compounded_log_returns.append(compound)
else:
self.compounded_log_returns.append(
self.compounded_log_returns[-1] +
compound
)
def calculate_period_returns(self, returns):
returns = np.array(returns)
return (1. + returns).prod() - 1
def update_current_max(self):
if len(self.compounded_log_returns) == 0:
return
if self.current_max < self.compounded_log_returns[-1]:
self.current_max = self.compounded_log_returns[-1]
def calculate_max_drawdown(self):
if len(self.compounded_log_returns) == 0:
return self.max_drawdown
cur_drawdown = 1.0 - math.exp(
self.compounded_log_returns[-1] -
self.current_max)
if self.max_drawdown < cur_drawdown:
return cur_drawdown
else:
return self.max_drawdown
def calculate_sharpe(self):
"""
http://en.wikipedia.org/wiki/Sharpe_ratio
"""
return sharpe_ratio(self.algorithm_volatility[-1],
self.algorithm_period_returns[-1],
self.treasury_period_return)
def calculate_sortino(self, mar=None):
"""
http://en.wikipedia.org/wiki/Sortino_ratio
"""
if mar is None:
mar = self.treasury_period_return
return sortino_ratio(np.array(self.algorithm_returns),
self.algorithm_period_returns[-1],
mar)
def calculate_information(self):
"""
http://en.wikipedia.org/wiki/Information_ratio
"""
A = np.array
return information_ratio(A(self.algorithm_returns),
A(self.benchmark_returns))
def calculate_alpha(self):
"""
http://en.wikipedia.org/wiki/Alpha_(investment)
"""
return alpha(self.algorithm_period_returns[-1],
self.treasury_period_return,
self.benchmark_period_returns[-1],
self.beta[-1])
def calculate_volatility(self, daily_returns):
return np.std(daily_returns, ddof=1) * math.sqrt(self.num_trading_days)
def calculate_beta(self):
"""
.. 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, ddof=1)
eigen_values = la.eigvals(C)
condition_number = max(eigen_values) / min(eigen_values)
algorithm_covariance = C[0][1]
benchmark_variance = C[1][1]
beta = algorithm_covariance / benchmark_variance
return (
beta,
algorithm_covariance,
benchmark_variance,
condition_number,
eigen_values
)
zipline/finance/risk/period.py
+278
View File
@@ -0,0 +1,278 @@
#
# Copyright 2013 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logbook
import math
import numpy as np
import numpy.linalg as la
import zipline.finance.trading as trading
import pandas as pd
from . risk import (
alpha,
check_entry,
choose_treasury,
information_ratio,
sharpe_ratio,
sortino_ratio,
)
log = logbook.Logger('Risk Period')
class RiskMetricsPeriod(object):
def __init__(self, start_date, end_date, returns,
benchmark_returns=None):
treasury_curves = trading.environment.treasury_curves
if treasury_curves.index[-1] >= start_date:
mask = ((treasury_curves.index >= start_date) &
(treasury_curves.index <= end_date))
self.treasury_curves = treasury_curves[mask]
else:
# our test is beyond the treasury curve history
# so we'll use the last available treasury curve
self.treasury_curves = treasury_curves[-1:]
self.start_date = start_date
self.end_date = end_date
if benchmark_returns is None:
benchmark_returns = [
x for x in trading.environment.benchmark_returns
if x.date >= returns[0].date and
x.date <= returns[-1].date
]
self.algorithm_returns = self.mask_returns_to_period(returns)
self.benchmark_returns = self.mask_returns_to_period(benchmark_returns)
self.calculate_metrics()
def calculate_metrics(self):
self.benchmark_period_returns = \
self.calculate_period_returns(self.benchmark_returns)
self.algorithm_period_returns = \
self.calculate_period_returns(self.algorithm_returns)
if not self.algorithm_returns.index.equals(
self.benchmark_returns.index
):
message = "Mismatch between benchmark_returns ({bm_count}) and \
algorithm_returns ({algo_count}) in range {start} : {end}"
message = message.format(
bm_count=len(self.benchmark_returns),
algo_count=len(self.algorithm_returns),
start=self.start_date,
end=self.end_date
)
raise Exception(message)
self.num_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 = choose_treasury(
self.treasury_curves,
self.start_date,
self.end_date
)
self.sharpe = self.calculate_sharpe()
self.sortino = self.calculate_sortino()
self.information = self.calculate_information()
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:
"""
period_label = self.end_date.strftime("%Y-%m")
rval = {
'trading_days': self.num_trading_days,
'benchmark_volatility': self.benchmark_volatility,
'algo_volatility': self.algorithm_volatility,
'treasury_period_return': self.treasury_period_return,
'algorithm_period_return': self.algorithm_period_returns,
'benchmark_period_return': self.benchmark_period_returns,
'sharpe': self.sharpe,
'sortino': self.sortino,
'information': self.information,
'beta': self.beta,
'alpha': self.alpha,
'excess_return': self.excess_return,
'max_drawdown': self.max_drawdown,
'period_label': period_label
}
return {k: None if check_entry(k, v) else v
for k, v in rval.iteritems()}
def __repr__(self):
statements = []
metrics = [
"algorithm_period_returns",
"benchmark_period_returns",
"excess_return",
"num_trading_days",
"benchmark_volatility",
"algorithm_volatility",
"sharpe",
"sortino",
"information",
"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 mask_returns_to_period(self, daily_returns):
if isinstance(daily_returns, list):
returns = pd.Series([x.returns for x in daily_returns],
index=[x.date for x in daily_returns])
else: # otherwise we're receiving an index already
returns = daily_returns
trade_days = trading.environment.trading_days
trade_day_mask = returns.index.normalize().isin(trade_days)
mask = ((returns.index >= self.start_date) &
(returns.index <= self.end_date) & trade_day_mask)
returns = returns[mask]
return returns
def calculate_period_returns(self, returns):
period_returns = (1. + returns).prod() - 1
return period_returns
def calculate_volatility(self, daily_returns):
return np.std(daily_returns, ddof=1) * math.sqrt(self.num_trading_days)
def calculate_sharpe(self):
"""
http://en.wikipedia.org/wiki/Sharpe_ratio
"""
return sharpe_ratio(self.algorithm_volatility,
self.algorithm_period_returns,
self.treasury_period_return)
def calculate_sortino(self, mar=None):
"""
http://en.wikipedia.org/wiki/Sortino_ratio
"""
if mar is None:
mar = self.treasury_period_return
return sortino_ratio(self.algorithm_returns,
self.algorithm_period_returns,
mar)
def calculate_information(self):
"""
http://en.wikipedia.org/wiki/Information_ratio
"""
return information_ratio(self.algorithm_returns,
self.benchmark_returns)
def calculate_beta(self):
"""
.. 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, ddof=1)
eigen_values = la.eigvals(C)
condition_number = max(eigen_values) / min(eigen_values)
algorithm_covariance = C[0][1]
benchmark_variance = C[1][1]
beta = algorithm_covariance / benchmark_variance
return (
beta,
algorithm_covariance,
benchmark_variance,
condition_number,
eigen_values
)
def calculate_alpha(self):
"""
http://en.wikipedia.org/wiki/Alpha_(investment)
"""
return alpha(self.algorithm_period_returns,
self.treasury_period_return,
self.benchmark_period_returns,
self.beta)
def calculate_max_drawdown(self):
compounded_returns = []
cur_return = 0.0
for r in self.algorithm_returns:
try:
cur_return += math.log(1.0 + r)
#this is a guard for a single day returning -100%
except ValueError:
log.debug("{cur} return, zeroing the returns".format(
cur=cur_return))
cur_return = 0.0
# BUG? Shouldn't this be set to log(1.0 + 0) ?
compounded_returns.append(cur_return)
cur_max = None
max_drawdown = None
for cur in compounded_returns:
if cur_max is None or cur > cur_max:
cur_max = cur
drawdown = (cur - cur_max)
if max_drawdown is None or drawdown < max_drawdown:
max_drawdown = drawdown
if max_drawdown is None:
return 0.0
return 1.0 - math.exp(max_drawdown)
zipline/finance/risk/report.py
+149
View File
@@ -0,0 +1,149 @@
#
# Copyright 2013 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Risk Report
===========
+-----------------+----------------------------------------------------+
| key | value |
+=================+====================================================+
| 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. |
+-----------------+----------------------------------------------------+
| algo_volatility | The volatility of the algo between self.start_date |
| | and self.end_date. |
+-----------------+----------------------------------------------------+
| treasury_period\| The return of treasuries over the period. Treasury |
| _return | maturity is chosen to match the duration of the |
| | test period. |
+-----------------+----------------------------------------------------+
| sharpe | The sharpe ratio based on the _algorithm_ (rather |
| | than the static portfolio) returns. |
+-----------------+----------------------------------------------------+
| information | The information ratio based on the _algorithm_ |
| | (rather than the static portfolio) returns. |
+-----------------+----------------------------------------------------+
| beta | The _algorithm_ beta to the benchmark. |
+-----------------+----------------------------------------------------+
| alpha | The _algorithm_ alpha to the benchmark. |
+-----------------+----------------------------------------------------+
| excess_return | The excess return of the algorithm over the |
| | treasuries. |
+-----------------+----------------------------------------------------+
| max_drawdown | The largest relative peak to relative trough move |
| | for the portfolio returns between self.start_date |
| | and self.end_date. |
+-----------------+----------------------------------------------------+
"""
import logbook
import datetime
from dateutil.relativedelta import relativedelta
from zipline.utils.date_utils import epoch_now
from . period import RiskMetricsPeriod
log = logbook.Logger('Risk Report')
class RiskReport(object):
def __init__(self, algorithm_returns, sim_params, benchmark_returns=None):
"""
algorithm_returns needs to be a list of daily_return objects
sorted in date ascending order
"""
self.algorithm_returns = algorithm_returns
self.sim_params = sim_params
self.benchmark_returns = benchmark_returns
self.created = epoch_now()
if len(self.algorithm_returns) == 0:
start_date = self.sim_params.period_start
end_date = self.sim_params.period_end
else:
# FIXME: Papering over multiple algorithm_return types
if isinstance(self.algorithm_returns, list):
start_date = self.algorithm_returns[0].date
end_date = self.algorithm_returns[-1].date
else:
start_date = self.algorithm_returns.index[0]
end_date = self.algorithm_returns.index[-1]
self.month_periods = self.periods_in_range(1, start_date, end_date)
self.three_month_periods = self.periods_in_range(3, start_date,
end_date)
self.six_month_periods = self.periods_in_range(6, start_date, end_date)
self.year_periods = self.periods_in_range(12, start_date, end_date)
def to_dict(self):
"""
RiskMetrics are calculated for rolling windows in four lengths::
- 1_month
- 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.
See :py:meth:`RiskMetrics.to_dict` for the detailed list of fields
provided for each period.
"""
return {
'one_month': [x.to_dict() for x in self.month_periods],
'three_month': [x.to_dict() for x in self.three_month_periods],
'six_month': [x.to_dict() for x in self.six_month_periods],
'twelve_month': [x.to_dict() for x in self.year_periods],
'created': self.created
}
def periods_in_range(self, months_per, start, end):
one_day = datetime.timedelta(days=1)
ends = []
cur_start = start.replace(day=1)
# in edge cases (all sids filtered out, start/end are adjacent)
# a test will not generate any returns data
if len(self.algorithm_returns) == 0:
return ends
#ensure that we have an end at the end of a calendar month, in case
#the return series ends mid-month...
the_end = end.replace(day=1) + relativedelta(months=1) - one_day
while True:
cur_end = cur_start + relativedelta(months=months_per) - one_day
if(cur_end > the_end):
break
cur_period_metrics = RiskMetricsPeriod(
start_date=cur_start,
end_date=cur_end,
returns=self.algorithm_returns,
benchmark_returns=self.benchmark_returns
)
ends.append(cur_period_metrics)
cur_start = cur_start + relativedelta(months=1)
return ends
zipline/finance/risk/risk.py
-645
View File
@@ -56,18 +56,11 @@ Risk Report
"""
import logbook
import datetime
import math
import numpy as np
import numpy.linalg as la
from dateutil.relativedelta import relativedelta
import zipline.finance.trading as trading
from zipline.utils.date_utils import epoch_now
import zipline.utils.math_utils as zp_math
import pandas as pd
log = logbook.Logger('Risk')
@@ -290,641 +283,3 @@ that date doesn't exceed treasury history range."
term=treasury_duration
)
raise Exception(message)
class RiskMetricsPeriod(object):
def __init__(self, start_date, end_date, returns,
benchmark_returns=None):
treasury_curves = trading.environment.treasury_curves
if treasury_curves.index[-1] >= start_date:
mask = ((treasury_curves.index >= start_date) &
(treasury_curves.index <= end_date))
self.treasury_curves = treasury_curves[mask]
else:
# our test is beyond the treasury curve history
# so we'll use the last available treasury curve
self.treasury_curves = treasury_curves[-1:]
self.start_date = start_date
self.end_date = end_date
if benchmark_returns is None:
benchmark_returns = [
x for x in trading.environment.benchmark_returns
if x.date >= returns[0].date and
x.date <= returns[-1].date
]
self.algorithm_returns = self.mask_returns_to_period(returns)
self.benchmark_returns = self.mask_returns_to_period(benchmark_returns)
self.calculate_metrics()
def calculate_metrics(self):
self.benchmark_period_returns = \
self.calculate_period_returns(self.benchmark_returns)
self.algorithm_period_returns = \
self.calculate_period_returns(self.algorithm_returns)
if not self.algorithm_returns.index.equals(
self.benchmark_returns.index
):
message = "Mismatch between benchmark_returns ({bm_count}) and \
algorithm_returns ({algo_count}) in range {start} : {end}"
message = message.format(
bm_count=len(self.benchmark_returns),
algo_count=len(self.algorithm_returns),
start=self.start_date,
end=self.end_date
)
raise Exception(message)
self.num_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 = choose_treasury(
self.treasury_curves,
self.start_date,
self.end_date
)
self.sharpe = self.calculate_sharpe()
self.sortino = self.calculate_sortino()
self.information = self.calculate_information()
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:
"""
period_label = self.end_date.strftime("%Y-%m")
rval = {
'trading_days': self.num_trading_days,
'benchmark_volatility': self.benchmark_volatility,
'algo_volatility': self.algorithm_volatility,
'treasury_period_return': self.treasury_period_return,
'algorithm_period_return': self.algorithm_period_returns,
'benchmark_period_return': self.benchmark_period_returns,
'sharpe': self.sharpe,
'sortino': self.sortino,
'information': self.information,
'beta': self.beta,
'alpha': self.alpha,
'excess_return': self.excess_return,
'max_drawdown': self.max_drawdown,
'period_label': period_label
}
return {k: None if check_entry(k, v) else v
for k, v in rval.iteritems()}
def __repr__(self):
statements = []
metrics = [
"algorithm_period_returns",
"benchmark_period_returns",
"excess_return",
"num_trading_days",
"benchmark_volatility",
"algorithm_volatility",
"sharpe",
"sortino",
"information",
"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 mask_returns_to_period(self, daily_returns):
if isinstance(daily_returns, list):
returns = pd.Series([x.returns for x in daily_returns],
index=[x.date for x in daily_returns])
else: # otherwise we're receiving an index already
returns = daily_returns
trade_days = trading.environment.trading_days
trade_day_mask = returns.index.normalize().isin(trade_days)
mask = ((returns.index >= self.start_date) &
(returns.index <= self.end_date) & trade_day_mask)
returns = returns[mask]
return returns
def calculate_period_returns(self, returns):
period_returns = (1. + returns).prod() - 1
return period_returns
def calculate_volatility(self, daily_returns):
return np.std(daily_returns, ddof=1) * math.sqrt(self.num_trading_days)
def calculate_sharpe(self):
"""
http://en.wikipedia.org/wiki/Sharpe_ratio
"""
return sharpe_ratio(self.algorithm_volatility,
self.algorithm_period_returns,
self.treasury_period_return)
def calculate_sortino(self, mar=None):
"""
http://en.wikipedia.org/wiki/Sortino_ratio
"""
if mar is None:
mar = self.treasury_period_return
return sortino_ratio(self.algorithm_returns,
self.algorithm_period_returns,
mar)
def calculate_information(self):
"""
http://en.wikipedia.org/wiki/Information_ratio
"""
return information_ratio(self.algorithm_returns,
self.benchmark_returns)
def calculate_beta(self):
"""
.. 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, ddof=1)
eigen_values = la.eigvals(C)
condition_number = max(eigen_values) / min(eigen_values)
algorithm_covariance = C[0][1]
benchmark_variance = C[1][1]
beta = algorithm_covariance / benchmark_variance
return (
beta,
algorithm_covariance,
benchmark_variance,
condition_number,
eigen_values
)
def calculate_alpha(self):
"""
http://en.wikipedia.org/wiki/Alpha_(investment)
"""
return alpha(self.algorithm_period_returns,
self.treasury_period_return,
self.benchmark_period_returns,
self.beta)
def calculate_max_drawdown(self):
compounded_returns = []
cur_return = 0.0
for r in self.algorithm_returns:
try:
cur_return += math.log(1.0 + r)
#this is a guard for a single day returning -100%
except ValueError:
log.debug("{cur} return, zeroing the returns".format(
cur=cur_return))
cur_return = 0.0
# BUG? Shouldn't this be set to log(1.0 + 0) ?
compounded_returns.append(cur_return)
cur_max = None
max_drawdown = None
for cur in compounded_returns:
if cur_max is None or cur > cur_max:
cur_max = cur
drawdown = (cur - cur_max)
if max_drawdown is None or drawdown < max_drawdown:
max_drawdown = drawdown
if max_drawdown is None:
return 0.0
return 1.0 - math.exp(max_drawdown)
class RiskMetricsCumulative(object):
"""
:Usage:
Instantiate RiskMetricsCumulative once.
Call update() method on each dt to update the metrics.
"""
def __init__(self, sim_params):
self.treasury_curves = trading.environment.treasury_curves
self.start_date = sim_params.period_start.replace(
hour=0, minute=0, second=0, microsecond=0
)
self.end_date = sim_params.period_end.replace(
hour=0, minute=0, second=0, microsecond=0
)
all_trading_days = trading.environment.trading_days
mask = ((all_trading_days >= self.start_date) &
(all_trading_days <= self.end_date))
self.trading_days = all_trading_days[mask]
if sim_params.period_end not in self.trading_days:
last_day = pd.tseries.index.DatetimeIndex(
[sim_params.period_end]
)
self.trading_days = self.trading_days.append(last_day)
self.sim_params = sim_params
if sim_params.emission_rate == 'daily':
self.initialize_daily_indices()
elif sim_params.emission_rate == 'minute':
self.initialize_minute_indices(sim_params)
self.algorithm_returns = None
self.benchmark_returns = None
self.compounded_log_returns = []
self.moving_avg = []
self.algorithm_volatility = []
self.benchmark_volatility = []
self.algorithm_period_returns = []
self.benchmark_period_returns = []
self.algorithm_covariance = None
self.benchmark_variance = None
self.condition_number = None
self.eigen_values = None
self.sharpe = []
self.sortino = []
self.information = []
self.beta = []
self.alpha = []
self.max_drawdown = 0
self.current_max = -np.inf
self.excess_returns = []
self.daily_treasury = {}
def initialize_minute_indices(self, sim_params):
self.algorithm_returns_cont = pd.Series(index=pd.date_range(
sim_params.first_open, sim_params.last_close,
freq="Min"))
self.benchmark_returns_cont = pd.Series(index=pd.date_range(
sim_params.first_open, sim_params.last_close,
freq="Min"))
def initialize_daily_indices(self):
self.algorithm_returns_cont = pd.Series(index=self.trading_days)
self.benchmark_returns_cont = pd.Series(index=self.trading_days)
@property
def last_return_date(self):
return self.algorithm_returns.index[-1]
def update(self, dt, algorithm_returns, benchmark_returns):
self.algorithm_returns_cont[dt] = algorithm_returns
self.algorithm_returns = self.algorithm_returns_cont.valid()
self.benchmark_returns_cont[dt] = benchmark_returns
self.benchmark_returns = self.benchmark_returns_cont.valid()
self.num_trading_days = len(self.algorithm_returns)
self.update_compounded_log_returns()
self.algorithm_period_returns.append(
self.calculate_period_returns(self.algorithm_returns))
self.benchmark_period_returns.append(
self.calculate_period_returns(self.benchmark_returns))
if not self.algorithm_returns.index.equals(
self.benchmark_returns.index
):
message = "Mismatch between benchmark_returns ({bm_count}) and \
algorithm_returns ({algo_count}) in range {start} : {end} on {dt}"
message = message.format(
bm_count=len(self.benchmark_returns),
algo_count=len(self.algorithm_returns),
start=self.start_date,
end=self.end_date,
dt=dt
)
raise Exception(message)
self.update_current_max()
self.benchmark_volatility.append(
self.calculate_volatility(self.benchmark_returns))
self.algorithm_volatility.append(
self.calculate_volatility(self.algorithm_returns))
# caching the treasury rates for the minutely case is a
# big speedup, because it avoids searching the treasury
# curves on every minute.
treasury_end = self.algorithm_returns.index[-1].replace(
hour=0, minute=0)
if treasury_end not in self.daily_treasury:
treasury_period_return = choose_treasury(
self.treasury_curves,
self.start_date,
self.algorithm_returns.index[-1]
)
self.daily_treasury[treasury_end] =\
treasury_period_return
self.treasury_period_return = \
self.daily_treasury[treasury_end]
self.excess_returns.append(
self.algorithm_period_returns[-1] - self.treasury_period_return)
self.beta.append(self.calculate_beta()[0])
self.alpha.append(self.calculate_alpha())
self.sharpe.append(self.calculate_sharpe())
self.sortino.append(self.calculate_sortino())
self.information.append(self.calculate_information())
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:
"""
period_label = self.last_return_date.strftime("%Y-%m")
rval = {
'trading_days': len(self.algorithm_returns.valid()),
'benchmark_volatility': self.benchmark_volatility[-1],
'algo_volatility': self.algorithm_volatility[-1],
'treasury_period_return': self.treasury_period_return,
'algorithm_period_return': self.algorithm_period_returns[-1],
'benchmark_period_return': self.benchmark_period_returns[-1],
'beta': self.beta[-1],
'alpha': self.alpha[-1],
'excess_return': self.excess_returns[-1],
'max_drawdown': self.max_drawdown,
'period_label': period_label
}
rval['sharpe'] = self.sharpe[-1]
rval['sortino'] = self.sortino[-1]
rval['information'] = self.information[-1]
return {k: None
if check_entry(k, v)
else v for k, v in rval.iteritems()}
def __repr__(self):
statements = []
metrics = [
"algorithm_period_returns",
"benchmark_period_returns",
"excess_returns",
"trading_days",
"benchmark_volatility",
"algorithm_volatility",
"sharpe",
"sortino",
"information",
"algorithm_covariance",
"benchmark_variance",
"beta",
"alpha",
"max_drawdown",
"algorithm_returns",
"benchmark_returns",
"condition_number",
"eigen_values"
]
for metric in metrics:
value = getattr(self, metric)
if isinstance(value, list):
if len(value) == 0:
value = np.nan
else:
value = value[-1]
statements.append("{m}:{v}".format(m=metric, v=value))
return '\n'.join(statements)
def update_compounded_log_returns(self):
if len(self.algorithm_returns) == 0:
return
try:
compound = math.log(1 + self.algorithm_returns[
self.algorithm_returns.last_valid_index()])
except ValueError:
compound = 0.0
# BUG? Shouldn't this be set to log(1.0 + 0) ?
if len(self.compounded_log_returns) == 0:
self.compounded_log_returns.append(compound)
else:
self.compounded_log_returns.append(
self.compounded_log_returns[-1] +
compound
)
def calculate_period_returns(self, returns):
returns = np.array(returns)
return (1. + returns).prod() - 1
def update_current_max(self):
if len(self.compounded_log_returns) == 0:
return
if self.current_max < self.compounded_log_returns[-1]:
self.current_max = self.compounded_log_returns[-1]
def calculate_max_drawdown(self):
if len(self.compounded_log_returns) == 0:
return self.max_drawdown
cur_drawdown = 1.0 - math.exp(
self.compounded_log_returns[-1] -
self.current_max)
if self.max_drawdown < cur_drawdown:
return cur_drawdown
else:
return self.max_drawdown
def calculate_sharpe(self):
"""
http://en.wikipedia.org/wiki/Sharpe_ratio
"""
return sharpe_ratio(self.algorithm_volatility[-1],
self.algorithm_period_returns[-1],
self.treasury_period_return)
def calculate_sortino(self, mar=None):
"""
http://en.wikipedia.org/wiki/Sortino_ratio
"""
if mar is None:
mar = self.treasury_period_return
return sortino_ratio(np.array(self.algorithm_returns),
self.algorithm_period_returns[-1],
mar)
def calculate_information(self):
"""
http://en.wikipedia.org/wiki/Information_ratio
"""
A = np.array
return information_ratio(A(self.algorithm_returns),
A(self.benchmark_returns))
def calculate_alpha(self):
"""
http://en.wikipedia.org/wiki/Alpha_(investment)
"""
return alpha(self.algorithm_period_returns[-1],
self.treasury_period_return,
self.benchmark_period_returns[-1],
self.beta[-1])
def calculate_volatility(self, daily_returns):
return np.std(daily_returns, ddof=1) * math.sqrt(self.num_trading_days)
def calculate_beta(self):
"""
.. 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, ddof=1)
eigen_values = la.eigvals(C)
condition_number = max(eigen_values) / min(eigen_values)
algorithm_covariance = C[0][1]
benchmark_variance = C[1][1]
beta = algorithm_covariance / benchmark_variance
return (
beta,
algorithm_covariance,
benchmark_variance,
condition_number,
eigen_values
)
class RiskReport(object):
def __init__(self, algorithm_returns, sim_params, benchmark_returns=None):
"""
algorithm_returns needs to be a list of daily_return objects
sorted in date ascending order
"""
self.algorithm_returns = algorithm_returns
self.sim_params = sim_params
self.benchmark_returns = benchmark_returns
self.created = epoch_now()
if len(self.algorithm_returns) == 0:
start_date = self.sim_params.period_start
end_date = self.sim_params.period_end
else:
# FIXME: Papering over multiple algorithm_return types
if isinstance(self.algorithm_returns, list):
start_date = self.algorithm_returns[0].date
end_date = self.algorithm_returns[-1].date
else:
start_date = self.algorithm_returns.index[0]
end_date = self.algorithm_returns.index[-1]
self.month_periods = self.periods_in_range(1, start_date, end_date)
self.three_month_periods = self.periods_in_range(3, start_date,
end_date)
self.six_month_periods = self.periods_in_range(6, start_date, end_date)
self.year_periods = self.periods_in_range(12, start_date, end_date)
def to_dict(self):
"""
RiskMetrics are calculated for rolling windows in four lengths::
- 1_month
- 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.
See :py:meth:`RiskMetrics.to_dict` for the detailed list of fields
provided for each period.
"""
return {
'one_month': [x.to_dict() for x in self.month_periods],
'three_month': [x.to_dict() for x in self.three_month_periods],
'six_month': [x.to_dict() for x in self.six_month_periods],
'twelve_month': [x.to_dict() for x in self.year_periods],
'created': self.created
}
def periods_in_range(self, months_per, start, end):
one_day = datetime.timedelta(days=1)
ends = []
cur_start = start.replace(day=1)
# in edge cases (all sids filtered out, start/end are adjacent)
# a test will not generate any returns data
if len(self.algorithm_returns) == 0:
return ends
#ensure that we have an end at the end of a calendar month, in case
#the return series ends mid-month...
the_end = end.replace(day=1) + relativedelta(months=1) - one_day
while True:
cur_end = cur_start + relativedelta(months=months_per) - one_day
if(cur_end > the_end):
break
cur_period_metrics = RiskMetricsPeriod(
start_date=cur_start,
end_date=cur_end,
returns=self.algorithm_returns,
benchmark_returns=self.benchmark_returns
)
ends.append(cur_period_metrics)
cur_start = cur_start + relativedelta(months=1)
return ends