Untitled
unknown
plain_text
a month ago
19 kB
2
Indexable
import pandas as pd
import numpy as np
from datetime import datetime
from dateutil.relativedelta import relativedelta
from hmmlearn.hmm import GaussianHMM
###############################################
# 1. Data Loading
###############################################
def load_price_data(filepath):
"""
Load historical prices from an Excel file.
Assumes that the first column is dates and the remaining columns are tickers.
"""
df = pd.read_excel(filepath, index_col=0)
df.index = pd.to_datetime(df.index)
return df
def load_macro_data(filepath):
"""
Load macro indicators from an Excel file.
Assumes first column is Date, then VIX, Consumer Confidence, IG Spreads.
"""
df = pd.read_excel(filepath, index_col=0)
df.index = pd.to_datetime(df.index)
df.columns = ['VIX', 'ConsumerConfidence', 'IGSpreads']
# Fill missing values
df = df.fillna(method='ffill').fillna(method='bfill')
return df
###############################################
# 2. Helper: Observation Dates
###############################################
def get_observation_dates(start_date, end_date, rebalance_period):
dates = []
current = start_date
while current <= end_date:
dates.append(current)
current += relativedelta(months=rebalance_period)
return dates
###############################################
# 3. Initialize the Portfolio & CASH
###############################################
def initialize_portfolio(prices, date, tickers, initial_aum):
portfolio = {}
allocation = initial_aum / len(tickers)
for ticker in tickers:
price = prices.loc[date, ticker]
portfolio[ticker] = allocation / price
return portfolio
###############################################
# 4. Lookback Metric Computation
###############################################
def compute_lookback_metric(prices, current_date, ticker, lookback_period, metric_type='simple'):
prices = prices.sort_index()
lookback_date = current_date - relativedelta(months=lookback_period)
current_price = prices[ticker].asof(current_date)
lookback_price = prices[ticker].asof(lookback_date)
if pd.isna(current_price) or pd.isna(lookback_price):
raise ValueError(f"Missing price data for {ticker} on {current_date} or {lookback_date}.")
if metric_type == 'simple':
metric = (current_price / lookback_price) - 1
elif metric_type == 'sma':
window = prices[ticker].loc[lookback_date:current_date]
if window.empty:
raise ValueError(f"No price data for {ticker} between {lookback_date} and {current_date}.")
sma = window.mean()
metric = (current_price - sma) / sma
else:
raise ValueError("Invalid metric type. Choose 'simple' or 'sma'.")
return metric
###############################################
# 5. Ranking Assets by Momentum
###############################################
def rank_assets(prices, current_date, tickers, lookback_period, metric_type):
metrics = {}
for ticker in tickers:
metric = compute_lookback_metric(prices, current_date, ticker, lookback_period, metric_type)
metrics[ticker] = metric
sorted_tickers = sorted(metrics, key=metrics.get, reverse=True)
ranks = {ticker: rank+1 for rank, ticker in enumerate(sorted_tickers)}
return sorted_tickers, ranks, metrics
###############################################
# 6. Compute Current Portfolio Value
###############################################
def compute_portfolio_value(portfolio, prices, current_date):
value = 0
for ticker, quantity in portfolio.items():
price = prices.loc[current_date, ticker]
value += quantity * price
return value
###############################################
# 7. Rebalance the Portfolio (Momentum Portion Only)
###############################################
def rebalance_portfolio(portfolio, prices, current_date, tickers, sorted_tickers,
internal_rebalance_ratios, rebalance_ratio):
portfolio_value = compute_portfolio_value(portfolio, prices, current_date)
rebalance_amount = portfolio_value * rebalance_ratio
target_trades = {ticker: rebalance_amount * internal_rebalance_ratios[i]
for i, ticker in enumerate(sorted_tickers)}
total_sold = 0
actual_trades = {}
for ticker, target_trade in target_trades.items():
price = prices.loc[current_date, ticker]
if target_trade < 0:
available_notional = portfolio[ticker] * price
sell_target = abs(target_trade)
actual_sell = min(available_notional, sell_target)
actual_trades[ticker] = -actual_sell
total_sold += actual_sell
else:
actual_trades[ticker] = 0
total_buy_target = sum(trade for trade in target_trades.values() if trade > 0)
if total_buy_target > 0:
for ticker, target_trade in target_trades.items():
if target_trade > 0:
proportion = target_trade / total_buy_target
buy_amount = total_sold * proportion
actual_trades[ticker] = buy_amount
new_portfolio = portfolio.copy()
for ticker, trade_notional in actual_trades.items():
price = prices.loc[current_date, ticker]
qty_change = trade_notional / price
new_portfolio[ticker] += qty_change
return new_portfolio, actual_trades, portfolio_value
###############################################
# 8. Macro: Composite Risk & HMM Functions
###############################################
def scale(x, low, high):
"""Safely scale x between 0 and 1 given bounds low and high."""
if high - low == 0:
return 0.0
return np.clip((x - low) / (high - low), 0, 1)
def compute_composite_risk(vix, consumer_confidence, ig_spreads, params):
"""
Convert each macro indicator into a risk score (0 to 1) using preset bounds.
For Consumer Confidence, higher values mean lower risk, so we invert the score.
"""
vix_low, vix_high = params['VIX']
cc_low, cc_high = params['ConsumerConfidence']
ig_low, ig_high = params['IGSpreads']
risk_vix = scale(vix, vix_low, vix_high)
risk_cc = 1 - scale(consumer_confidence, cc_low, cc_high)
risk_ig = scale(ig_spreads, ig_low, ig_high)
composite_risk = (risk_vix + risk_cc + risk_ig) / 3.0
if np.isnan(composite_risk) or np.isinf(composite_risk):
composite_risk = 0.5
return composite_risk
def fit_hmm_on_series(series, n_components=2):
"""
Fit a Gaussian HMM on a 1D numpy array of composite risk values.
Returns the fitted model and index of the bear state (state with higher mean).
"""
X = series.reshape(-1, 1)
model = GaussianHMM(n_components=n_components, covariance_type="diag", n_iter=1000, random_state=42)
model.fit(X)
state_means = model.means_.flatten()
bear_state = np.argmax(state_means)
return model, bear_state
def compute_bear_probability(model, bear_state, current_value):
"""
Compute the probability that the current composite risk value is generated by the bear state.
"""
prob = model.predict_proba(np.array([[current_value]]))[0]
p_bear = prob[bear_state]
return p_bear
def regime_allocation_from_hmm(p_bear, max_alloc=1.0, min_alloc=0.3, threshold=0.8):
"""
Map bear probability to target momentum allocation.
If p_bear < threshold, return max_alloc. Otherwise, linearly interpolate.
"""
if p_bear < threshold:
return max_alloc
else:
return max_alloc - (max_alloc - min_alloc) * ((p_bear - threshold) / (1 - threshold))
###############################################
# 9. Simulate the Strategy with HMM-Based Regime Switching & CASH Adjustments
###############################################
def simulate_strategy_with_hmm(prices, macro, eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period, rebalance_ratio,
lookback_period, metric_type,
internal_rebalance_ratios,
macro_params,
max_alloc=1.0, min_alloc=0.3,
hmm_window=24):
"""
Runs the simulation using an HMM on the composite risk signal.
For each rebalance date:
- Retrieve macro data (with missing values filled).
- Smooth the indicators using a 1-month moving average, using .asof() to fetch the nearest value.
- Compute composite risk.
- Append the risk to a rolling window; if enough points exist, fit a Gaussian HMM and compute the bear probability.
- Map the bear probability to target momentum allocation.
- Rebalance the momentum portfolio and adjust CASH accordingly.
"""
tickers = eq_tickers + fi_tickers + alts_tickers
obs_dates = get_observation_dates(start_date, end_date, rebalance_period)
results = []
# Rolling composite risk series for HMM fitting.
composite_risk_series = []
# Initial portfolio: fully invested in momentum, CASH = 0.
portfolio = initialize_portfolio(prices, start_date, tickers, initial_aum)
CASH = 0.0
portfolio_value = compute_portfolio_value(portfolio, prices, start_date)
total_aum = portfolio_value + CASH
prev_target_alloc = max_alloc
prev_regime = 'risk-on'
results.append({
'Date': start_date,
'Portfolio Value': portfolio_value,
'Momentum AUM': portfolio_value,
'CASH': CASH,
'Total AUM': total_aum,
'VIX': np.nan,
'ConsumerConfidence': np.nan,
'IGSpreads': np.nan,
'Composite Risk': np.nan,
'Bear Prob': np.nan,
'Target Momentum Alloc': prev_target_alloc,
'Regime': prev_regime,
'Return': 0,
**{f'qty_{ticker}': portfolio[ticker] for ticker in tickers},
**{f'notional_{ticker}': portfolio[ticker] * prices.loc[start_date, ticker] for ticker in tickers},
})
prev_total_aum = total_aum
vix_monthly = macro['VIX'].resample('M').last()
cc_monthly = macro['ConsumerConfidence'].resample('M').last()
ig_monthly = macro['IGSpreads'].resample('M').last()
for current_date in obs_dates[1:]:
# Retrieve macro data using asof to get nearest available value.
try:
vix = macro['VIX'].asof(current_date)
cc = macro['ConsumerConfidence'].asof(current_date)
ig = macro['IGSpreads'].asof(current_date)
except Exception as e:
raise ValueError(f"Error retrieving macro data for {current_date}: {e}")
# Smooth the indicators
window = 3
vix_roll = macro['VIX'].rolling(window=window, min_periods=1).mean()
cc_roll = macro['ConsumerConfidence'].rolling(window=window, min_periods=1).mean()
ig_roll = macro['IGSpreads'].rolling(window=window, min_periods=1).mean()
vix_smoothed = vix_roll.asof(current_date)
cc_smoothed = cc_roll.asof(current_date)
ig_smoothed = ig_roll.asof(current_date)
# Compute composite risk.
composite_risk = compute_composite_risk(vix, cc, ig, macro_params)
composite_risk_series.append(composite_risk)
# Fit HMM if we have enough data.
if len(composite_risk_series) >= hmm_window:
window_data = np.array(composite_risk_series[-hmm_window:])
if np.any(np.isnan(window_data)) or np.any(np.isinf(window_data)):
window_data = np.nan_to_num(window_data, nan=0.5, posinf=0.5, neginf=0.5)
model, bear_state = fit_hmm_on_series(window_data)
p_bear = compute_bear_probability(model, bear_state, composite_risk)
else:
p_bear = 0.0 # Default to risk-on if insufficient data.
# Map bear probability to target momentum allocation.
target_alloc = regime_allocation_from_hmm(p_bear, max_alloc, min_alloc, threshold=0.8)
regime = 'risk-on' if target_alloc >= max_alloc * 0.99 else 'risk-off'
# print(f"Date: {current_date.date()}, VIX: {vix:.2f}, CC: {cc:.2f}, IG: {ig:.2f}")
# print(f"Composite Risk: {composite_risk:.2f}, Bear Prob: {p_bear:.2f}, Target Alloc: {target_alloc:.2f}, Regime: {regime}")
# Rebalance momentum portfolio.
sorted_tickers, ranks, metrics = rank_assets(prices, current_date, tickers, lookback_period, metric_type)
portfolio, trades, pre_rebalance_value = rebalance_portfolio(
portfolio, prices, current_date, tickers, sorted_tickers,
internal_rebalance_ratios, rebalance_ratio)
# Compute momentum portfolio value and total investment.
mom_value = compute_portfolio_value(portfolio, prices, current_date)
total_investment = mom_value + CASH
# Determine desired momentum value and desired CASH.
desired_mom_value = target_alloc * total_investment
desired_cash = total_investment - desired_mom_value
# Adjust portfolio and CASH.
if mom_value > desired_mom_value:
excess = mom_value - desired_mom_value
for ticker in portfolio.keys():
price = prices.loc[current_date, ticker]
ticker_value = portfolio[ticker] * price
sell_amount = (ticker_value / mom_value) * excess
qty_to_sell = sell_amount / price
portfolio[ticker] -= qty_to_sell
CASH += excess
adjustment_note = f"Sold excess {excess:.2f} to CASH."
elif mom_value < desired_mom_value:
shortage = desired_mom_value - mom_value
if CASH >= shortage:
for ticker in portfolio.keys():
price = prices.loc[current_date, ticker]
ticker_value = portfolio[ticker] * price
target_weight = ticker_value / mom_value if mom_value > 0 else 1/len(portfolio)
invest_amount = target_weight * shortage
qty_to_buy = invest_amount / price
portfolio[ticker] += qty_to_buy
CASH -= shortage
adjustment_note = f"Bought into portfolio using {shortage:.2f} CASH."
else:
invest_amount = CASH
for ticker in portfolio.keys():
price = prices.loc[current_date, ticker]
ticker_value = portfolio[ticker] * price
target_weight = ticker_value / mom_value if mom_value > 0 else 1/len(portfolio)
qty_to_buy = (target_weight * invest_amount) / price
portfolio[ticker] += qty_to_buy
CASH = 0
adjustment_note = f"Partial buy using all available CASH."
else:
adjustment_note = "No cash adjustment needed."
mom_value = compute_portfolio_value(portfolio, prices, current_date)
total_aum = mom_value + CASH
ret = (total_aum - prev_total_aum) / prev_total_aum
prev_total_aum = total_aum
row = {
'Date': current_date,
'Portfolio Value': mom_value,
'Momentum AUM': mom_value,
'CASH': CASH,
'Total AUM': total_aum,
'Pre-Rebalance Value': pre_rebalance_value,
'VIX': vix,
'ConsumerConfidence': cc,
'IGSpreads': ig,
'Composite Risk': composite_risk,
'Bear Prob': p_bear,
'Target Momentum Alloc': target_alloc,
'Regime': regime,
'Adjustment Note': adjustment_note,
'Return': ret
}
for ticker in tickers:
row[f'qty_{ticker}'] = portfolio[ticker]
row[f'notional_{ticker}'] = portfolio[ticker] * prices.loc[current_date, ticker]
row[f'weight_{ticker}'] = (portfolio[ticker] * prices.loc[current_date, ticker]) / mom_value if mom_value > 0 else np.nan
row[f'rank_{ticker}'] = ranks.get(ticker, np.nan)
row[f'metric_{ticker}'] = metrics.get(ticker, np.nan)
row[f'trade_{ticker}'] = trades.get(ticker, 0)
results.append(row)
result_df = pd.DataFrame(results)
result_df.set_index('Date', inplace=True)
diag_columns = ['Total AUM', 'Momentum AUM', 'CASH', 'VIX', 'ConsumerConfidence',
'IGSpreads', 'Composite Risk', 'Bear Prob', 'Target Momentum Alloc', 'Regime',
'Adjustment Note', 'Return']
for prefix in ['qty_', 'notional_', 'weight_', 'rank_', 'metric_', 'trade_']:
diag_columns.extend([f'{prefix}{ticker}' for ticker in tickers])
result_df = result_df[diag_columns]
return result_df
###############################################
# 10. Main – Example Usage
###############################################
if __name__ == '__main__':
# Define asset tickers.
eq_tickers = ['SPY US Equity']
fi_tickers = ['TLT US Equity', 'HYG US Equity']
alts_tickers = ['GLD US Equity', 'SHV US Equity', 'VNQ US Equity']
initial_aum = 100e6 # 100 million
start_date = pd.to_datetime('2012-01-01')
end_date = pd.to_datetime('2025-02-01')
rebalance_period = 1 # monthly (or adjust as desired)
rebalance_ratio = 0.2 # 20% of current momentum AUM rebalanced each period
lookback_period = 6 # 3-month lookback
metric_type = 'simple'
internal_rebalance_ratios = [0.7, 0.3, 0, 0, -0.3, -0.7]
# Macro parameters (bounds for each indicator)
macro_params = {
'VIX': (10, 40),
'ConsumerConfidence': (45, 130), # Adjusted lower bound to 80
'IGSpreads': (100, 200)
}
# File paths.
price_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\Codes\Historic Prices.xlsx"
macro_filepath = r"\\asiapac.nom\data\MUM\IWM\India_IWM_IPAS\Reet\Momentum Strategy\macro_indic.xlsx"
prices = load_price_data(price_filepath)
macro = load_macro_data(macro_filepath)
# Set hmm_window to 24 months (or adjust as needed)
result_df = simulate_strategy_with_hmm(prices, macro, eq_tickers, fi_tickers, alts_tickers,
initial_aum, start_date, end_date,
rebalance_period, rebalance_ratio,
lookback_period, metric_type,
internal_rebalance_ratios,
macro_params,
max_alloc=1.0, min_alloc=0.5,
hmm_window=24)
pd.set_option('display.float_format', lambda x: f'{x:,.2f}')
# print(result_df)Editor is loading...
Leave a Comment