"""GATE PORT06: trend_max=3.0/ema_long=200 sulle 6 fade LIVE (improvement-sweep punto 7). Contesto. Il backtest canonico (build_everything) applica il filtro trend alle fade, ma in produzione le SleeveSpec di _defs.py NON lo passano -> le fade live fadano anche i trend/crolli estesi (es. MR01/MR02_ETH long ripetuti nel crash ETH del 2026-06-05). Anche il test PORT06 del loss-guard Hurst (fade_lossguard_port_test) fu misurato su entries GIA' trend-filtrate -> la config live attuale (hurst SENZA trend) non e' mai stata gateata. hurst e trend si sovrappongono (entrambi tagliano il regime trending): il delta marginale va misurato sul path live, non assunto dai numeri canonici. Confronto, a livello PORT06 (stessa matematica pesi cap di Portfolio.backtest): LIVE = fade con hurst_max=0.55 + EXIT-16 (sl_confirm 0.5 ATR), trend OFF [prod oggi] CANDIDATO = LIVE + trend_max=3.0 / ema_long=200 [proposta] TREND-ONLY= EXIT-16 + trend, senza hurst [diagnostica overlap] Parita' preliminare: il replay (mode=orig, trend ON, no hurst, no EXIT-16) deve riprodurre le equity fade canoniche (come exit16_port06_impact). La maschera Hurst e' quella ESATTA del live: fade_base.hurst_skip_mask (close-only, window=100, step=6) — non la cache di regime_lab. GATE (sweep): PROMOSSO se OOS Sharpe non peggiora (>= base-0.02) E il DD scende, e in FULL non degrada. uv run python scripts/analysis/trendmax_port06_impact.py """ from __future__ import annotations import sys from pathlib import Path import numpy as np import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) from src.data.downloader import load_data from src.strategies.fade_base import hurst_skip_mask from scripts.analysis.strategy_research import atr from scripts.analysis.risk_management import strats_for, FEE_RT, LEV, POS, INIT from scripts.analysis.combine_portfolio import ( _norm, IDX, port_returns, metrics, SPLIT, OOS_DATE, ) from scripts.portfolios._defs import PORTFOLIOS from src.portfolio import weighting as W BUFFER = 0.5 # EXIT-16 close-confirm (come in produzione) HURST_MAX = 0.55 # loss-guard live TREND_MAX = 3.0 EMA_LONG = 200 def build_trades_variant(ents, df, mode, trend_max, hurst_mask=None, buffer=BUFFER, lev=LEV, fee_rt=FEE_RT, ema_long=EMA_LONG): """Engine di exit16_port06_impact.build_trades_variant + skip Hurst opzionale. mode="orig" : SL intrabar al livello (SL prima del TP) == canonico. mode="exit16" : SL intrabar OFF; TP intrabar al livello (priorita' nel bar); SL solo se il CLOSE sfonda sl0 -/+ buffer*ATR14[j], fill a close[j]. trend_max : None = filtro OFF (live attuale); 3.0 = candidato. hurst_mask : bool[i]=True -> salta l'ingresso (loss-guard live). """ h, l, c = df["high"].values, df["low"].values, df["close"].values n = len(c) a = atr(df, 14) el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values fee = fee_rt * lev out = [] last = -1 for e in ents: i, d = e["i"], e["d"] if i <= last or i + 1 >= n: continue if hurst_mask is not None and hurst_mask[i]: continue if trend_max is not None and a[i] and abs(c[i] - el[i]) / a[i] > trend_max: continue entry = c[i] tp, sl0, mb = e["tp"], e["sl"], e["max_bars"] exit_p = c[min(i + mb, n - 1)] j = min(i + mb, n - 1) for k in range(1, mb + 1): j = i + k if j >= n: exit_p = c[n - 1] break if mode == "orig": hs = (d == 1 and l[j] <= sl0) or (d == -1 and h[j] >= sl0) ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp) if hs: exit_p = sl0 break if ht: exit_p = tp break if k == mb: exit_p = c[j] else: # exit16 ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp) if ht: exit_p = tp break aj = a[j] if np.isfinite(a[j]) else 0.0 confirm = (d == 1 and c[j] < sl0 - buffer * aj) or \ (d == -1 and c[j] > sl0 + buffer * aj) if confirm: exit_p = c[j] break if k == mb: exit_p = c[j] ret = (exit_p - entry) / entry * d * lev - fee out.append((i, j, ret)) last = j return out def equity_from_trades(df, trades): ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) n = len(df) eq = np.full(n, INIT, dtype=float) cap = INIT for i, j, ret in sorted(trades, key=lambda t: t[1]): cap = max(cap + cap * POS * ret, 10.0) eq[j:] = cap s = pd.Series(eq, index=ts).resample("1D").last().reindex(IDX).ffill().bfill() return _norm(s) def port_metrics(members: dict[str, pd.Series], p): ids = p.sleeve_ids dr = pd.DataFrame({i: members[i].pct_change().fillna(0.0) for i in ids}) w = W.weight_vector(p.weighting, ids, dr, weights=p.weights, caps=p.caps, clusters=p.clusters, lookback=p.vol_lookback) drp = port_returns({i: members[i] for i in ids}, w) return metrics(drp), metrics(drp, lo=SPLIT) def _dd(s): pk = s.cummax() return float(((pk - s) / pk).max() * 100) def main(): p = PORTFOLIOS["PORT06"] fade_ids = [s.sid for s in p.sleeves if s.sid.startswith("MR")] print("=" * 100) print(" GATE PORT06 — trend_max=3.0/ema200 sulle fade LIVE (hurst 0.55 + EXIT-16 gia' attivi)") print(f" fade sleeve: {fade_ids} | caps={p.caps}") print("=" * 100) print("\n[1] build_everything() canonico (cache)...") from src.portfolio.sleeves import all_sleeve_equities eq_base = dict(all_sleeve_equities()) print(f" sleeve totali: {len(eq_base)}") # dati + entries + maschera hurst (identica al live: close-only, w=100, step=6) dfs, masks, entries = {}, {}, {} for asset in ("BTC", "ETH"): dfs[asset] = load_data(asset, "1h") masks[asset] = hurst_skip_mask(dfs[asset], HURST_MAX) print(f" {asset}: {len(dfs[asset])} barre, hurst-skip {masks[asset].mean()*100:.1f}% delle barre") for nm, (fn, params) in strats_for(asset).items(): sid = f"{nm}_{asset}" if sid in fade_ids: entries[sid] = (asset, fn(dfs[asset], **params)) # --- [2] PARITA': mode=orig, trend ON, no hurst, no exit16 == canonico --- print("\n[2] PARITA' replay (orig, trend ON, no hurst) vs canonico:") print(f" {'sleeve':<10s}{'corr':>10s}{'ret_canon%':>14s}{'ret_replay%':>14s}{'diff%':>9s}") parity_ok = True for sid in fade_ids: asset, ents = entries[sid] rep = equity_from_trades(dfs[asset], build_trades_variant( ents, dfs[asset], mode="orig", trend_max=TREND_MAX)) base = eq_base[sid] corr = base.pct_change().fillna(0).corr(rep.pct_change().fillna(0)) rb = (base.iloc[-1] / base.iloc[0] - 1) * 100 rr = (rep.iloc[-1] / rep.iloc[0] - 1) * 100 flag = "" if (corr > 0.999 and abs(rr - rb) <= max(1.0, abs(rb) * 0.01)) else " <-- MISMATCH" if flag: parity_ok = False print(f" {sid:<10s}{corr:>10.5f}{rb:>14.1f}{rr:>14.1f}{rr-rb:>+9.2f}{flag}") print(f"\n PARITA' {'OK' if parity_ok else 'FALLITA'}.") if not parity_ok: print(" >>> STOP: non procedo col gate su un engine non in parita'.") return # --- [3] varianti live-path delle 6 fade --- VARIANTS = { "LIVE": dict(trend_max=None, hurst=True), # produzione oggi "CANDIDATO": dict(trend_max=TREND_MAX, hurst=True), # + trend filter "TREND-ONLY": dict(trend_max=TREND_MAX, hurst=False), # swap hurst->trend "NESSUNO": dict(trend_max=None, hurst=False), # solo EXIT-16 (baseline filtri) "TREND-2.5": dict(trend_max=2.5, hurst=False), # sensibilita' soglia "TREND-3.5": dict(trend_max=3.5, hurst=False), } eqs = {v: {} for v in VARIANTS} ntr = {v: {} for v in VARIANTS} for sid in fade_ids: asset, ents = entries[sid] for v, cfg in VARIANTS.items(): tr = build_trades_variant(ents, dfs[asset], mode="exit16", trend_max=cfg["trend_max"], hurst_mask=masks[asset] if cfg["hurst"] else None) eqs[v][sid] = equity_from_trades(dfs[asset], tr) ntr[v][sid] = len(tr) # --- [4] PORT06 per variante --- print("\n" + "=" * 100) print(f" [4] PORT06 (pesi cap, OOS da {OOS_DATE}) — fade in path LIVE (exit16+hurst)") print("=" * 100) print(f" {'variante':<12s}{'FULL Sh':>9s}{'FULL DD%':>10s}{'FULL CAGR':>11s}" f" | {'OOS Sh':>8s}{'OOS DD%':>9s}{'OOS CAGR':>10s}") print(" " + "-" * 94) res = {} for v in VARIANTS: members = dict(eq_base) for sid in fade_ids: members[sid] = eqs[v][sid] f, o = port_metrics(members, p) res[v] = (f, o) print(f" {v:<12s}{f['sharpe']:>9.2f}{f['dd']:>10.2f}{f['cagr']:>10.0f}%" f" | {o['sharpe']:>8.2f}{o['dd']:>9.2f}{o['cagr']:>9.0f}%") f_l, o_l = res["LIVE"] f_c, o_c = res["CANDIDATO"] print(" " + "-" * 94) print(f" {'DELTA C-L':<12s}{f_c['sharpe']-f_l['sharpe']:>+9.2f}{f_c['dd']-f_l['dd']:>+10.2f}" f"{f_c['cagr']-f_l['cagr']:>+10.0f}% | {o_c['sharpe']-o_l['sharpe']:>+8.2f}" f"{o_c['dd']-o_l['dd']:>+9.2f}{o_c['cagr']-o_l['cagr']:>+9.0f}%") # --- [5] per-sleeve --- print("\n Per-sleeve (FULL): ret% | DD% | n trade [LIVE -> CANDIDATO]") print(f" {'sleeve':<10s}{'ret L%':>10s}{'ret C%':>10s}{'DD L%':>8s}{'DD C%':>8s}" f"{'ntr L':>7s}{'ntr C':>7s}") for sid in fade_ids: el_, ec = eqs["LIVE"][sid], eqs["CANDIDATO"][sid] rl = (el_.iloc[-1] / el_.iloc[0] - 1) * 100 rc = (ec.iloc[-1] / ec.iloc[0] - 1) * 100 print(f" {sid:<10s}{rl:>10.1f}{rc:>10.1f}{_dd(el_):>8.1f}{_dd(ec):>8.1f}" f"{ntr['LIVE'][sid]:>7d}{ntr['CANDIDATO'][sid]:>7d}") # --- GATE --- print("\n" + "=" * 100) print(" GATE (sweep): PROMOSSO se OOS Sharpe >= LIVE-0.02 E OOS DD scende, e FULL non degrada") print("=" * 100) oos_sh_ok = o_c["sharpe"] >= o_l["sharpe"] - 0.02 oos_dd_ok = o_c["dd"] < o_l["dd"] full_ok = f_c["sharpe"] >= f_l["sharpe"] - 0.02 and f_c["dd"] <= f_l["dd"] + 0.20 promoted = oos_sh_ok and oos_dd_ok and full_ok print(f" OOS Sharpe {o_l['sharpe']:.2f} -> {o_c['sharpe']:.2f} ({'OK' if oos_sh_ok else 'KO'})") print(f" OOS DD% {o_l['dd']:.2f} -> {o_c['dd']:.2f} ({'OK' if oos_dd_ok else 'KO'})") print(f" FULL Sharpe {f_l['sharpe']:.2f} -> {f_c['sharpe']:.2f} | " f"FULL DD {f_l['dd']:.2f} -> {f_c['dd']:.2f} ({'OK' if full_ok else 'KO'})") print("\n VERDETTO: " + (">>> PROMOSSO <<<" if promoted else ">>> BOCCIATO <<<")) if __name__ == "__main__": main()