# Untitled

unknown
plain_text
10 days ago
18 kB
2
Indexable
Never
```import pandas as pd
import numpy as np
import warnings
import itertools

# Filter out the SettingWithCopyWarning
warnings.filterwarnings('ignore')
import matplotlib.pyplot as plt
from IPython.display import display

def max_dd(returns):
"""Determines the maximum drawdown of a strategy.

Args:
returns : (pandas.Series) Daily returns of the strategy, noncumulative.

Returns:
(float): Maximum drawdown.
"""
r = returns.add(1).cumprod()
dd = r.div(r.cummax()).sub(1)
mdd = dd.min()
end = dd.idxmin()
start = r.loc[:end].idxmax()
return {"Max Drawdown": round(mdd * 100, 2),
"Start": start.strftime("%m/%d/%Y"),
"End": end.strftime("%m/%d/%Y")}

def period_volatility(x, annualization_factor=252):
"""
Determines the volatility of a strategy.

Args:
x : (pandas.Series) Daily returns of the strategy, noncumulative.

annualization_factor : int(default 252 days), factor used for annualization of values (252 if data is daily,
52 if data is weekly,
12 if data is monthly,
4 of data is quarterly)

Returns:
(float) Volatility.
"""
return np.std(x) * (annualization_factor ** 0.5)

def period_downside_volatility(x, annualization_factor=252):
"""
Determines the downside volatility(using only negative returns) of a strategy.

Args:
x : (pandas.Series) Daily returns of the strategy, noncumulative.

annualization_factor : int(default 252 days), factor used for annualization of values (252 if data is daily,
52 if data is weekly,
12 if data is monthly,
4 of data is quarterly)

Returns:
(float) Downside volatility.
"""
return (np.std(x[x < 0]) * (annualization_factor ** 0.5))

def cumulative_return(x):
"""
Calculates the cumulative return of a strategy.

Args:
x : (pandas.Series) Daily returns of the strategy, noncumulative.

Returns:
(float) Cumulative return.
"""
return (np.prod(1 + x) - 1)

def period_return(x,
annualize=True,
annualization_factor=252):
"""
Determines return of a strategy.

Args:
x : (pandas.Series) Daily returns of the strategy, noncumulative.
annualize: boolean, False if annulization is not used, True otherwise
annualization_factor : int(default 252 days), factor used for annualization of values (252 if data is daily,
52 if data is weekly,
12 if data is monthly,
4 of data is quarterly)

Returns:
(float) Return
"""

if annualize:
ann_factor = len(x) / annualization_factor
return ((1 + cumulative_return(x)) ** (1 / ann_factor)) - 1
else:
return cumulative_return(x)

def period_sharpe_ratio(x,
annualize=True,
annualization_factor=252):
"""
Determines Sharpe ratio of a strategy.

Args:
x : (pandas.Series) Daily returns of the strategy, noncumulative.
annualize: boolean, False if annulization is not used, True otherwise
annualization_factor : int(default 252 days), factor used for annualization of values (252 if data is daily,
52 if data is weekly,
12 if data is monthly,
4 of data is quarterly)

Returns:
(float) Sharpe ratio
"""
a = period_return(x, annualize=annualize, annualization_factor=annualization_factor)
b = period_volatility(x, annualization_factor=annualization_factor)
return a / b

def sortino_ratio(x, annualize=True, annualization_factor=252):
"""
Determines Sortino ratio of a strategy.

Args:
x : (pandas.Series) Daily returns of the strategy, noncumulative.
annualize: boolean, False if annulization is not used, True otherwise
annualization_factor : int(default 252 days), factor used for annualization of values (252 if data is daily,
52 if data is weekly,
12 if data is monthly,
4 of data is quarterly)

Returns:
(float) Sortino ratio
"""
a = period_return(x, annualize=annualize, annualization_factor=annualization_factor)
b = period_downside_volatility(x, annualization_factor=annualization_factor)
return a / b

def calmar_ratio(x, annualize=True, annualization_factor=252):
"""
Determines Calmar ratio of a strategy.

Args:
x : (pandas.Series) Daily returns of the strategy, noncumulative.
annualize: boolean, False if annulization is not used, True otherwise
annualization_factor : int(default 252 days), factor used for annualization of values (252 if data is daily,
52 if data is weekly,
12 if data is monthly,
4 of data is quarterly)

Returns:
(float) Calmar ratio
"""
dd = max_dd(x)
dd = dd["Max Drawdown"]
if dd < 0:
a = period_return(x, annualize=annualize, annualization_factor=annualization_factor) * 100
b = abs(dd)
return a / b
else:
return np.nan

####

def iqr_indicator(x):
q3, q1 = np.percentile(x, [75, 25])
iqr = q3 - q1
return iqr

def qcd_indicator(x):
q3, q1 = np.percentile(x, [75, 25])
qcd = (q3 - q1) / (q3 + q1)
return qcd

def cv_indicator(x):
return np.std(x) / np.mean(x)

def generate_simulation_params_list(param_dict):
# Extract the keys and values from the dictionary
param_names = list(param_dict.keys())
param_values = list(param_dict.values())

# Generate all possible combinations of the parameter values
param_combinations = list(itertools.product(*param_values))

# Convert each combination into a dictionary with the corresponding parameter names
full_list = [dict(zip(param_names, combination)) for combination in param_combinations]

return full_list

def calculate_metrics(x):
return pd.Series({
'Sharpe': period_sharpe_ratio(x),
'CAGR': cumulative_return(x) * 100,
'Calmar': calmar_ratio(x),
'Sortino': sortino_ratio(x),
'MaxDD': max_dd(x)['Max Drawdown'],
'Vol': period_volatility(x) * 100
})

def simulation_results(backtest_results, list_params, eop_freq='Y', weekdays=False):
sim_key = '_'.join([str(x) for x in list_params.values()])

backtest_results['positions'] = backtest_results['quantity'].diff()
backtest_results['positions'].fillna(0, inplace=True)

backtest_results['signal_diff'] = backtest_results['signals'].ffill().fillna(0).astype(int).diff()

n_trades = pd.DataFrame(
backtest_results.groupby(pd.Grouper(freq='Y')).apply(lambda x: len(x[(x['signal_diff'] != 0)])))
n_trades.columns = ['trades']
n_trades['year'] = n_trades.index.year.astype(str)
n_trades['metric'] = f"EOY_Trades"
n_trades = n_trades.set_index(['metric', 'year'])[['trades']].T
n_trades.index = [sim_key]
n_trades[('ITD', 'Trades')] = n_trades.sum(axis=1)

backtest_results['buysell'] = np.where(backtest_results['positions'] > 0, 'BUY', 'SELL')
backtest_results['buysell'] = np.where(backtest_results['positions'] == 0, np.nan, backtest_results['buysell'])
backtest_results['return'] = backtest_results['total_nav'].pct_change()

log_returns = np.log(1 + backtest_results[['return']].copy())
log_returns_daily = log_returns.resample('D').sum()
log_returns_daily['return'] = np.exp(log_returns_daily['return']) - 1

if weekdays == True:
returns_daily = log_returns_daily.copy()
else:
returns_daily = log_returns_daily[log_returns_daily.index.dayofweek < 5]

monthly_returns = pd.DataFrame(
returns_daily.groupby(pd.Grouper(freq='1W', label='right')).apply(lambda x: period_return(x['return'],
annualize=False) * 100).rename(
'return'))
monthly_returns.index = pd.to_datetime(monthly_returns.index)
monthly_returns['month_year'] = monthly_returns.index.month_name().str[
:3] + '_' + monthly_returns.index.year.astype(str)
monthly_returns['metric'] = 'Monthly_RET'
monthly_returns = monthly_returns.set_index(['metric', 'month_year'])[['return']].T
monthly_returns.index = [sim_key]

# Apply the function to each group for period defined in config
metrics_table_eop = returns_daily.groupby(pd.Grouper(freq=eop_freq))['return'].apply(calculate_metrics).reset_index()
metrics_table_eop.columns = ['date', 'metric', 'measure']
metrics_table_eop['year'] = metrics_table_eop['date'].dt.year.astype(str) + '_' + metrics_table_eop['date'].dt.month_name().str[
:3]

metrics_table_eop['metric'] = metrics_table_eop['metric'].apply(lambda x: f"EOP_{x}")
metrics_table_eop = metrics_table_eop.set_index(['metric', 'year'])[['measure']].T
metrics_table_eop.index = [sim_key]

# Apply the function to each group
metrics_table_itd = pd.DataFrame(calculate_metrics(returns_daily['return']), columns=[sim_key]).T
metrics_table_itd.columns = pd.MultiIndex.from_product([['ITD'], metrics_table_itd.columns])
final_df = pd.concat([metrics_table_itd, metrics_table_eop, monthly_returns, n_trades], axis=1)

return final_df, returns_daily['return'].rename(sim_key)

def assign_rankings(simulation_df_results, include_itd =True):
ranking_rules = {'Sharpe': False, 'CAGR': False, 'Calmar': False, 'Sortino': False,
'MaxDD': True, 'Vol': True, 'Trades': False, 'HitRatio':False}

ranking_simulations = []

data = simulation_df_results.loc[:, ('Monthly_RET', slice(None))]

def pct_positive(row):
positive_values = row[row > 0].count()
total_values = row.count()
return (positive_values / total_values) * 100

# Applying the function to each row
data['hit_ratio_monret'] = data.apply(pct_positive, axis=1)
filtered_byhitratio = data[data['hit_ratio_monret'] >= 50]
filtered_byhitratio['monret_mean'] = filtered_byhitratio.loc[:, ('Monthly_RET', slice(None))].mean(axis=1)
filtered_by_mean = filtered_byhitratio[filtered_byhitratio['monret_mean'] > 0]
filtered_by_mean['IQR'] = filtered_by_mean.apply(iqr_indicator, axis=1).rank(ascending=True)
filtered_by_mean['QCD'] = filtered_by_mean.apply(qcd_indicator, axis=1).rank(ascending=True)
filtered_by_mean['CV'] = filtered_by_mean.apply(cv_indicator, axis=1).rank(ascending=True)
ranking_simulations.append(filtered_by_mean['IQR'])
ranking_simulations.append(filtered_by_mean['QCD'])
ranking_simulations.append(filtered_by_mean['CV'])

if include_itd:
for c in simulation_df_results.loc[filtered_by_mean.index, ('ITD', slice(None))].columns:
df_temp = simulation_df_results.loc[filtered_by_mean.index, c]
df_temp = df_temp.rank(ascending=ranking_rules[c[-1]])

print(df_temp.columns)

ranking_simulations.append(df_temp.rename(columns = {c:f"ITD_{c[-1]}"}))

cols_eoy = ['EOP_Sharpe', 'EOP_Vol', 'EOP_CAGR', 'EOP_Calmar', 'EOP_Sortino']
for c in cols_eoy:
name = c.split('_')[-1]
df_temp = simulation_df_results.loc[filtered_by_mean.index, (c, slice(None))]
df_temp[f'IQR_{name}'] = df_temp.apply(iqr_indicator, axis=1) * (1 if ranking_rules[name] else -1)
df_temp[f'Mean_{name}'] = df_temp.mean(axis=1)
df_temp[f"Rank_{name}"] = df_temp[[f'IQR_{name}', f'Mean_{name}']].apply(tuple, axis=1) \
.rank(method='dense', ascending=ranking_rules[name]).astype(int)

ranking_simulations.append(df_temp[f"Rank_{name}"])

df_temp = pd.DataFrame(filtered_by_mean.loc[:, 'hit_ratio_monret'])
df_temp['Rank_Hitratio']= df_temp['hit_ratio_monret'].rank(method='dense', ascending=ranking_rules['HitRatio']).astype(int)
ranking_simulations.append(df_temp["Rank_Hitratio"])

final_ranks=pd.concat(ranking_simulations, axis=1)
print(final_ranks)
#cols_to_rank =['IQR', 'QCD', 'CV', 'ITD_Sharpe', 'ITD_CAGR', 'ITD_Calmar', 'ITD_Sortino', 'ITD_MaxDD', 'ITD_Vol', 'ITD_Trades', 'Rank_Hitratio']
cols_to_rank = [
('ITD', 'Sharpe'), ('ITD', 'CAGR'), ('ITD', 'Calmar'), ('ITD', 'Sortino'),
('ITD', 'MaxDD'), ('ITD', 'Vol'), ('ITD', 'Trades'), 'IQR', 'QCD', 'CV', 'Rank_Hitratio'
]
final_ranks['Final_Rank'] = final_ranks[cols_to_rank].sum(axis=1)

return final_ranks

def selected_by_corr(allrets, top_N, negative_only=False):

if negative_only:
# Filter for negative returns only
allrets = allrets.applymap(lambda x: x if x < 0 else 0)

cor_matrix = allrets.corr()
mean_cor = cor_matrix.mean()
selected_models = mean_cor.nsmallest(top_N).index
return selected_models

def selected_by_cov(allrets, top_N, negative_only=False):
if negative_only:
# Filter for negative returns only
allrets = allrets.applymap(lambda x: x if x < 0 else 0)

cov_matrix = allrets.cov()
mean_cov = cov_matrix.mean()
selected_models = mean_cov.nsmallest(top_N).index
return selected_models

def equal_weight_ensemble(df):
num_models= len(df.columns)
weights = 1 / num_models
equally_weighted_returns = df.apply(lambda x: x * weights).sum(axis=1)
return equally_weighted_returns

def decorralated_portfolio(returns_all,rankings,top_rank=10,bottom_corr=3):
decorralated_models = selected_by_corr(returns_all[rankings.sort_values('Final_Rank').head(top_rank).index], bottom_corr)
return decorralated_models

def performance(
returns,
annualize=True,
annualization_factor=252,
color="brown",
label="LONG",
verbose=True,
plot=True):
"""Calculates basic performance measures on series of returns(Return, Volatility, Sharpe).

Args:
returns: (pandas Series), Daily(period) noncumulative returns of the strategy
annualize: boolean,False if annulization is not used, True otherwise
annualization_factor : int(default 252 days), factor used for annualization of values (252 if data is daily,
52 if data is weekly,
12 if data is monthly,
4 of data is quarterly)
color: the color used for the plot and printing if verbose=True
label: the name for the strategy in returns
verbose: (boolean) True if you want to print the output
plot: (boolean) True if you want to plot cumulative return of the strategy (returns)

Returns:
pandas DataFrame, plot or print of the calculated measures
"""

Volatility = np.round(period_volatility(returns, annualization_factor=annualization_factor) * 100, 2)
Return = np.round(period_return(returns, annualize=annualize, annualization_factor=annualization_factor) * 100, 2)
Sharpe = np.round(period_sharpe_ratio(returns, annualize=annualize, annualization_factor=annualization_factor), 2)
Sortino = np.round(sortino_ratio(returns, annualize=annualize, annualization_factor=annualization_factor), 2)
Calmar = np.round(calmar_ratio(returns, annualize=annualize, annualization_factor=annualization_factor), 2)

max_drawdown = max_dd(returns)
Max_DD = max_drawdown["Max Drawdown"]
start = max_drawdown["Start"]
end = max_drawdown["End"]

stat = {
"Return": [Return],
"Annualized Volatility": [Volatility],
"Sharpe": [Sharpe],
"Sortino ratio": [Sortino],
"Calmar ratio": [Calmar],
"Max Draw-Down": [Max_DD],
"MDD Start": [start],
"MDD End": [end]
}

df_output = pd.DataFrame(stat)
df_to_print = df_output.rename(index={0: label})

if verbose:
pd.options.display.notebook_repr_html = True

def color_b(s):
return np.where(s == label, f"background-color: {color};", "")

display(df_to_print.style.apply_index(color_b))

if plot:
start = returns.index.min()
ptf_cum = pd.Series([1], index=[start])
ptf_cum = pd.concat([ptf_cum, np.cumprod(returns[returns.index > start] + 1)])
ptf_cum.plot(figsize=(12, 5), color=color, label=label)
if label.startswith('_'):
pass
else:
plt.legend()
return df_to_print[["Return", "Annualized Volatility", "Sharpe",
"Sortino ratio", "Calmar ratio",
"Max Draw-Down"]]```
Leave a Comment