research: stress-test TP01 — robusto come strategia DIFENSIVA (DD-cut), edge ritorno hold-out sottile
Stress sul modulo integrato: FULL regge fee 0.40% + lag + ampio plateau parametri (orizzonti 20/60/120 fa Sh 1.61, non cherry-pick); deflated-Sharpe DSR 0.999 a N=100 (no multiple-testing artifact). MA il ritorno nel hold-out 2025-26 e' SOTTILE (+2.8%/Sh0.27 a 0.10%, ~flat a 0.40%/lag2): TP01 PROTEGGE il drawdown (8% vs 60% buy&hold) piu' di quanto profitti. Proprieta' robusta e deployabile = taglio DD; alpha = no. Da monitorare col paper trader prima di scalare. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,121 @@
|
||||
"""STRESS-TEST di TP01 (integrato da strategy-research-2026-06) — robustezza avversariale.
|
||||
|
||||
Usa il modulo VERO integrato (src/strategies/trend_portfolio). Oltre a hold-out/cross-asset/multi-TF
|
||||
(gia' in verify_tp01.py), qui: sweep FEE (fino 0.40% RT), LAG di esecuzione + slippage, PLATEAU dei
|
||||
parametri (config cherry-picked?), DEFLATED-SHARPE (multiple-testing track A-E).
|
||||
|
||||
uv run python scripts/analysis/stress_tp01.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import sys
|
||||
from pathlib import Path
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
from scipy.stats import norm, skew, kurtosis
|
||||
from src.data.downloader import load_data
|
||||
from src.strategies.trend_portfolio import TrendPortfolio, resample_4h, simple_returns, CANONICAL
|
||||
|
||||
HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC")
|
||||
DF4H = {a: resample_4h(load_data(a, "1h")) for a in ("BTC", "ETH")}
|
||||
|
||||
|
||||
def combo(cfg, lag_bars=0, fee_side=0.0005):
|
||||
"""Rendimenti per-barra del portafoglio 50/50 con config cfg, lag extra e fee dati."""
|
||||
tp = TrendPortfolio(**{**cfg, "fee_side": fee_side})
|
||||
series = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
df = DF4H[a]
|
||||
r = simple_returns(df["close"].values.astype(float))
|
||||
tgt = tp.target_series(df)
|
||||
held = np.zeros(len(tgt))
|
||||
s = 1 + lag_bars
|
||||
held[s:] = tgt[:-s] # tenuta = decisa s barre prima (causale + lag)
|
||||
net = held * r - fee_side * np.abs(np.diff(held, prepend=0.0))
|
||||
net[0] = 0.0
|
||||
series[a] = pd.Series(np.clip(net, -0.99, None), index=pd.to_datetime(df["datetime"]))
|
||||
J = pd.concat(series, axis=1, join="inner").fillna(0.0)
|
||||
return 0.5 * J["BTC"].values + 0.5 * J["ETH"].values, J.index
|
||||
|
||||
|
||||
def met(combo_r, idx):
|
||||
rr = combo_r[np.isfinite(combo_r)]
|
||||
if len(rr) < 2 or np.std(rr) == 0:
|
||||
return dict(sh=0, ret=0, dd=0)
|
||||
bpy = 86400 * 365.25 / pd.Series(idx).diff().dt.total_seconds().median()
|
||||
eq = np.cumprod(1 + rr); pk = np.maximum.accumulate(eq)
|
||||
return dict(sh=float(np.mean(rr) / np.std(rr) * np.sqrt(bpy)),
|
||||
ret=float(eq[-1] - 1), dd=float(np.max((pk - eq) / pk)))
|
||||
|
||||
|
||||
def full_ho(cfg, lag_bars=0, fee_side=0.0005):
|
||||
cr, idx = combo(cfg, lag_bars, fee_side)
|
||||
ho = idx >= HOLDOUT
|
||||
return met(cr, idx), met(cr[ho], idx[ho])
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 88)
|
||||
print(" STRESS-TEST TP01 (PORT LF4h canonica) — robustezza avversariale")
|
||||
print("=" * 88)
|
||||
|
||||
base_f, base_h = full_ho(CANONICAL)
|
||||
print(f"\n BASELINE (4h, fee 0.10% RT): FULL Sh {base_f['sh']:.2f} ret {base_f['ret']*100:+.0f}% DD {base_f['dd']*100:.1f}%"
|
||||
f" | HOLD-OUT Sh {base_h['sh']:.2f} ret {base_h['ret']*100:+.1f}% DD {base_h['dd']*100:.1f}%")
|
||||
|
||||
print("\n (1) SWEEP FEE (RT) — regge fino a 0.40%?")
|
||||
print(f" {'fee RT':<10s}{'FULL Sh':>9s}{'FULL ret':>10s}{'HOLD Sh':>9s}{'HOLD ret':>10s}")
|
||||
for frt in (0.0, 0.001, 0.002, 0.004):
|
||||
f, h = full_ho(CANONICAL, fee_side=frt / 2)
|
||||
print(f" {frt*100:>5.2f}% {f['sh']:>8.2f}{f['ret']*100:>+9.0f}%{h['sh']:>9.2f}{h['ret']*100:>+9.1f}%")
|
||||
|
||||
print("\n (2) LAG di esecuzione + slippage (fee 0.20% per simulare slippage)")
|
||||
print(f" {'scenario':<22s}{'FULL Sh':>9s}{'HOLD Sh':>9s}{'HOLD ret':>10s}")
|
||||
for name, lag, frt in [("base", 0, 0.001), ("lag 1 barra (4h)", 1, 0.001),
|
||||
("lag 2 barre", 2, 0.001), ("lag1 + fee0.20% slip", 1, 0.002)]:
|
||||
f, h = full_ho(CANONICAL, lag_bars=lag, fee_side=frt / 2)
|
||||
print(f" {name:<22s}{f['sh']:>8.2f}{h['sh']:>9.2f}{h['ret']*100:>+9.1f}%")
|
||||
|
||||
print("\n (3) PLATEAU PARAMETRI — la config canonica e' un picco o un altopiano?")
|
||||
print(f" {'variazione':<26s}{'FULL Sh':>9s}{'HOLD Sh':>9s}")
|
||||
grid = [
|
||||
("canonica (vt.20 lev2 30/90/180 vw30)", CANONICAL),
|
||||
("target_vol 0.15", {**CANONICAL, "target_vol": 0.15}),
|
||||
("target_vol 0.25", {**CANONICAL, "target_vol": 0.25}),
|
||||
("leverage 1.5", {**CANONICAL, "leverage": 1.5}),
|
||||
("leverage 3.0", {**CANONICAL, "leverage": 3.0}),
|
||||
("horizons 20/60/120", {**CANONICAL, "horizons_days": (20, 60, 120)}),
|
||||
("horizons 60/120/240", {**CANONICAL, "horizons_days": (60, 120, 240)}),
|
||||
("vol_win 20", {**CANONICAL, "vol_win_days": 20}),
|
||||
("vol_win 45", {**CANONICAL, "vol_win_days": 45}),
|
||||
]
|
||||
sr_trials = []
|
||||
for name, cfg in grid:
|
||||
f, h = full_ho(cfg)
|
||||
cr, idx = combo(cfg)
|
||||
sr_trials.append(cr[np.isfinite(cr)].mean() / cr[np.isfinite(cr)].std()) # Sharpe per-barra
|
||||
print(f" {name:<26s}{f['sh']:>8.2f}{h['sh']:>9.2f}")
|
||||
|
||||
print("\n (4) DEFLATED SHARPE — corregge il multiple-testing (track A-E + sweep). DSR>0.95 = regge")
|
||||
cr, idx = combo(CANONICAL)
|
||||
rr = cr[np.isfinite(cr)]
|
||||
sr = rr.mean() / rr.std(); T = len(rr)
|
||||
g3 = float(skew(rr)); g4 = float(kurtosis(rr, fisher=False))
|
||||
var_sr = float(np.var(sr_trials, ddof=1))
|
||||
ge = 0.5772156649
|
||||
for N in (10, 40, 100): # N = numero di trial/config provati (conservativo)
|
||||
z1 = norm.ppf(1 - 1.0 / N); z2 = norm.ppf(1 - 1.0 / (N * np.e))
|
||||
sr0 = np.sqrt(var_sr) * ((1 - ge) * z1 + ge * z2)
|
||||
den = np.sqrt(max(1 - g3 * sr + (g4 - 1) / 4.0 * sr ** 2, 1e-9))
|
||||
dsr = float(norm.cdf((sr - sr0) * np.sqrt(T - 1) / den))
|
||||
bpy = 86400 * 365.25 / pd.Series(idx).diff().dt.total_seconds().median()
|
||||
print(f" N={N:>3d} trial -> soglia-max-attesa Sh {sr0*np.sqrt(bpy):.2f} | DSR {dsr:.3f} [{'REGGE' if dsr>0.95 else 'NON regge'}]")
|
||||
|
||||
print("\n" + "=" * 88)
|
||||
print(" Verdetto: TP01 robusto se regge fee 0.40%+lag (HOLD positivo), plateau (no picco), DSR>0.95.")
|
||||
print("=" * 88)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user