Add 28 research scripts covering DCA simulation, momentum evaluation, Sharpe optimization, trend rider analysis, and US fundamentals exploration.
116 lines
4.2 KiB
Python
116 lines
4.2 KiB
Python
"""Vol-targeting overlay on V5/V6 blends — tests if dynamic exposure scaling
|
|
can lift realized Sharpe past 1.30 toward 1.50+.
|
|
|
|
The vol-target post-processor scales total weights by min(1, target_vol /
|
|
realized_vol_20d) using the strategy's *own* realized 20-day vol from the
|
|
prior backtest output. It shrinks exposure (toward cash) in high-vol
|
|
regimes — same effect as a deleveraging manager.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
|
|
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,
|
|
portfolio_returns,
|
|
)
|
|
from research.trend_rider_v6_eval import load_combined_panel
|
|
from strategies.permanent import ETF_UNIVERSE
|
|
from strategies.trend_rider_v5 import TrendRiderV5
|
|
from strategies.trend_rider_v6 import TrendRiderV6
|
|
|
|
|
|
IS_START = "2015-01-02"
|
|
IS_END = "2020-12-31"
|
|
OOS_START = "2021-01-01"
|
|
OOS_END = "2026-05-07"
|
|
|
|
|
|
def _fmt(x):
|
|
return f"{x*100:7.2f}%"
|
|
|
|
|
|
def vol_target_overlay(weights: pd.DataFrame, prices: pd.DataFrame,
|
|
target_vol: float, vol_window: int = 20,
|
|
lookback_lag: int = 1) -> pd.DataFrame:
|
|
"""Scale weights so realized 20-day portfolio vol ≈ target_vol.
|
|
|
|
`lookback_lag` ensures PIT-safety: scaling at row t uses vol estimate
|
|
available at end of row t-1.
|
|
"""
|
|
rets = portfolio_returns(weights, prices, transaction_cost=0.0)
|
|
realized = rets.rolling(vol_window).std(ddof=1) * np.sqrt(252)
|
|
realized = realized.shift(lookback_lag)
|
|
realized = realized.fillna(target_vol) # warmup: no scaling
|
|
scale = (target_vol / realized.replace(0.0, np.nan)).clip(upper=1.0).fillna(1.0)
|
|
out = weights.mul(scale, axis=0)
|
|
return out
|
|
|
|
|
|
def evaluate_blend(name, blend, panel, label_prefix="", txn=0.001):
|
|
rows = []
|
|
for window_name, (s, e) in {"FULL": (IS_START, OOS_END),
|
|
"IS": (IS_START, IS_END),
|
|
"OOS": (OOS_START, OOS_END)}.items():
|
|
ev = evaluate_weights(name, blend, panel[blend.columns], txn, s, e)
|
|
print(f" [{window_name}] {label_prefix}{name:<28s} "
|
|
f"CAGR {_fmt(ev.cagr)} Vol {_fmt(ev.volatility)} "
|
|
f"Sharpe {ev.sharpe:5.2f} MDD {_fmt(ev.max_drawdown)} "
|
|
f"Calmar {ev.calmar:5.2f} X {ev.final_multiple:6.2f}")
|
|
rows.append({"window": window_name, "name": name, **ev.__dict__})
|
|
return rows
|
|
|
|
|
|
def main() -> None:
|
|
panel = load_combined_panel()
|
|
etf_set = (set(ETF_UNIVERSE)
|
|
| {"QQQ", "TQQQ", "UPRO", "GLD", "DBC", "SHY", "SPY",
|
|
"YINN", "CHAU", "7200.HK", "7500.HK"})
|
|
stock_universe = [c for c in panel.columns if c not in etf_set]
|
|
|
|
v5 = TrendRiderV5()
|
|
v6_best = TrendRiderV6(
|
|
signal_name="rec_mfilt+deep_upvol", top_n=10,
|
|
tier2_leverage_overlay=0.50,
|
|
stock_universe=stock_universe,
|
|
)
|
|
v5_w = v5.generate_signals(panel)
|
|
v6_w = v6_best.generate_signals(panel)
|
|
|
|
# Align columns
|
|
cols = sorted(set(v5_w.columns) | set(v6_w.columns))
|
|
v5_a = v5_w.reindex(columns=cols).fillna(0.0)
|
|
v6_a = v6_w.reindex(index=v5_a.index, columns=cols).fillna(0.0)
|
|
|
|
print(f"V5 vs V6 corr = {portfolio_returns(v5_a, panel[cols], 0.001).corr(portfolio_returns(v6_a, panel[cols], 0.001)):.3f}")
|
|
|
|
print("\n=== V5 + V6 blends WITH vol targeting ===")
|
|
blend_ratios = [(0.50, 0.50), (0.70, 0.30), (0.40, 0.60), (0.30, 0.70)]
|
|
targets = [0.20, 0.22, 0.25, 0.30]
|
|
|
|
for w5, w6 in blend_ratios:
|
|
blend = v5_a * w5 + v6_a * w6
|
|
for tgt in targets:
|
|
sized = vol_target_overlay(blend, panel[blend.columns], target_vol=tgt)
|
|
evaluate_blend(f"V5={w5:.0%}+V6={w6:.0%} vt{tgt:.2f}", sized, panel,
|
|
label_prefix="")
|
|
print()
|
|
|
|
# Vol target on pure V5 / V6 too
|
|
print("\n=== Pure strategies WITH vol targeting ===")
|
|
for tgt in targets:
|
|
for nm, w in [("V5", v5_a), ("V6best", v6_a)]:
|
|
sized = vol_target_overlay(w, panel[w.columns], target_vol=tgt)
|
|
evaluate_blend(f"{nm} vt{tgt:.2f}", sized, panel)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|