Untitled
unknown
plain_text
20 days ago
29 kB
2
Indexable
import pandas as pd import numpy as np from datetime import datetime from dateutil.relativedelta import relativedelta import statsmodels.api as sm from statsmodels.tools import add_constant from scipy.stats import zscore ############################################### # 0. Helper Functions for VIX Slope Signal ############################################### def calculate_slopes(series, window=5): """ Calculate a simple slope as the difference over the window, divided by the window. """ return series.diff(window) / window def split_slope(slope_series): """ Split a slope series into its positive and negative components. Positive slopes remain; negative slopes become their absolute value. """ slope_pos = slope_series.clip(lower=0) slope_neg = (-slope_series.clip(upper=0)) return slope_pos, slope_neg def logistic_regression(X, y): """ Fit a logistic regression model using statsmodels. """ X_const = add_constant(X, has_constant='add') model = sm.Logit(y, X_const).fit(disp=0) return model ############################################### # 1. Data Loading (Filtering Out Weekends) ############################################### 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. Removes weekend data. """ df = pd.read_excel(filepath, index_col=0) df.index = pd.to_datetime(df.index) df = df.sort_index() # Filter out weekends (Saturday=5, Sunday=6) df = df[df.index.dayofweek < 5] return df def load_macro_data(filepath): """ Load macro indicators from an Excel file. Loads VIX and VIX3M from the 'Eq' sheet, LF98TRUU Index from the 'FI' sheet (loaded but not used for signals), and other macro indicators from the 'Macro' sheet. Removes weekend data. Also computes slopes for selected macro indicators. """ # VIX data vix_data = pd.read_excel(filepath, sheet_name='Eq', index_col=0, parse_dates=True, usecols=[0, 4, 5]) vix_data.columns = ['VIX', 'VIX3M'] vix_data = vix_data[vix_data.index.dayofweek < 5] # FI data (loaded but not used in the strategy) cdx_data = pd.read_excel(filepath, sheet_name='FI', index_col=0, parse_dates=True, usecols=[0, 2], skiprows=1) cdx_data.columns = ['LF98TRUU'] cdx_data = cdx_data[cdx_data.index.dayofweek < 5] # Macro data (assumed to include columns like "CESIUSD Index", etc.) macro_data = pd.read_excel(filepath, sheet_name='Macro', index_col=0, parse_dates=True, usecols=range(8), skiprows=1) macro_data = macro_data[macro_data.index.dayofweek < 5] # Compute slopes for selected macro indicators. macro_data["Surprise Index Slope"] = macro_data["CESIUSD Index"].diff() macro_data["Jobless Claims Slope"] = macro_data["INJCJC Index"].diff() macro_data["Copper Gold Slope"] = macro_data['.HG/GC G Index'].diff() combined_data = pd.concat([vix_data, cdx_data, macro_data], axis=1) combined_data = combined_data.fillna(method='ffill').fillna(method='bfill') combined_data = combined_data.sort_index() # ensure sorted index return combined_data ############################################### # 2. Helper: Observation Dates (Monthly) ############################################### def get_observation_dates(prices, start_date, end_date, rebalance_period): dates = [] current_date = start_date while current_date < end_date: candidate_date = (current_date + relativedelta(months=rebalance_period)).replace(day=1) while candidate_date not in prices.index: candidate_date += pd.Timedelta(days=1) if candidate_date.month != (current_date + relativedelta(months=rebalance_period)).month: candidate_date = None break if candidate_date is None or candidate_date > end_date: break dates.append(candidate_date) current_date = candidate_date return dates ############################################### # 3. Portfolio Initialization ############################################### def initialize_portfolio(prices, date, tickers, initial_aum): portfolio = {} mask = prices.index < date prev_date = prices.index[mask][-1] allocation = initial_aum / len(tickers) for ticker in tickers: price = prices.loc[prev_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() mask = prices.index < current_date prev_date = prices.index[mask][-1] lookback_date = prev_date - relativedelta(months=lookback_period) current_price = prices[ticker].asof(prev_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 {prev_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 Momentum Portfolio ############################################### def rebalance_portfolio(portfolio, prices, current_date, tickers, sorted_tickers, internal_rebalance_ratios, rebalance_ratio): mask = prices.index < current_date prev_date = prices.index[mask][-1] prev_prices = prices.loc[prev_date] curr_prices = prices.loc[current_date] portfolio_value = sum(portfolio[ticker] * prev_prices[ticker] for ticker in tickers) 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(): if target_trade < 0: available_notional = portfolio[ticker] * curr_prices[ticker] 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(t for t in target_trades.values() if t > 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(): execution_price = curr_prices[ticker] qty_change = trade_notional / execution_price new_portfolio[ticker] += qty_change return new_portfolio, actual_trades, portfolio_value def adjust_overweight(portfolio, prices, current_date, sorted_tickers, threshold=0.70): mask = prices.index < current_date prev_date = prices.index[mask][-1] prev_prices = prices.loc[prev_date] curr_prices = prices.loc[current_date] portfolio_value = sum(portfolio[ticker] * prev_prices[ticker] for ticker in portfolio) weights = {ticker: (portfolio[ticker] * prev_prices[ticker]) / portfolio_value for ticker in portfolio} new_portfolio = portfolio.copy() for overweight in portfolio: if weights[overweight] > threshold: extra_weight = weights[overweight] - threshold extra_value = extra_weight * portfolio_value execution_price_over = curr_prices[overweight] qty_reduce = extra_value / execution_price_over new_portfolio[overweight] -= qty_reduce remaining_value = extra_value for candidate in sorted_tickers: if candidate == overweight: continue candidate_value = new_portfolio[candidate] * curr_prices[candidate] candidate_weight = candidate_value / portfolio_value if candidate_weight < threshold: capacity = (threshold - candidate_weight) * portfolio_value allocation = min(remaining_value, capacity) qty_add = allocation / curr_prices[candidate] new_portfolio[candidate] += qty_add remaining_value -= allocation if remaining_value <= 0: break return new_portfolio ############################################### # 8. VIX Slope Signal Generation (Logistic Regression) ############################################### def run_vix_slope_signal_backtest(prices, macro_data, start_date='2008-01-01', slope_window=5, lookback_window=252, forecast_horizon=5, prob_threshold=0.6): """ Backtest a VIX slope–based signal. Calculates the VIX spread slopes, splits them into positive and negative parts, and uses a rolling logistic regression to predict the probability of a negative SPX return. IMPORTANT: The regression is run on data up to a given day and then its output is shifted forward one trading day. This means if the regression (using yesterday’s data) indicates 'risk_off', that signal will be applied for today. Returns a DataFrame with the predicted probability, signal, and target allocation, with the index shifted to represent the next trading day. """ macro_data = macro_data.copy() macro_data['VIX_Spread'] = macro_data['VIX'] - macro_data['VIX3M'] macro_data['Slope'] = calculate_slopes(macro_data['VIX_Spread'], window=slope_window) macro_data['Slope+'], macro_data['Slope−'] = split_slope(macro_data['Slope']) prices = prices.copy() prices['SPX_Return'] = prices['SPY US Equity'].pct_change(forecast_horizon).shift(-forecast_horizon) merged = macro_data[['Slope+', 'Slope−']].join(prices[['SPX_Return']], how='inner').dropna() merged = merged[merged.index >= start_date] results = [] full_data = macro_data[['Slope+', 'Slope−']].join(prices[['SPX_Return']], how='inner').dropna() for i in range(len(merged)): current_date = merged.index[i] full_data_current_idx = full_data.index.get_loc(current_date) if full_data_current_idx < lookback_window: continue train = full_data.iloc[full_data_current_idx - lookback_window:full_data_current_idx] X_train = train[['Slope+', 'Slope−']] y_train = (train['SPX_Return'] < 0).astype(int) try: model = logistic_regression(X_train, y_train) X_test_data = merged[['Slope+', 'Slope−']].iloc[i:i+1] X_test = add_constant(X_test_data, has_constant='add') prob = model.predict(X_test)[0] actual_return = merged['SPX_Return'].iloc[i] # New logic for signal and target allocation based on predicted probability if prob <= 0.6: signal = 'risk_on' target_allocation = 1.0 elif 0.6 < prob <= 0.8: signal = 'risk_off' target_allocation = 0.8 else: # prob > 0.8 signal = 'risk_off' target_allocation = 0.2 results.append({ 'Date': current_date, 'Predicted_Prob_Negative': prob, 'Actual_Return': actual_return, 'Signal': signal, 'Target_Allocation': target_allocation, 'Slope+': merged['Slope+'].iloc[i], 'Slope−': merged['Slope−'].iloc[i] }) except Exception as e: print(f"Error at date {current_date}: {str(e)}") continue df_results = pd.DataFrame(results).set_index('Date') # Shift the predicted signal to the next trading day df_results.index = df_results.index.to_series().shift(1, freq='B') df_results = df_results.dropna() return df_results ############################################### # 9. Helper Functions for Cash (using a cash ticker) ############################################### def invest_cash_into_portfolio(portfolio, prices, current_date, cash_qty, cash_ticker): """ When switching from risk-off to risk-on, sell the cash instrument (e.g. SHV) and reinvest its proceeds into the portfolio using previous-day weights. """ cash_price = prices.loc[current_date, cash_ticker] cash_available = cash_qty * cash_price if cash_available <= 0: return portfolio, cash_qty, f"No {cash_ticker} to reinvest." mask = prices.index < current_date prev_date = prices.index[mask][-1] mom_value = compute_portfolio_value(portfolio, prices, current_date) new_portfolio = portfolio.copy() for ticker in portfolio: prev_price = prices.loc[prev_date, ticker] total_qty = (portfolio[ticker] * (mom_value + cash_available)) / mom_value if mom_value > 0 else 1/len(portfolio) new_portfolio[ticker] = total_qty return new_portfolio, 0.0, f"Reinvested {cash_available:,.2f} from {cash_ticker}" def allocate_cash_from_portfolio(portfolio, prices, current_date, target_alloc, cash_qty, cash_ticker): """ Adjusts portfolio positions based on target allocation for cash. When deploying cash, scales up existing positions proportionally to reach target allocation. """ mask = prices.index < current_date prev_date = prices.index[mask][-1] curr_value = compute_portfolio_value(portfolio, prices, prev_date) cash_price = prices.loc[prev_date, cash_ticker] cash_equiv = cash_qty * cash_price total_aum = curr_value + cash_equiv desired_cash = (1 - target_alloc) * total_aum cash_price = prices.loc[current_date, cash_ticker] cash_equiv = cash_qty * cash_price current_cash = cash_equiv new_portfolio = portfolio.copy() new_cash_qty = cash_qty note = "" if desired_cash > current_cash: cash_to_raise = desired_cash - current_cash curr_value = compute_portfolio_value(portfolio, prices, current_date) for ticker in portfolio: price = prices.loc[current_date, ticker] sell_ratio = (cash_to_raise / curr_value) qty_to_sell = portfolio[ticker] * sell_ratio new_portfolio[ticker] -= qty_to_sell new_cash_qty += cash_to_raise / cash_price note = f"Raised {cash_to_raise:,.2f} into {cash_ticker}" elif desired_cash < current_cash: current_portfolio_value = sum( portfolio[ticker] * prices.loc[current_date, ticker] for ticker in portfolio ) desired_risk_allocation = total_aum * target_alloc scaling_factor = desired_risk_allocation / current_portfolio_value for ticker in portfolio: new_portfolio[ticker] = portfolio[ticker] * scaling_factor excess_cash = current_cash - desired_cash new_cash_qty -= excess_cash / cash_price note = f"Deployed {excess_cash:,.2f} from {cash_ticker} into portfolio" return new_portfolio, new_cash_qty, note ############################################### # 10. Simulation: Refined Strategy with Cash & VIX Slope Signal ############################################### def simulate_strategy(prices, macro_data, eq_tickers, fi_tickers, alts_tickers, initial_aum, start_date, end_date, rebalance_period, rebalance_ratio, lookback_period, metric_type, internal_rebalance_ratios, cash_ticker='SHV US Equity', macro_max_alloc=1.0, macro_min_alloc=0.6, slope_window=5, lookback_window=252, forecast_horizon=5, prob_threshold=0.6): """ Simulation of a momentum strategy using VIX slope signals generated via logistic regression. FI signals are removed. A designated cash_ticker is used for the cash component. Extra portfolio value is moved to cash when in risk-off, and reinvested when switching back to risk-on. """ # Define tickers for momentum (exclude cash_ticker) all_tickers = eq_tickers + fi_tickers + alts_tickers momentum_tickers = [t for t in all_tickers if t != cash_ticker] monthly_dates = get_observation_dates(prices, start_date, end_date, rebalance_period) daily_dates = prices.index.sort_values() daily_dates = daily_dates[(daily_dates >= start_date) & (daily_dates <= end_date)] # Prepare macro data: calculate VIX spread and its EMA, mean, and std. macro_data = macro_data.copy() macro_data['VIX_Spread'] = macro_data['VIX'] - macro_data['VIX3M'] macro_data['VIX_Spread_EMA'] = macro_data["VIX_Spread"].ewm(span=5, adjust=False).mean() macro_data["Mean"] = macro_data['VIX_Spread_EMA'].rolling(window=504).mean() macro_data["Std"] = macro_data['VIX_Spread_EMA'].rolling(window=504).std() # Generate VIX slope signals via the new backtest function. vix_signal_df = run_vix_slope_signal_backtest(prices, macro_data, start_date=start_date, slope_window=slope_window, lookback_window=lookback_window, forecast_horizon=forecast_horizon, prob_threshold=prob_threshold) # Initialize portfolio (momentum securities only) and cash (in cash_ticker) portfolio = initialize_portfolio(prices, start_date, momentum_tickers, initial_aum) cash_qty = 0.0 # cash position held as cash_ticker units current_regime = 'risk-on' target_alloc = 1.0 previous_regime = current_regime previous_target_alloc = target_alloc prev_total_aum = initial_aum results = [] for current_date in daily_dates: daily_note = "No adjustment" cash_adjustment = 0.0 # --- Determine VIX signal from the backtest results --- if current_date in vix_signal_df.index: vix_target_alloc = vix_signal_df.loc[current_date, 'Target_Allocation'] vix_signal = vix_signal_df.loc[current_date, 'Signal'] else: vix_target_alloc = macro_max_alloc vix_signal = 'risk_on' # --- Determine portfolio value and SPY/HYG weights --- mask = prices.index < current_date prev_date = prices.index[mask][-1] mom_value = compute_portfolio_value(portfolio, prices, prev_date) spy_weight = (portfolio.get('SPY US Equity', 0) * prices.loc[prev_date, 'SPY US Equity']) / mom_value if mom_value > 0 else 0 hyg_weight = (portfolio.get('HYG US Equity', 0) * prices.loc[prev_date, 'HYG US Equity']) / mom_value if mom_value > 0 else 0 # --- Determine daily regime based solely on VIX slope signal --- if vix_signal == 'risk_off': if (spy_weight + hyg_weight) < 0.40: current_regime = 'risk-on' target_alloc = 1.0 daily_note = "Forced regime to risk-on & target alloc 100% due to SPY+HYG < 40%" else: current_regime = 'risk_off' target_alloc = vix_target_alloc else: current_regime = 'risk-on' target_alloc = 1.0 # --- Cash rebalancing logic: Adjust cash position when regime or target allocation changes --- if (previous_regime != current_regime) or (current_regime == 'risk_off' and target_alloc != previous_target_alloc): if previous_regime == 'risk_off' and current_regime == 'risk-on' and cash_qty > 0: portfolio, cash_qty, note_update = invest_cash_into_portfolio(portfolio, prices, current_date, cash_qty, cash_ticker) daily_note += " | " + note_update elif (previous_regime == 'risk-on' and current_regime == 'risk_off') or (current_regime == 'risk_off' and target_alloc != previous_target_alloc): portfolio, cash_qty, note_update = allocate_cash_from_portfolio(portfolio, prices, current_date, target_alloc, cash_qty, cash_ticker) daily_note += " | " + note_update previous_regime = current_regime previous_target_alloc = target_alloc # --- Monthly Rebalancing --- if current_date in monthly_dates: sorted_tickers, ranks, metrics = rank_assets(prices, current_date, momentum_tickers, lookback_period, metric_type) temp_portfolio, trades, pre_rebalance_value = rebalance_portfolio(portfolio, prices, current_date, momentum_tickers, sorted_tickers, internal_rebalance_ratios, rebalance_ratio) temp_portfolio = adjust_overweight(temp_portfolio, prices, current_date, sorted_tickers, threshold=0.70) temp_value = compute_portfolio_value(temp_portfolio, prices, current_date) spy_temp = temp_portfolio.get('SPY US Equity', 0) * prices.loc[current_date, 'SPY US Equity'] hyg_temp = temp_portfolio.get('HYG US Equity', 0) * prices.loc[current_date, 'HYG US Equity'] combined_weight = (spy_temp + hyg_temp) / temp_value if temp_value > 0 else 0 if (current_regime == 'risk_off') and (combined_weight < 0.40): current_regime = 'risk-on' target_alloc = 1.0 daily_note += " | Monthly: Forced risk-on due to SPY+HYG weight < 40% after simulation." total_aum = compute_portfolio_value(portfolio, prices, current_date) + cash_qty * prices.loc[current_date, cash_ticker] simulated_value = temp_value new_portfolio = {} for ticker in temp_portfolio: price = prices.loc[current_date, ticker] simulated_weight = (temp_portfolio[ticker] * price) / simulated_value if simulated_value > 0 else 1/len(temp_portfolio) new_qty = (total_aum * simulated_weight) / price new_portfolio[ticker] = new_qty portfolio = new_portfolio cash_qty = 0 else: portfolio = temp_portfolio curr_value = compute_portfolio_value(portfolio, prices, current_date) total_aum = curr_value + cash_qty * prices.loc[current_date, cash_ticker] desired_value = target_alloc * total_aum if curr_value > desired_value: portfolio, cash_qty, note_update = allocate_cash_from_portfolio(portfolio, prices, current_date, target_alloc, cash_qty, cash_ticker) daily_note += " | Monthly: " + note_update elif curr_value < desired_value and cash_qty > 0: portfolio, cash_qty, note_update = allocate_cash_from_portfolio(portfolio, prices, current_date, target_alloc, cash_qty, cash_ticker) daily_note += " | Monthly: " + note_update # --- Update daily AUM calculation --- current_mom_value = compute_portfolio_value(portfolio, prices, current_date) cash_price = prices.loc[current_date, cash_ticker] cash_value = cash_qty * cash_price total_aum = current_mom_value + cash_value ret = (total_aum - prev_total_aum) / prev_total_aum if prev_total_aum > 0 else 0 prev_total_aum = total_aum # --- Log daily results --- row = { 'Date': current_date, 'Momentum AUM': current_mom_value, 'Cash Qty': cash_qty, 'Cash Price': cash_price, 'Cash Value': cash_value, 'Total AUM': total_aum, 'Current Regime': current_regime, 'Target Alloc': target_alloc, 'VIX Target': vix_target_alloc, 'VIX Signal': vix_signal, 'Adjustment Note': daily_note, 'Cash Adjustment': cash_adjustment, 'Return': ret, 'Event': 'Monthly Rebalance' if current_date in monthly_dates else 'Daily Check' } for ticker in momentum_tickers: price = prices.loc[current_date, ticker] qty = portfolio[ticker] notional = qty * price row[f'qty_{ticker}'] = qty row[f'notional_{ticker}'] = notional row[f'weight_{ticker}'] = (notional / current_mom_value) if current_mom_value > 0 else np.nan row[f'rank_{ticker}'] = ranks.get(ticker, np.nan) if current_date in monthly_dates else np.nan row[f'metric_{ticker}'] = metrics.get(ticker, np.nan) if current_date in monthly_dates else np.nan row[f'trade_{ticker}'] = trades.get(ticker, 0) if current_date in monthly_dates else 0 results.append(row) result_df = pd.DataFrame(results) result_df.set_index('Date', inplace=True) return result_df ############################################### # 11. 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', 'IGSB US Equity'] initial_aum = 100e6 start_date = pd.to_datetime('2008-01-01') end_date = pd.to_datetime('2025-02-01') rebalance_period = 1 rebalance_ratio = 0.2 lookback_period = 6 metric_type = 'simple' internal_rebalance_ratios = [0.8, 0.2, 0, -0.2, -0.8] # File paths (adjust these to your environment). 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\Momentum Strategy Overlay Data.xlsx" prices = load_price_data(price_filepath) macro_data = load_macro_data(macro_filepath) # Run simulation. result_df = simulate_strategy(prices, macro_data, eq_tickers, fi_tickers, alts_tickers, initial_aum, start_date, end_date, rebalance_period, rebalance_ratio, lookback_period, metric_type, internal_rebalance_ratios, cash_ticker='SHV US Equity', macro_max_alloc=1.0, macro_min_alloc=0.6, slope_window=5, lookback_window=252, forecast_horizon=5, prob_threshold=0.6) pd.set_option('display.float_format', lambda x: f'{x:,.2f}') # For example, print the last few rows: # print(result_df[['Total AUM', 'Momentum AUM', 'Cash Price']].tail())
Editor is loading...
Leave a Comment