"""P0 validation for TrendRiderV5 — walk-forward + bootstrap. Critical question: were V5's panic-demote thresholds curve-fit to the 2024-08 carry-trade unwind? Test by optimizing on IS (2015-2020, which does NOT contain 2024-08) and evaluating on OOS (2021-2026, which DOES). If IS-best params still rescue the OOS drawdown, the mechanism is real. """ from __future__ import annotations import os import sys from dataclasses import asdict from itertools import product import numpy as np import pandas as pd sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from research.trend_rider_robustness import ( buy_hold_weights, evaluate_weights, load_price_panel, portfolio_returns, ) from research.trend_rider_p0 import block_bootstrap, bootstrap_summary from strategies.permanent import TrendRiderV3 from strategies.trend_rider_v5 import TrendRiderV5 IS_START = "2015-01-02" IS_END = "2020-12-31" OOS_START = "2021-01-01" OOS_END = "2026-05-07" def _fmt(x: float) -> str: return f"{x * 100:7.2f}%" def print_eval(label: str, ev) -> None: print( f" {label:<36s} " f"CAGR {_fmt(ev.cagr)} Sharpe {ev.sharpe:5.2f} " f"MDD {_fmt(ev.max_drawdown)} Calmar {ev.calmar:5.2f} " f"X {ev.final_multiple:6.2f}" ) def panic_grid() -> list[dict]: return [ { "panic_vol_ratio": vr, "panic_peak_drop_pct": pd_, "panic_vol_short": vs, "panic_peak_window": pw, } for vr, pd_, vs, pw in product( [1.4, 1.5, 1.6, 1.7, 1.8, 2.0], [0.03, 0.04, 0.05, 0.06], [3, 5, 7], [3, 5, 7], ) ] def main() -> None: prices = load_price_panel() print(f"Panel: {prices.index.min().date()} to {prices.index.max().date()}") # ----- Walk-forward: choose panic config by IS Calmar (CAGR/|MDD|) ----- print("\n" + "=" * 78) print(f"P0.1 — Walk-forward (IS panic-grid optimization → OOS test)") print(f" IS: {IS_START} → {IS_END} (does NOT contain 2024-08 crash)") print(f" OOS: {OOS_START} → {OOS_END}") print("=" * 78) grid = panic_grid() is_rows = [] oos_rows = [] for kwargs in grid: strat = TrendRiderV5(**kwargs) weights = strat.generate_signals(prices) ev_is = evaluate_weights("is", weights, prices[weights.columns], 0.001, IS_START, IS_END) ev_oos = evaluate_weights("oos", weights, prices[weights.columns], 0.001, OOS_START, OOS_END) is_rows.append({**asdict(ev_is), **kwargs, "scope": "IS"}) oos_rows.append({**asdict(ev_oos), **kwargs, "scope": "OOS"}) is_df = pd.DataFrame(is_rows) oos_df = pd.DataFrame(oos_rows) is_df["calmar"] = is_df["cagr"] / is_df["max_drawdown"].abs().replace(0.0, np.nan) oos_df["calmar"] = oos_df["cagr"] / oos_df["max_drawdown"].abs().replace(0.0, np.nan) # Rank by IS Calmar is_df = is_df.sort_values("calmar", ascending=False).reset_index(drop=True) print(f"\n Grid size: {len(grid)}, top 5 by IS Calmar:") show_cols = ["cagr", "sharpe", "max_drawdown", "calmar", "panic_vol_ratio", "panic_peak_drop_pct", "panic_vol_short", "panic_peak_window"] print(is_df[show_cols].head(5).to_string(index=False)) # IS-best by Calmar best = is_df.iloc[0] best_kwargs = {k: best[k] for k in ("panic_vol_ratio", "panic_peak_drop_pct", "panic_vol_short", "panic_peak_window")} best_kwargs["panic_vol_short"] = int(best_kwargs["panic_vol_short"]) best_kwargs["panic_peak_window"] = int(best_kwargs["panic_peak_window"]) best_kwargs["panic_vol_ratio"] = float(best_kwargs["panic_vol_ratio"]) best_kwargs["panic_peak_drop_pct"] = float(best_kwargs["panic_peak_drop_pct"]) print(f"\n IS-best (by Calmar): {best_kwargs}") print(f" IS CAGR {best['cagr']*100:.2f}% MDD {best['max_drawdown']*100:.2f}% " f"Calmar {best['calmar']:.2f}") # OOS performance of IS-best isbest_strat = TrendRiderV5(**best_kwargs) w_isbest = isbest_strat.generate_signals(prices) is_best_oos = evaluate_weights("is_best_OOS", w_isbest, prices[w_isbest.columns], 0.001, OOS_START, OOS_END) print(f" Same params, OOS performance:") print_eval("IS-best (OOS)", is_best_oos) # Compare with V3 default and V5 (default panic = 1.6/4%) on each window cmp_strats = { "V3 default": TrendRiderV3(), "V5 default (1.6 / 4%)": TrendRiderV5(), f"V5 IS-best (Calmar)": TrendRiderV5(**best_kwargs), } print("\n Comparison on full / IS / OOS:") for window_name, (s, e) in {"FULL": (IS_START, OOS_END), "IS": (IS_START, IS_END), "OOS": (OOS_START, OOS_END)}.items(): print(f" --- {window_name} ({s} → {e}) ---") for n, strat in cmp_strats.items(): w = strat.generate_signals(prices) ev = evaluate_weights(n, w, prices[w.columns], 0.001, s, e) print_eval(n, ev) spy_w = buy_hold_weights(prices, "SPY") ev = evaluate_weights("SPY B&H", spy_w, prices[spy_w.columns], 0.0, s, e) print_eval("SPY B&H", ev) # IS-OOS decay analysis decay_cagr = best["cagr"] - is_best_oos.cagr print(f"\n Decay (IS-best CAGR IS → OOS): {decay_cagr*100:+.2f}%") print(f" IS-best preserved OOS MDD: {is_best_oos.max_drawdown*100:.2f}% " f"(V3 OOS MDD = -37.54%)") # ----- Bootstrap on V5 default returns ----- print("\n" + "=" * 78) print("P0.2 — Block bootstrap (V5 default, block_len=21, n_boot=5000)") print("=" * 78) v5 = TrendRiderV5() weights = v5.generate_signals(prices) rets = portfolio_returns(weights, prices[weights.columns], 0.001) rets = rets[(rets.index >= IS_START) & (rets.index <= OOS_END)] boot = block_bootstrap(rets, n_boot=5000, block_len=21, seed=42) print("\n Full-sample bootstrap (2015-2026):") print(bootstrap_summary(boot).round(4).to_string()) p_neg = float((boot["cagr"] < 0).mean()) p_below_spy = float((boot["cagr"] < 0.15).mean()) p_dd_30 = float((boot["max_drawdown"] < -0.30).mean()) p_dd_40 = float((boot["max_drawdown"] < -0.40).mean()) p_dd_50 = float((boot["max_drawdown"] < -0.50).mean()) print(f"\n P(CAGR<0) = {p_neg:.3f}") print(f" P(CAGR= OOS_START] boot_oos = block_bootstrap(rets_oos, n_boot=5000, block_len=21, seed=43) print("\n OOS-only bootstrap (2021-2026):") print(bootstrap_summary(boot_oos).round(4).to_string()) p_dd_30_oos = float((boot_oos["max_drawdown"] < -0.30).mean()) p_dd_40_oos = float((boot_oos["max_drawdown"] < -0.40).mean()) print(f"\n OOS P(MaxDD<-30%) = {p_dd_30_oos:.3f}") print(f" OOS P(MaxDD<-40%) = {p_dd_40_oos:.3f}") if __name__ == "__main__": main()