Untitled
unknown
plain_text
21 days ago
15 kB
3
Indexable
import pandas as pd import numpy as np from statsmodels.tools.tools import add_constant from statsmodels.discrete.discrete_model import Logit # === STEP 1: LOAD MACRO DATA === # def load_macro_data(filepath): """ Load macro indicators from Excel. - VIX, VIX3M from 'Eq' sheet - Remove weekends """ 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] return vix_data # === STEP 2: SLOPE + LOGIT MODEL === # def calculate_slopes(series, window=5): return series.rolling(window).apply(lambda x: np.polyfit(np.arange(len(x)), x, 1)[0], raw=True) def split_slope(slope_series): return slope_series.clip(lower=0), slope_series.clip(upper=0) def logistic_regression(X, y): X = add_constant(X) model = Logit(y, X).fit(disp=False) return model # === STEP 3: BACKTEST === # 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): 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) # Merge daily (no resampling) 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) # Modified this part: X_test_data = merged[['Slope+', 'Slope−']].iloc[i:i+1] # Make sure to use the same structure as training data X_test = add_constant(X_test_data, has_constant='add') prob = model.predict(X_test)[0] actual_return = merged['SPX_Return'].iloc[i] signal = 'risk_off' if prob > prob_threshold else 'risk_on' results.append({ 'Date': current_date, 'Predicted_Prob_Negative': prob, 'Actual_Return': actual_return, 'Signal': signal, 'Was_Negative': actual_return < 0, 'Correct_Prediction': (actual_return < 0 and signal == 'risk_off'), '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') return df_results # === STEP 4: EVALUATION & EXPORT === # def evaluate_and_export_results(df_results, lookback_window, prob_threshold): total_neg = df_results['Was_Negative'].sum() correct = df_results['Correct_Prediction'].sum() false_alarms = ((df_results['Signal'] == 'risk_off') & (~df_results['Was_Negative'])).sum() precision = correct / (correct + false_alarms + 1e-8) print("\n📊 Accuracy Report (Threshold = {:.2f}):".format(prob_threshold)) print(f"Negative Return Days: {total_neg}") print(f"Correct Risk-Off Calls: {correct}") print(f"False Alarms: {false_alarms}") print(f"Precision on Negatives: {precision:.4f}") filename = f"logit_vix_signals_L{lookback_window}_T{int(prob_threshold*100)}.xlsx" df_results.to_excel(filename) print(f"\n📁 Exported to: {filename}") # === STEP 5: EXECUTE === # 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 = pd.read_excel(price_filepath, sheet_name='Sheet1', index_col=0, parse_dates=True) macro_data = load_macro_data(macro_filepath) lookback_window = 504 start_date = '2008-01-01' prob_threshold = 0.6 # Increase to reduce false alarms df_results = run_vix_slope_signal_backtest(prices, macro_data, start_date=start_date, lookback_window=lookback_window, prob_threshold=prob_threshold) evaluate_and_export_results(df_results, lookback_window, prob_threshold) yes now my simulate strategy code wants to call this function for each trade date, but I only want this regression etc to return the porbability, risk off or risk on signal and target allocation signal. can we intgreate that in my simulate strategy function below. you cannot change the way simulate strategy works. but you can change the generate vix signal function 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): """ Simulation of a momentum strategy that uses daily VIX slope signals. For each day, we use the original backtest methodology: - Run the regression over data from (current_date - lookback_window) to (current_date - 1) - Compute the signal for that date. Then we shift the entire signal DataFrame by one business day so that yesterday's regression applies to today. """ # Define tickers for momentum (excluding the 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, std if needed) macro = macro_data.copy() macro['VIX_Spread'] = macro['VIX'] - macro['VIX3M'] macro['VIX_Spread_EMA'] = macro["VIX_Spread"].ewm(span=5, adjust=False).mean() macro["Mean"] = macro['VIX_Spread_EMA'].rolling(window=504).mean() macro["Std"] = macro['VIX_Spread_EMA'].rolling(window=504).std() # Obtain the full set of VIX signals using the original backtest function vix_signal_df = run_vix_slope_signal_backtest(prices, macro, start_date=start_date, slope_window=slope_window, lookback_window=lookback_window, forecast_horizon=forecast_horizon) # Shift the index by one business day so that yesterday’s signal applies to today vix_signal_df.index = vix_signal_df.index.to_series().shift(1, freq='B') vix_signal_df = vix_signal_df.dropna() # Initialize portfolio (for momentum securities) and cash portfolio = initialize_portfolio(prices, start_date, momentum_tickers, initial_aum) cash_qty = 0.0 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 # Get today's VIX signal from the shifted DataFrame (if available) if current_date in vix_signal_df.index: row = vix_signal_df.loc[current_date] prob = row['Predicted_Prob_Negative'] vix_signal = row['Signal'] vix_target_alloc = row['Target_Allocation'] else: # Default to risk_on if no signal is available prob = 1.0 vix_signal = 'risk_on' vix_target_alloc = 1.0 # Determine portfolio value and benchmark 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 the VIX signal if vix_signal == 'risk_off': if (spy_weight + hyg_weight) < 0.40: current_regime = 'risk-on' target_alloc = 1.0 daily_note = "Forced risk-on (SPY+HYG < 40%)" else: current_regime = 'risk_off' target_alloc = vix_target_alloc else: current_regime = 'risk-on' target_alloc = 1.0 # Cash rebalancing: adjust only 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 (SPY+HYG weight < 40%)" 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 and compute return 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, 'Predicted Prob': prob, 'Adjustment Note': daily_note, 'Cash Adjustment': cash_adjustment, 'Return': ret, 'Event': 'Monthly Rebalance' if current_date in monthly_dates else 'Daily Check' } results.append(row) result_df = pd.DataFrame(results) result_df.set_index('Date', inplace=True) return result_df
Editor is loading...
Leave a Comment