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