Untitled
unknown
plain_text
a month ago
12 kB
5
Indexable
from itertools import product, combinations
import pandas as pd
import numpy as np
def generate_internal_rebalance_ratios(n, ratio_type='default'):
"""
Generate an internal rebalance ratio list of length n.
All ratios are manually defined for n from 2 to 10.
Parameters:
n (int): Number of assets (2 to 10).
ratio_type (str): 'default', 'equal', 'gradual', or 'top_heavy'.
Returns:
list: Internal rebalance ratios.
"""
if n < 2 or n > 10:
raise ValueError("Number of assets must be between 2 and 10.")
ratios = {
'default': {
4: [0.7, 0.3, -0.3, -0.7],
5: [0.7, 0.3, 0, -0.3, -0.7],
6: [0.7, 0.2, 0.1, -0.1, -0.2, -0.7],
7: [0.7, 0.2, 0.1, 0, -0.1, -0.2, -0.7],
8: [0.7, 0.2, 0.1, 0, 0, -0.1, -0.2, -0.7],
9: [0.7, 0.2, 0.1, 0, 0, 0, -0.1, -0.2, -0.7],
10: [0.7, 0.2, 0.1, 0, 0, 0, 0, -0.1, -0.2, -0.7]
},
'equal': {
4: [0.5, 0.5, -0.5, -0.5],
5: [0.5, 0.5, 0, -0.5, -0.5],
6: [0.4, 0.3, 0.3, -0.3, -0.3, -0.4],
7: [0.4, 0.3, 0.3, 0, -0.3, -0.3, -0.4],
8: [0.25, 0.25, 0.25, 0.25, -00.25, -0.25, -0.25, -0.25],
9: [0.25, 0.25, 0.25, 0.25, 0, -00.25, -0.25, -0.25, -0.25],
10: [0.2, 0.2, 0.2, 0.2, 0.2, -0.2, -0.2, -0.2, -0.2, -0.2]
},
'gradual': {
4: [0.6, 0.4, -0.4, -0.6],
5: [0.6, 0.4, 0, -0.4, -0.6],
6: [0.6, 0.2, 0.2, -0.2, -0.2, -0.6],
7: [0.6, 0.2, 0.2, 0, -0.2, -0.2, -0.6],
8: [0.5, 0.3, 0.2, 0, 0, -0.2, -0.3, -0.5],
9: [0.4, 0.3, 0.2, 0.1, 0, -0.1, -0.2, -0.3, -0.4],
10: [0.4, 0.3, 0.2, 0.1, 0, -0, -0.1, -0.2, -0.3, -0.4]
},
'top_heavy': {
4: [0.8, 0.2, -0.2, -0.8],
5: [0.7, 0.3, 0, -0.3, -0.7],
6: [0.7, 0.3, 0, 0, -0.3, -0.7],
7: [0.7, 0.3, 0, 0, 0, -0.3, -0.7],
8: [0.7, 0.3, 0.0, 0, 0, 0, -0.3, -0.7],
9: [0.7, 0.3, 0, 0, 0, 0, 0, -0.3, -0.7],
10: [0.6, 0.4, 0, 0, 0, 0, 0, 0, -0.4, -0.6]
}
}
if ratio_type not in ratios:
raise ValueError("Invalid ratio_type. Choose from 'default', 'equal', 'gradual', or 'top_heavy'.")
return ratios[ratio_type][n]
# Load the price data once.
filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\Codes\Historic Prices.xlsx"
prices = load_price_data(filepath)
# Load the risk free rate data.
risk_free_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\Codes\Risk Free Rate.xlsx"
risk_free_df = pd.read_excel(risk_free_filepath, sheet_name="Risk Free Rate", parse_dates=[0])
risk_free_df.columns = ['Date', '1m', '3m']
# Define the asset tickers.
base_tickers = ['SPY US Equity', 'TLT US Equity', 'GLD US Equity', 'SHV US Equity']
additional_tickers = ['EEM US Equity', 'QQQ US Equity', 'LQD US Equity', 'HYG US Equity', 'VNQ US Equity', 'DBC US Equity']
# Define simulation parameters.
initial_aum = 100e6
start_date = pd.to_datetime('2008-01-01')
end_date = pd.to_datetime('2025-02-01')
# Parameter grid.
metric_types = ['simple']
lookback_periods = [1, 3, 6, 9, 12]
rebalance_ratios = [0.05, 0.1, 0.2]
rebalance_periods = [1, 2, 3]
ratio_types = ['default', 'equal', 'gradual', "top_heavy"]
# Store results.
results = []
# Generate all possible combinations of parameters.
combinations_params = list(product(
metric_types, lookback_periods, rebalance_ratios, rebalance_periods, ratio_types
))
def calculate_correlations(prices, tickers):
returns = prices[tickers].pct_change().dropna()
return returns.corr()
def generate_valid_combinations(base, additional, prices, max_additional=6):
valid_combinations = [base] # Start with the base combination.
correlations = calculate_correlations(prices, base + additional)
for i in range(1, max_additional + 1):
for add in combinations(additional, i):
candidate = base + list(add)
if len(candidate) <= 5 or len(candidate) == 10 or len(candidate) == 9:
valid_combinations.append(candidate)
elif 6 <= len(candidate) <= 8:
is_valid = True
for additional_asset in add:
for base_asset in base:
if correlations.loc[additional_asset, base_asset] > 0.65:
is_valid = False
break
if not is_valid:
break
if is_valid:
valid_combinations.append(candidate)
return valid_combinations
# Generate valid ticker combinations.
ticker_combinations = generate_valid_combinations(base_tickers, additional_tickers, prices)
# Define a function to calculate Sharpe ratio.
def calculate_sharpe_ratio(returns, period):
volatility = returns.std()
avg_return = returns.mean()
# Map rebalance period to an annualization factor.
annualization_factor = {1: 12, 2: 6, 3: 4}.get(period, np.nan)
return (avg_return / volatility) * (annualization_factor ** 0.5) if volatility > 0 else np.nan
# Iterate over valid ticker combinations and parameter grid.
for ticker_subset in ticker_combinations:
for (metric_type, lookback_period, rebalance_ratio, rebalance_period, ratio_type) in combinations_params:
try:
# Generate the internal rebalance ratios based on number of tickers and ratio type.
internal_rebalance = generate_internal_rebalance_ratios(len(ticker_subset), ratio_type)
# Run the simulation with the selected tickers.
result_df_intermediate = simulate_strategy(
prices, ticker_subset, [], [], initial_aum, start_date, end_date,
rebalance_period, rebalance_ratio, lookback_period, metric_type, internal_rebalance
)
# Ensure that there is a 'Date' column (if not, reset the index).
if 'Date' not in result_df_intermediate.columns:
result_df_intermediate = result_df_intermediate.reset_index().rename(columns={'index': 'Date'})
# Merge risk free rate data on the 'Date' column.
merged_df = pd.merge(result_df_intermediate, risk_free_df, on='Date', how='left')
# Adjust risk free rate based on the rebalance period.
if rebalance_period == 1:
merged_df['rf_adjust'] = merged_df['1m'] / 12
elif rebalance_period == 2:
merged_df['rf_adjust'] = merged_df['1m'] / 6
elif rebalance_period == 3:
merged_df['rf_adjust'] = merged_df['1m'] / 4
else:
merged_df['rf_adjust'] = 0
# Compute excess returns by subtracting the risk free rate from monthly returns.
merged_df['Excess Return'] = merged_df['Return'] - (merged_df['rf_adjust']/100)
# Calculate performance metrics using excess returns.
final_aum = merged_df['Portfolio Value'].iloc[-1]
cumulative_return = (final_aum / initial_aum) - 1
num_months = (end_date.year - start_date.year) * 12 + (end_date.month - start_date.month)
annualized_return = (1 + cumulative_return) ** (12 / num_months) - 1
annualization_factor = {1: 12, 2: 6, 3: 4}.get(rebalance_period, 12) # Default to 12 if not found
excess_returns = merged_df['Excess Return'].dropna()
avg_monthly_excess_return = excess_returns.mean()
monthly_excess_volatility = excess_returns.std()
sharpe_ratio = calculate_sharpe_ratio(excess_returns, rebalance_period)
annualized_volatility = monthly_excess_volatility * (annualization_factor ** 0.5)
annualised_risk_premium = avg_monthly_excess_return*annualization_factor
result_df_intermediate = result_df_intermediate.set_index("Date")
# Extract year from the 'Date' column
returns = result_df_intermediate['Return']
def compute_max_drawdown(returns):
"""Compute the maximum drawdown from a series of returns."""
cumulative = (1 + returns).cumprod()
running_max = cumulative.cummax()
drawdown = (cumulative - running_max) / running_max
return drawdown.min()
max_drawdown = compute_max_drawdown(returns)
# Calculate minimum and maximum annual returns
annual_returns = returns.groupby(returns.index.year).apply(lambda x: (1 + x).prod() - 1)
min_annual_return = annual_returns.min()
max_annual_return = annual_returns.max()
# Store the results.
results.append({
'Ticker Subset': tuple(ticker_subset),
'Num Assets': len(ticker_subset),
'Internal Rebalance Type': ratio_type,
'Internal Rebalance': internal_rebalance,
'Rebalance Ratio': rebalance_ratio,
'Rebalance Period': rebalance_period,
'Lookback Period': lookback_period,
'Annualized Return': annualized_return,
'Annualised risk Premium': annualised_risk_premium,
'Annualised excess Volatility': annualized_volatility,
'Sharpe Ratio': sharpe_ratio,
'Minimum Annual Return' : min_annual_return,
'Maximum Annual Return' : max_annual_return,
'Maximum Drawdown' : max_drawdown
})
except ValueError as e:
print(f"Skipping combination due to error: {e}")
continue
# Create the results DataFrame.
summary_df = pd.DataFrame(results)
# Sort by Sharpe Ratio (Descending).
summary_df = summary_df.sort_values(by='Sharpe Ratio', ascending=False).reset_index(drop=True)
# Display the final results.
pd.set_option('display.float_format', lambda x: f'{x:,.2f}')
pd.set_option('display.max_columns', None)
pd.set_option('display.width', None)
print(summary_df)
summary_df.to_excel(r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\summary_df_startin_2007_final_v3.xlsx")
summary_df = pd.read_excel(r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\summary_df_startin_2007_final_v3.xlsx")
summary_df = summary_df.copy()
summary_df['Best_Sharpe'] = summary_df.groupby(['Num Assets'])['Sharpe Ratio'].transform('max') == summary_df['Sharpe Ratio']
# Flag rows with best Annualized Return in each group
summary_df['Best_Annualized'] = summary_df.groupby(['Num Assets'])['Maximum Drawdown'].transform('max') == summary_df['Maximum Drawdown']
# summary_df['Best_Volatility'] = summary_df.groupby(['Num Assets', 'Rebalance Period', 'Lookback Period', 'Internal Rebalance'])['Monthly Volatility'].transform('min') == summary_df['Monthly Volatility']
# Optionally, combine them into one flag if you prefer (i.e. highlight if a row is best in either)
summary_df['Optimal'] = summary_df['Best_Sharpe'] & summary_df['Best_Annualized']
summary_df.to_excel(r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\summary_df_startin_2007_final_v3.xlsx")Editor is loading...
Leave a Comment