"""GATE PORT06 โ€” griglia ETH (vincitore gioco "Grid Traders", sessione 3). Il gioco (scripts/games/grid_*, regola STRATEGIA_GRIGLIA.md) ha promosso una griglia geometrica asimmetrica su ETH: range profondo sotto, corto sopra, 4 livelli (passo ~5%), SL catastrofale. Ma il motore del gioco somma i PnL REALIZZATI per trade e NON misura l'equity mark-to-market: l'inventario a tranche dentro un drawdown e' rischio vero che il fitness non vede. Questo gate risponde alla domanda "si puo' inserire?" con il metodo del progetto: [1] STANDALONE mark-to-market (engine MTM dedicato, fill onesti): equity per barra = capitale + inventario valutato al close; fee 0.10% RT (taker; i fill ai livelli sarebbero LIMIT->maker, quindi conservativo); SL gap-aware (gap sotto lo stop -> fill all'open, non al livello); flat-skip (nessun fill sulle candele O=H=L=C di ETH 15m, live-realizable). Metriche FULL/OOS con le stesse funzioni degli altri gate + stress fee 2x. [2] CORRELAZIONE coi 19 sleeve PORT06 (il sospetto: e' la stessa reversione ETH delle fade MR, incassata con inventory risk). [3] ROBUSTEZZA: plateau range_down x range_up attorno al vincitore. [4] GATE PORT06: baseline vs +GRID (full e half size). Promosso solo se OOS Sharpe non peggiora E DD non sale (criterio standard). uv run python scripts/analysis/grid_game_gate.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 scripts.analysis.combine_portfolio import port_returns, metrics, SPLIT, OOS_DATE, IDX from scripts.portfolios._defs import PORTFOLIOS from src.portfolio import weighting as W from src.data.downloader import load_data POS, LEV = 0.15, 3.0 # config canonica sleeve (== build_everything) FEE_SIDE = 0.0005 # 0.05%/lato = 0.10% RT # top-3 del torneo (data/games/grid_result.json) WINNER_15M = dict(tf="15m", range_down=0.171, range_up=0.046, levels=4, sl_buf=0.124, tp_buf=0.048, max_bars=2143) TOP2_30M = dict(tf="30m", range_down=0.158, range_up=0.048, levels=4, sl_buf=0.081, tp_buf=0.044, max_bars=613) TOP3_1H = dict(tf="1h", range_down=0.134, range_up=0.053, levels=4, sl_buf=0.150, tp_buf=0.063, max_bars=562) _RESAMPLE = {"30m": ("15m", "30min")} def _load(asset, tf): if tf in _RESAMPLE: base, rule = _RESAMPLE[tf] d = load_data(asset, base).copy() d["dt"] = pd.to_datetime(d["datetime"]) g = d.set_index("dt").resample(rule).agg( {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}).dropna(subset=["open", "close"]).reset_index() g["datetime"] = g["dt"] return g d = load_data(asset, tf).copy() d["dt"] = pd.to_datetime(d["datetime"]) return d def grid_mtm(asset="ETH", *, tf, range_down, range_up, levels, sl_buf, tp_buf, max_bars, pos=POS, lev=LEV, fee_side=FEE_SIDE, flat_skip=True, close_only=False, deploy_mask=None, df=None): """Griglia STRATEGIA_GRIGLIA.md con contabilita' mark-to-market. Ritorna (equity daily Series base 1.0, stats dict). Causale: deploy sul close, fill dalle barre successive lungo il percorso O->L->H->C / O->H->L->C. `deploy_mask` (opzionale, np.bool array lungo come la serie, causale): se fornito, una NUOVA griglia si deploya SOLO dove mask[j]=True (regime-gate); None = comportamento storico (deploy sempre). Una griglia gia' attiva non viene interrotta dal mask (gestisce il suo episodio fino a SL/TP/timeout). `df` (opzionale): OHLCV gia' caricato (per il feed LIVE del GridWorker); None = carica da _load(asset, tf) (comportamento storico, parita' col gate). """ df = _load(asset, tf) if df is None else df op = df["open"].to_numpy(float) hi = df["high"].to_numpy(float) lo = df["low"].to_numpy(float) cl = df["close"].to_numpy(float) dt = (pd.to_datetime(df["datetime"]) if "datetime" in df.columns else pd.to_datetime(df["timestamp"], unit="ms", utc=True)).to_numpy() n = len(cl) ratio = ((1 + range_up) / (1 - range_down)) ** (1.0 / levels) if ratio - 1 <= 1.5 * 2 * fee_side: # vincolo break-even ยง4 raise ValueError("break-even violato") flat = (op == hi) & (op == lo) & (op == cl) capital = 1.0 eq = np.empty(n) eq[:20] = 1.0 # stato episodio active = False lv = []; filled = []; tn = 0.0; sl = tp = 0.0; ep_end = 0 n_open = 0 trades = wins = stops = 0 deploy_i = -1 def mtm(px): u = 0.0 for k in range(levels): if filled[k]: u += tn * (px / lv[k] - 1.0) return capital + u i = 20 for j in range(20, n): if not active: if deploy_mask is not None and not deploy_mask[j]: eq[j] = capital # regime-gate: niente deploy, resta in cash continue # deploy sul close di j (fill da j+1) px = cl[j] rl_ = px * (1 - range_down) lv = [rl_ * ratio ** k for k in range(levels + 1)] sl = rl_ * (1 - sl_buf) tp = lv[levels] * (1 + tp_buf) filled = [False] * levels n_open = 0 tn = capital * pos * lev / levels # notional per tranche ep_end = j + max_bars active = True deploy_i = j eq[j] = capital continue if flat_skip and flat[j]: eq[j] = mtm(cl[j]) continue cur = cl[j - 1] if close_only: # fill solo su attraversamento del CLOSE (le wick non fillano): # stress anti-spike-print del feed testnet pts = (cl[j],) else: pts = (op[j], lo[j], hi[j], cl[j]) if cl[j] >= op[j] \ else (op[j], hi[j], lo[j], cl[j]) died = False for pi, q in enumerate(pts): q = float(q) if q == cur: continue if q < cur: from bisect import bisect_left k1 = bisect_left(lv, q) k2 = bisect_left(lv, cur) - 1 for k in range(min(k2, levels - 1), max(k1, 0) - 1, -1): if not filled[k]: filled[k] = True n_open += 1 capital -= fee_side * tn # fee ingresso if q <= sl: # STOP: gap all'open -> fill all'open, altrimenti al livello sl fill = q if (pi == 0 and q <= sl) else sl for k in range(levels): if filled[k]: r = fill / lv[k] capital += tn * (r - 1.0) - fee_side * tn * r filled[k] = False trades += 1 stops += 1 n_open = 0 died = True cur = q break else: from bisect import bisect_right m1 = bisect_right(lv, cur) m2 = bisect_right(lv, q) - 1 for m in range(max(m1, 1), min(m2, levels) + 1): k = m - 1 if filled[k]: r = lv[m] / lv[k] capital += tn * (r - 1.0) - fee_side * tn * r filled[k] = False n_open -= 1 trades += 1 wins += 1 if q >= tp: for k in range(levels): if filled[k]: r = tp / lv[k] capital += tn * (r - 1.0) - fee_side * tn * r filled[k] = False trades += 1 wins += 1 n_open = 0 died = True cur = q break cur = q if not died and j >= ep_end: # timeout: liquida al close for k in range(levels): if filled[k]: r = cl[j] / lv[k] capital += tn * (r - 1.0) - fee_side * tn * r filled[k] = False trades += 1 wins += r > 1.0 + 2 * fee_side n_open = 0 died = True if died: active = False eq[j] = capital else: eq[j] = mtm(cl[j]) s = pd.Series(eq, index=pd.DatetimeIndex(dt)).resample("1D").last().dropna() s = s / s.iloc[0] return s, dict(trades=trades, win=100.0 * wins / max(1, trades), stops=stops) def std(eqd): """Metriche FULL/OOS con le funzioni standard del progetto.""" e = eqd.reindex(IDX).ffill().bfill() dr = e.pct_change().fillna(0.0) return metrics(dr), metrics(dr, lo=SPLIT) def main(): p = PORTFOLIOS["PORT06"] print("=" * 100) print(" GATE PORT06 โ€” griglia ETH (vincitore gioco Grid Traders) | " f"pos={POS} lev={LEV} | OOS da {OOS_DATE}") print("=" * 100) # [1] STANDALONE mark-to-market print("\n[1] STANDALONE mark-to-market (fee 0.10% RT, flat-skip, SL gap-aware):") print(f" {'cfg':<22s}{'trd':>7s}{'win%':>6s}{'stops':>6s}{'FULL%':>8s}{'CAGR%':>7s}" f"{'DD%':>7s}{'Shrp':>6s} | {'OOS%':>7s}{'oDD%':>6s}{'oShrp':>6s}") eqs = {} for tag, cfg in [("WINNER 15m", WINNER_15M), ("top2 30m", TOP2_30M), ("top3 1h", TOP3_1H)]: eqd, st = grid_mtm("ETH", **cfg) f, o = std(eqd) eqs[tag] = eqd print(f" {tag:<22s}{st['trades']:>7d}{st['win']:>6.1f}{st['stops']:>6d}" f"{f['ret']:>+8.0f}{f['cagr']:>7.1f}{f['dd']:>7.2f}{f['sharpe']:>6.2f}" f" | {o['ret']:>+7.0f}{o['dd']:>6.2f}{o['sharpe']:>6.2f}") # stress: fee 2x e no flat-skip sul winner eq2, st2 = grid_mtm("ETH", **WINNER_15M, fee_side=0.001) f2, o2 = std(eq2) eqnf, _ = grid_mtm("ETH", **WINNER_15M, flat_skip=False) fnf, onf = std(eqnf) print(f" {'winner fee 2x':<22s}{st2['trades']:>7d}{st2['win']:>6.1f}{'':>6s}" f"{f2['ret']:>+8.0f}{f2['cagr']:>7.1f}{f2['dd']:>7.2f}{f2['sharpe']:>6.2f}" f" | {o2['ret']:>+7.0f}{o2['dd']:>6.2f}{o2['sharpe']:>6.2f}") print(f" {'winner no-flat-skip':<22s}{'':>7s}{'':>6s}{'':>6s}" f"{fnf['ret']:>+8.0f}{fnf['cagr']:>7.1f}{fnf['dd']:>7.2f}{fnf['sharpe']:>6.2f}" f" | {onf['ret']:>+7.0f}{onf['dd']:>6.2f}{onf['sharpe']:>6.2f}") grid_eq = eqs["WINNER 15m"] # [2] CORRELAZIONI coi sleeve PORT06 from src.portfolio.sleeves import all_sleeve_equities eq_base = dict(all_sleeve_equities()) gr = grid_eq.reindex(IDX).ffill().bfill().pct_change().fillna(0.0) print("\n[2] CORRELAZIONE rendimenti giornalieri GRID_ETH15M vs sleeve PORT06 (top 8):") cors = {} for sid, e in eq_base.items(): r = e.reindex(IDX).ffill().bfill().pct_change().fillna(0.0) cors[sid] = gr.corr(r) for sid, cv in sorted(cors.items(), key=lambda kv: -abs(kv[1]))[:8]: print(f" {sid:<16s} {cv:+.3f}") # [3] ROBUSTEZZA: plateau range_down x range_up (15m, levels=4) print("\n[3] ROBUSTEZZA 15m (Sharpe FULL mark-to-market, levels=4, " "sl_buf=0.12, tp_buf=0.05, max_bars=2000):") rds = [0.13, 0.15, 0.17, 0.19] rus = [0.04, 0.05, 0.06, 0.08] print(" rd\\ru " + "".join(f"{ru:>7.2f}" for ru in rus)) cells = tot = 0 for rd in rds: row = f" {rd:>5.2f} " for ru in rus: eqd, _ = grid_mtm("ETH", tf="15m", range_down=rd, range_up=ru, levels=4, sl_buf=0.12, tp_buf=0.05, max_bars=2000) f, o = std(eqd) tot += 1 cells += (f["sharpe"] > 1) and (o["sharpe"] > 1) row += f"{f['sharpe']:>7.2f}" print(row) print(f" -> {cells}/{tot} celle con Sharpe>1 sia FULL che OOS") # [4] GATE PORT06 print("\n[4] GATE PORT06 โ€” baseline vs +GRID_ETH15M (full/half size):") def port_m(extra=None): members = dict(eq_base) ids = list(p.sleeve_ids) if extra is not None: members["GRID_ETH15M"] = extra ids = ids + ["GRID_ETH15M"] dr = pd.DataFrame({i: members[i].reindex(IDX).ffill().bfill() .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].reindex(IDX).ffill().bfill() for i in ids}, w) return metrics(drp), metrics(drp, lo=SPLIT) half = (1 + 0.5 * gr).cumprod() res = {"baseline": port_m(None), "+GRID full": port_m(grid_eq), "+GRID half": port_m(half)} print(f" {'variante':<12s} | {'FULL Sh':>8s}{'FULL DD%':>9s}{'CAGR%':>7s}" f" | {'OOS Sh':>7s}{'OOS DD%':>8s}") for tag, (f, o) in res.items(): print(f" {tag:<12s} | {f['sharpe']:>8.2f}{f['dd']:>9.2f}{f['cagr']:>7.1f}" f" | {o['sharpe']:>7.2f}{o['dd']:>8.2f}") fb, ob = res["baseline"] print("\n" + "=" * 100) print(" VERDETTO (criterio standard: OOS Sharpe non peggiora E DD non sale)") print("=" * 100) for tag in ("+GRID full", "+GRID half"): f, o = res[tag] ok = (o["sharpe"] >= ob["sharpe"] - 0.02 and o["dd"] <= ob["dd"] + 1e-9 and f["sharpe"] >= fb["sharpe"] - 0.02) print(f" {tag:<12s}: OOS Sh {ob['sharpe']:.2f}->{o['sharpe']:.2f} " f"DD {ob['dd']:.2f}->{o['dd']:.2f} | FULL Sh {fb['sharpe']:.2f}->{f['sharpe']:.2f} " f"DD {fb['dd']:.2f}->{f['dd']:.2f} => {'PROMOSSO' if ok else 'bocciato'}") if __name__ == "__main__": main()