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