mirror of
https://github.com/webclinic017/drift.git
synced 2026-08-16 20:38:07 +00:00
feat(Metrics): added probabilistic sharpe ratio (#82)
* feat(Metrics): added probabilistic sharpe ratio * Apply suggestions from code review
This commit is contained in:
Mark Aron Szulyovszky
@@ -0,0 +1,336 @@
|
||||
import warnings
|
||||
import pandas as pd
|
||||
import scipy.stats as ss
|
||||
import numpy as np
|
||||
|
||||
def timing_of_flattening_and_flips(target_positions: pd.Series) -> pd.DatetimeIndex:
|
||||
"""
|
||||
Advances in Financial Machine Learning, Snippet 14.1, page 197
|
||||
Derives the timestamps of flattening or flipping trades from a pandas series
|
||||
of target positions. Can be used for position changes analysis, such as
|
||||
frequency and balance of position changes.
|
||||
Flattenings - times when open position is bing closed (final target position is 0).
|
||||
Flips - times when positive position is reversed to negative and vice versa.
|
||||
:param target_positions: (pd.Series) Target position series with timestamps as indices
|
||||
:return: (pd.DatetimeIndex) Timestamps of trades flattening, flipping and last bet
|
||||
"""
|
||||
|
||||
empty_positions = target_positions[(target_positions == 0)].index # Empty positions index
|
||||
previous_positions = target_positions.shift(1) # Timestamps pointing at previous positions
|
||||
|
||||
# Index of positions where previous one wasn't empty
|
||||
previous_positions = previous_positions[(previous_positions != 0)].index
|
||||
|
||||
# FLATTENING - if previous position was open, but current is empty
|
||||
flattening = empty_positions.intersection(previous_positions)
|
||||
|
||||
# Multiplies current position with value of next one
|
||||
multiplied_posions = target_positions.iloc[1:] * target_positions.iloc[:-1].values
|
||||
|
||||
# FLIPS - if current position has another direction compared to the next
|
||||
flips = multiplied_posions[(multiplied_posions < 0)].index
|
||||
flips_and_flattenings = flattening.union(flips).sort_values()
|
||||
if target_positions.index[-1] not in flips_and_flattenings: # Appending with last bet
|
||||
flips_and_flattenings = flips_and_flattenings.append(target_positions.index[-1:])
|
||||
|
||||
return flips_and_flattenings
|
||||
|
||||
|
||||
def average_holding_period(target_positions: pd.Series) -> float:
|
||||
"""
|
||||
Advances in Financial Machine Learning, Snippet 14.2, page 197
|
||||
Estimates the average holding period (in days) of a strategy, given a pandas series
|
||||
of target positions using average entry time pairing algorithm.
|
||||
Idea of an algorithm:
|
||||
* entry_time = (previous_time * weight_of_previous_position + time_since_beginning_of_trade * increase_in_position )
|
||||
/ weight_of_current_position
|
||||
* holding_period ['holding_time' = time a position was held, 'weight' = weight of position closed]
|
||||
* res = weighted average time a trade was held
|
||||
:param target_positions: (pd.Series) Target position series with timestamps as indices
|
||||
:return: (float) Estimated average holding period, NaN if zero or unpredicted
|
||||
"""
|
||||
|
||||
holding_period = pd.DataFrame(columns=['holding_time', 'weight'])
|
||||
entry_time = 0
|
||||
position_difference = target_positions.diff()
|
||||
|
||||
# Time elapsed from the starting time for each position
|
||||
time_difference = (target_positions.index - target_positions.index[0]) / np.timedelta64(1, 'D')
|
||||
for i in range(1, target_positions.size):
|
||||
|
||||
# Increased or unchanged position
|
||||
if float(position_difference.iloc[i] * target_positions.iloc[i - 1]) >= 0:
|
||||
if float(target_positions.iloc[i]) != 0: # And not an empty position
|
||||
entry_time = (entry_time * target_positions.iloc[i - 1] +
|
||||
time_difference[i] * position_difference.iloc[i]) / target_positions.iloc[i]
|
||||
|
||||
# Decreased
|
||||
if float(position_difference.iloc[i] * target_positions.iloc[i - 1]) < 0:
|
||||
hold_time = time_difference[i] - entry_time
|
||||
|
||||
# Flip of a position
|
||||
if float(target_positions.iloc[i] * target_positions.iloc[i - 1]) < 0:
|
||||
weight = abs(target_positions.iloc[i - 1])
|
||||
holding_period.loc[target_positions.index[i], ['holding_time', 'weight']] = (hold_time, weight)
|
||||
entry_time = time_difference[i] # Reset entry time
|
||||
|
||||
# Only a part of position is closed
|
||||
else:
|
||||
weight = abs(position_difference.iloc[i])
|
||||
holding_period.loc[target_positions.index[i], ['holding_time', 'weight']] = (hold_time, weight)
|
||||
|
||||
if float(holding_period['weight'].sum()) > 0: # If there were closed trades at all
|
||||
avg_holding_period = float((holding_period['holding_time'] * \
|
||||
holding_period['weight']).sum() / holding_period['weight'].sum())
|
||||
else:
|
||||
avg_holding_period = float('nan')
|
||||
|
||||
return avg_holding_period
|
||||
|
||||
|
||||
def bets_concentration(returns: pd.Series) -> float:
|
||||
"""
|
||||
Advances in Financial Machine Learning, Snippet 14.3, page 201
|
||||
Derives the concentration of returns from given pd.Series of returns.
|
||||
Algorithm is based on Herfindahl-Hirschman Index where return weights
|
||||
are taken as an input.
|
||||
:param returns: (pd.Series) Returns from bets
|
||||
:return: (float) Concentration of returns (nan if less than 3 returns)
|
||||
"""
|
||||
|
||||
if returns.size <= 2:
|
||||
return float('nan') # If less than 3 bets
|
||||
weights = returns / returns.sum() # Weights of each bet
|
||||
hhi = (weights ** 2).sum() # Herfindahl-Hirschman Index for weights
|
||||
hhi = float((hhi - returns.size ** (-1)) / (1 - returns.size ** (-1)))
|
||||
|
||||
return hhi
|
||||
|
||||
|
||||
def all_bets_concentration(returns: pd.Series, frequency: str = 'M') -> tuple:
|
||||
"""
|
||||
Advances in Financial Machine Learning, Snippet 14.3, page 201
|
||||
Given a pd.Series of returns, derives concentration of positive returns, negative returns
|
||||
and concentration of bets grouped by time intervals (daily, monthly etc.).
|
||||
If after time grouping less than 3 observations, returns nan.
|
||||
Properties or results:
|
||||
* low positive_concentration ⇒ no right fat-tail of returns (desirable)
|
||||
* low negative_concentration ⇒ no left fat-tail of returns (desirable)
|
||||
* low time_concentration ⇒ bets are not concentrated in time, or are evenly concentrated (desirable)
|
||||
* positive_concentration == 0 ⇔ returns are uniform
|
||||
* positive_concentration == 1 ⇔ only one non-zero return exists
|
||||
:param returns: (pd.Series) Returns from bets
|
||||
:param frequency: (str) Desired time grouping frequency from pd.Grouper
|
||||
:return: (tuple of floats) Concentration of positive, negative and time grouped concentrations
|
||||
"""
|
||||
|
||||
# Concentration of positive returns per bet
|
||||
positive_concentration = bets_concentration(returns[returns >= 0])
|
||||
|
||||
# Concentration of negative returns per bet
|
||||
negative_concentration = bets_concentration(returns[returns < 0])
|
||||
|
||||
# Concentration of bets/time period (month by default)
|
||||
time_concentration = bets_concentration(returns.groupby(pd.Grouper(freq=frequency)).count())
|
||||
|
||||
return (positive_concentration, negative_concentration, time_concentration)
|
||||
|
||||
|
||||
def drawdown_and_time_under_water(returns: pd.Series, dollars: bool = False) -> tuple:
|
||||
"""
|
||||
Advances in Financial Machine Learning, Snippet 14.4, page 201
|
||||
Calculates drawdowns and time under water for pd.Series of either relative price of a
|
||||
portfolio or dollar price of a portfolio.
|
||||
Intuitively, a drawdown is the maximum loss suffered by an investment between two consecutive high-watermarks.
|
||||
The time under water is the time elapsed between an high watermark and the moment the PnL (profit and loss)
|
||||
exceeds the previous maximum PnL. We also append the Time under water series with period from the last
|
||||
high-watermark to the last return observed.
|
||||
Return details:
|
||||
* Drawdown series index is the time of a high watermark and the value of a
|
||||
drawdown after it.
|
||||
* Time under water index is the time of a high watermark and how much time
|
||||
passed till the next high watermark in years. Also includes time between
|
||||
the last high watermark and last observation in returns as the last element.
|
||||
:param returns: (pd.Series) Returns from bets
|
||||
:param dollars: (bool) Flag if given dollar performance and not returns.
|
||||
If dollars, then drawdowns are in dollars, else as a %.
|
||||
:return: (tuple of pd.Series) Series of drawdowns and time under water
|
||||
"""
|
||||
|
||||
frame = returns.to_frame('pnl')
|
||||
frame['hwm'] = returns.expanding().max() # Adding high watermarks as column
|
||||
|
||||
# Grouped as min returns by high watermarks
|
||||
high_watermarks = frame.groupby('hwm').min().reset_index()
|
||||
high_watermarks.columns = ['hwm', 'min']
|
||||
|
||||
# Time high watermark occurred
|
||||
high_watermarks.index = frame['hwm'].drop_duplicates(keep='first').index
|
||||
|
||||
# Picking ones that had a drawdown after high watermark
|
||||
high_watermarks = high_watermarks[high_watermarks['hwm'] > high_watermarks['min']]
|
||||
if dollars:
|
||||
drawdown = high_watermarks['hwm'] - high_watermarks['min']
|
||||
else:
|
||||
drawdown = 1 - high_watermarks['min'] / high_watermarks['hwm']
|
||||
|
||||
time_under_water = ((high_watermarks.index[1:] - high_watermarks.index[:-1]) / np.timedelta64(1, 'Y')).values
|
||||
|
||||
# Adding also period from last High watermark to last return observed.
|
||||
time_under_water = np.append(time_under_water,
|
||||
(returns.index[-1] - high_watermarks.index[-1]) / np.timedelta64(1, 'Y'))
|
||||
|
||||
time_under_water = pd.Series(time_under_water, index=high_watermarks.index)
|
||||
|
||||
return drawdown, time_under_water
|
||||
|
||||
|
||||
def sharpe_ratio(returns: pd.Series, entries_per_year: int = 252, risk_free_rate: float = 0) -> float:
|
||||
"""
|
||||
Calculates annualized Sharpe ratio for pd.Series of normal or log returns.
|
||||
Risk_free_rate should be given for the same period the returns are given.
|
||||
For example, if the input returns are observed in 3 months, the risk-free
|
||||
rate given should be the 3-month risk-free rate.
|
||||
:param returns: (pd.Series) Returns - normal or log
|
||||
:param entries_per_year: (int) Times returns are recorded per year (252 by default)
|
||||
:param risk_free_rate: (float) Risk-free rate (0 by default)
|
||||
:return: (float) Annualized Sharpe ratio
|
||||
"""
|
||||
|
||||
sharpe_r = (returns.mean() - risk_free_rate) / returns.std() * (entries_per_year) ** (1 / 2)
|
||||
|
||||
return sharpe_r
|
||||
|
||||
|
||||
def information_ratio(returns: pd.Series, benchmark: float = 0, entries_per_year: int = 252) -> float:
|
||||
"""
|
||||
Calculates annualized information ratio for pd.Series of normal or log returns.
|
||||
Benchmark should be provided as a return for the same time period as that between
|
||||
input returns. For example, for the daily observations it should be the
|
||||
benchmark of daily returns.
|
||||
It is the annualized ratio between the average excess return and the tracking error.
|
||||
The excess return is measured as the portfolio’s return in excess of the benchmark’s
|
||||
return. The tracking error is estimated as the standard deviation of the excess returns.
|
||||
:param returns: (pd.Series) Returns - normal or log
|
||||
:param benchmark: (float) Benchmark for performance comparison (0 by default)
|
||||
:param entries_per_year: (int) Times returns are recorded per year (252 by default)
|
||||
:return: (float) Annualized information ratio
|
||||
"""
|
||||
|
||||
excess_returns = returns - benchmark
|
||||
information_r = sharpe_ratio(excess_returns, entries_per_year)
|
||||
|
||||
return information_r
|
||||
|
||||
|
||||
def probabilistic_sharpe_ratio(observed_sr: float, benchmark_sr: float, number_of_returns: int,
|
||||
skewness_of_returns: float = 0, kurtosis_of_returns: float = 3) -> float:
|
||||
"""
|
||||
Calculates the probabilistic Sharpe ratio (PSR) that provides an adjusted estimate of SR,
|
||||
by removing the inflationary effect caused by short series with skewed and/or
|
||||
fat-tailed returns.
|
||||
Given a user-defined benchmark Sharpe ratio and an observed Sharpe ratio,
|
||||
PSR estimates the probability that SR ̂is greater than a hypothetical SR.
|
||||
- It should exceed 0.95, for the standard significance level of 5%.
|
||||
- It can be computed on absolute or relative returns.
|
||||
:param observed_sr: (float) Sharpe ratio that is observed
|
||||
:param benchmark_sr: (float) Sharpe ratio to which observed_SR is tested against
|
||||
:param number_of_returns: (int) Times returns are recorded for observed_SR
|
||||
:param skewness_of_returns: (float) Skewness of returns (0 by default)
|
||||
:param kurtosis_of_returns: (float) Kurtosis of returns (3 by default)
|
||||
:return: (float) Probabilistic Sharpe ratio
|
||||
"""
|
||||
|
||||
test_value = ((observed_sr - benchmark_sr) * np.sqrt(number_of_returns - 1)) / \
|
||||
((1 - skewness_of_returns * observed_sr + (kurtosis_of_returns - 1) / \
|
||||
4 * observed_sr ** 2)**(1 / 2))
|
||||
|
||||
if np.isnan(test_value):
|
||||
warnings.warn('Test value is nan. Please check the input values.', UserWarning)
|
||||
return test_value
|
||||
|
||||
if isinstance(test_value, complex):
|
||||
warnings.warn('Output is a complex number. You may want to check the input skewness (too high), '
|
||||
'kurtosis (too low), or observed_sr values.', UserWarning)
|
||||
|
||||
if np.isinf(test_value):
|
||||
warnings.warn('Test value is infinite. You may want to check the input skewness, '
|
||||
'kurtosis, or observed_sr values.', UserWarning)
|
||||
|
||||
probab_sr = ss.norm.cdf(test_value)
|
||||
|
||||
return probab_sr
|
||||
|
||||
|
||||
def deflated_sharpe_ratio(observed_sr: float, sr_estimates: list, number_of_returns: int,
|
||||
skewness_of_returns: float = 0, kurtosis_of_returns: float = 3,
|
||||
estimates_param: bool = False, benchmark_out: bool = False) -> float:
|
||||
"""
|
||||
Calculates the deflated Sharpe ratio (DSR) - a PSR where the rejection threshold is
|
||||
adjusted to reflect the multiplicity of trials. DSR is estimated as PSR[SR∗], where
|
||||
the benchmark Sharpe ratio, SR∗, is no longer user-defined, but calculated from
|
||||
SR estimate trails.
|
||||
DSR corrects SR for inflationary effects caused by non-Normal returns, track record
|
||||
length, and multiple testing/selection bias.
|
||||
- It should exceed 0.95, for the standard significance level of 5%.
|
||||
- It can be computed on absolute or relative returns.
|
||||
Function allows the calculated SR benchmark output and usage of only
|
||||
standard deviation and number of SR trails instead of full list of trails.
|
||||
:param observed_sr: (float) Sharpe ratio that is being tested
|
||||
:param sr_estimates: (list) Sharpe ratios estimates trials list or
|
||||
properties list: [Standard deviation of estimates, Number of estimates]
|
||||
if estimates_param flag is set to True.
|
||||
:param number_of_returns: (int) Times returns are recorded for observed_SR
|
||||
:param skewness_of_returns: (float) Skewness of returns (0 by default)
|
||||
:param kurtosis_of_returns: (float) Kurtosis of returns (3 by default)
|
||||
:param estimates_param: (bool) Flag to use properties of estimates instead of full list
|
||||
:param benchmark_out: (bool) Flag to output the calculated benchmark instead of DSR
|
||||
:return: (float) Deflated Sharpe ratio or Benchmark SR (if benchmark_out)
|
||||
"""
|
||||
|
||||
# Calculating benchmark_SR from the parameters of estimates
|
||||
if estimates_param:
|
||||
benchmark_sr = sr_estimates[0] * \
|
||||
((1 - np.euler_gamma) * ss.norm.ppf(1 - 1 / sr_estimates[1]) +
|
||||
np.euler_gamma * ss.norm.ppf(1 - 1 / sr_estimates[1] * np.e ** (-1)))
|
||||
|
||||
# Calculating benchmark_SR from a list of estimates
|
||||
else:
|
||||
benchmark_sr = np.array(sr_estimates).std() * \
|
||||
((1 - np.euler_gamma) * ss.norm.ppf(1 - 1 / len(sr_estimates)) +
|
||||
np.euler_gamma * ss.norm.ppf(1 - 1 / len(sr_estimates) * np.e ** (-1)))
|
||||
|
||||
deflated_sr = probabilistic_sharpe_ratio(observed_sr, benchmark_sr, number_of_returns,
|
||||
skewness_of_returns, kurtosis_of_returns)
|
||||
|
||||
if benchmark_out:
|
||||
return benchmark_sr
|
||||
|
||||
return deflated_sr
|
||||
|
||||
|
||||
def minimum_track_record_length(observed_sr: float, benchmark_sr: float,
|
||||
skewness_of_returns: float = 0,
|
||||
kurtosis_of_returns: float = 3,
|
||||
alpha: float = 0.05) -> float:
|
||||
"""
|
||||
Calculates the minimum track record length (MinTRL) - "How long should a track
|
||||
record be in order to have statistical confidence that its Sharpe ratio is above
|
||||
a given threshold?”
|
||||
If a track record is shorter than MinTRL, we do not have enough confidence
|
||||
that the observed Sharpe ratio ̂is above the designated Sharpe ratio threshold.
|
||||
MinTRLis expressed in terms of number of observations, not annual or calendar terms.
|
||||
:param observed_sr: (float) Sharpe ratio that is being tested
|
||||
:param benchmark_sr: (float) Sharpe ratio to which observed_SR is tested against
|
||||
:param number_of_returns: (int) Times returns are recorded for observed_SR
|
||||
:param skewness_of_returns: (float) Skewness of returns (0 by default)
|
||||
:param kurtosis_of_returns: (float) Kurtosis of returns (3 by default)
|
||||
:param alpha: (float) Desired significance level (0.05 by default)
|
||||
:return: (float) Minimum number of track records
|
||||
"""
|
||||
|
||||
track_rec_length = 1 + (1 - skewness_of_returns * observed_sr +
|
||||
(kurtosis_of_returns - 1) / 4 * observed_sr ** 2) * \
|
||||
(ss.norm.ppf(1 - alpha) / (observed_sr - benchmark_sr)) ** (2)
|
||||
|
||||
return track_rec_length
|
||||
Reference in New Issue
Block a user