From d733196564eef34fa4a439ab1c97c2173fc114d4 Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Thu, 18 Jun 2026 12:29:25 +0000 Subject: [PATCH] =?UTF-8?q?feat(analysis):=20ladder=5Fsearch=20=E2=80=94?= =?UTF-8?q?=20harness=20caccia=20Price=20Ladder=20che=20passi=20il=20gate?= =?UTF-8?q?=20PORT06?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Goal "decine di agenti a cercare strategie Price Ladder" (Deribit). Riusa il motore grid_mtm mark-to-market ONESTO (SL gap-aware, flat-skip, fee 0.10% RT taker = CONSERVATIVO su Deribit, dove i fill ai livelli LIMIT sono maker ~0%) ed espone: - eval - scan (sub-griglia struttura, gate PORT06, baseline cachata) -> top celle con verdetto gate + max_corr coi 19 sleeve + FULL DD. Leva NUOVA: regime-gate `range` (deploy_mask in grid_mtm, retro-compatibile) = deploya la griglia SOLO in regime di range (|close-EMA200|/ATR < trend_max), dove la griglia vive, e non in trend dove muore. Contesto della ricerca: la griglia ETH del gioco e' BOCCIATA (ridondante, corr 0.40); i ladder BTC sono meno correlati (~0.18) e passano il gate, MA il nodo e' la FULL DD (coda di trend 2021/22, ~60% standalone) che il verdetto del gate NON controlla -> la harness la espone (full_dd standalone + full_full_dd di portafoglio). grid_mtm: aggiunto param deploy_mask (None = comportamento storico, parita' col gate). Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/analysis/grid_game_gate.py | 9 +- scripts/analysis/ladder_search.py | 191 +++++++++++++++++++++++++++++ 2 files changed, 199 insertions(+), 1 deletion(-) create mode 100644 scripts/analysis/ladder_search.py diff --git a/scripts/analysis/grid_game_gate.py b/scripts/analysis/grid_game_gate.py index 9e7c694..63ed7a0 100644 --- a/scripts/analysis/grid_game_gate.py +++ b/scripts/analysis/grid_game_gate.py @@ -68,11 +68,15 @@ def _load(asset, tf): 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): + close_only=False, deploy_mask=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 = _load(asset, tf) op = df["open"].to_numpy(float) @@ -106,6 +110,9 @@ def grid_mtm(asset="ETH", *, tf, range_down, range_up, levels, sl_buf, tp_buf, 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) diff --git a/scripts/analysis/ladder_search.py b/scripts/analysis/ladder_search.py new file mode 100644 index 0000000..aac3b4d --- /dev/null +++ b/scripts/analysis/ladder_search.py @@ -0,0 +1,191 @@ +"""LADDER SEARCH — harness per la caccia multi-agente a strategie Price Ladder (griglia). + +Goal 2026-06-18 (branch price_ladder_research): "decine di agenti a cercare strategie +Price Ladder". CONTESTO: il gioco "Grid Traders" trovo' gia' una griglia ETH asimmetrica +fortissima standalone (FULL Sharpe 5.61, OOS 4.21, plateau 16/16) ma BOCCIATA al gate +PORT06 -- ridondante con le fade ETH (corr +0.40 con MR02_ETH): full-size alza FULL ma +abbassa OOS 10.86->10.47. Quindi la ricerca NON e' "trovare un edge griglia" (gia' fatto) +ma trovarne uno che PASSI IL GATE = aggiunga DIVERSIFICAZIONE. Leve nuove: + - ASSET diverso da ETH (BTC: meno ridondante con la reversione ETH); + - REGIME-GATE: deployare la griglia SOLO in regime di range (non trend) -- il doc + STRATEGIA_GRIGLIA.md dice che la griglia muore in trend; gateare i deploy concentra + l'edge dove funziona E riduce la correlazione coi trend-follower; + - STRUTTURA: livelli, range asimmetrico, sl/tp buffer, max_bars, tf. + +Motore = grid_mtm (mark-to-market ONESTO: SL gap-aware, flat-skip, fee 0.10% RT) di +grid_game_gate.py, esteso con deploy_mask per il regime-gate (retro-compatibile). +Tutto NETTO, OOS held-out, leva 3x. Il giudizio che CONTA e' il gate PORT06. + +CLI (JSON): + uv run python scripts/analysis/ladder_search.py eval ETH 15m 0.171 0.046 4 0.124 0.048 2143 + uv run python scripts/analysis/ladder_search.py eval BTC 1h 0.13 0.05 4 0.12 0.05 1500 range 2.0 +""" +from __future__ import annotations + +import json +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.grid_game_gate import grid_mtm, std, _load +from scripts.analysis.combine_portfolio import port_returns, metrics, SPLIT, IDX +from scripts.portfolios._defs import PORTFOLIOS +from src.portfolio import weighting as W + +_BASE = None + + +def _baseline(): + global _BASE + if _BASE is None: + from src.portfolio.sleeves import all_sleeve_equities + _BASE = dict(all_sleeve_equities()) + return _BASE + + +def _atr(df, n=14): + h, l, c = df["high"].to_numpy(float), df["low"].to_numpy(float), df["close"].to_numpy(float) + pc = np.roll(c, 1); pc[0] = c[0] + tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc))) + return pd.Series(tr).rolling(n).mean().to_numpy() + + +def regime_mask(asset, tf, ema_n=200, trend_max=2.0): + """Mask CAUSALE 'range-bound' allineata alle righe di _load(asset,tf): + True dove |close - EMA(ema_n)| / ATR(14) < trend_max (prezzo vicino al trend = + regime di range -> la griglia puo' deployare). Decisione a close[j] con dati <= j.""" + df = _load(asset, tf) + c = df["close"].to_numpy(float) + ema = pd.Series(c).ewm(span=ema_n, adjust=False).mean().to_numpy() + a = _atr(df, 14) + with np.errstate(invalid="ignore", divide="ignore"): + dist = np.abs(c - ema) / np.where(a == 0, np.nan, a) + m = dist < trend_max + m[~np.isfinite(dist)] = False # warmup / ATR nan -> niente deploy + return m + + +def _gate(grid_eq): + """Gate PORT06 (stesso metodo di grid_game_gate): baseline vs +LADDER full/half. + Ritorna metriche base/full/half + max corr coi 19 sleeve (segnale ridondanza).""" + p = PORTFOLIOS["PORT06"] + base = _baseline() + + def port_m(extra): + members = dict(base); ids = list(p.sleeve_ids) + if extra is not None: + members["LADDER"] = extra; ids = ids + ["LADDER"] + 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) + + fb, ob = port_m(None) + gr = grid_eq.reindex(IDX).ffill().bfill().pct_change().fillna(0.0) + maxcorr = max(abs(gr.corr(e.reindex(IDX).ffill().bfill().pct_change().fillna(0.0))) + for e in base.values()) + half = (1 + 0.5 * gr).cumprod() + ff, of = port_m(grid_eq) + fh, oh = port_m(half) + + def ok(f, o): + return bool(o["sharpe"] >= ob["sharpe"] - 0.02 and o["dd"] <= ob["dd"] + 1e-9 + and f["sharpe"] >= fb["sharpe"] - 0.02) + return { + "base_full_sh": round(fb["sharpe"], 2), "base_full_dd": round(fb["dd"], 2), + "base_oos_sh": round(ob["sharpe"], 2), "base_oos_dd": round(ob["dd"], 2), + "full_oos_sh": round(of["sharpe"], 2), "full_oos_dd": round(of["dd"], 2), + "full_full_sh": round(ff["sharpe"], 2), "full_full_dd": round(ff["dd"], 2), + "half_oos_sh": round(oh["sharpe"], 2), "half_oos_dd": round(oh["dd"], 2), + "half_full_sh": round(fh["sharpe"], 2), "half_full_dd": round(fh["dd"], 2), + "max_corr_existing": round(float(maxcorr), 3), + "verdict_full": "PROMOSSO" if ok(ff, of) else "bocciato", + "verdict_half": "PROMOSSO" if ok(fh, oh) else "bocciato", + } + + +def evaluate(asset, tf, rd, ru, levels, sl_buf, tp_buf, max_bars, + regime="none", trend_max=2.0, ema_n=200, do_gate=True, do_fee2x=True, + flat_skip=True, close_only=False) -> dict: + mask = regime_mask(asset, tf, ema_n=ema_n, trend_max=trend_max) if regime == "range" else None + cfg = dict(tf=tf, range_down=rd, range_up=ru, levels=levels, + sl_buf=sl_buf, tp_buf=tp_buf, max_bars=max_bars) + try: + eqd, st = grid_mtm(asset, **cfg, deploy_mask=mask, flat_skip=flat_skip, close_only=close_only) + except ValueError as e: + return {"asset": asset, "tf": tf, "regime": regime, "error": str(e)} + f, o = std(eqd) + row = { + "asset": asset, "tf": tf, "regime": regime, "trend_max": trend_max, + "rd": rd, "ru": ru, "levels": levels, "sl_buf": sl_buf, "tp_buf": tp_buf, "max_bars": max_bars, + "trades": st["trades"], "win": round(st["win"], 1), "stops": st["stops"], + "full_sh": round(f["sharpe"], 2), "full_dd": round(f["dd"], 2), "full_ret": round(f["ret"], 0), + "oos_sh": round(o["sharpe"], 2), "oos_dd": round(o["dd"], 2), + } + if do_fee2x: + eq2, _ = grid_mtm(asset, **cfg, fee_side=0.001, deploy_mask=mask, flat_skip=flat_skip) + row["fee2x_oos_sh"] = round(std(eq2)[1]["sharpe"], 2) + if do_gate: + row.update(_gate(eqd)) + return row + + +# griglia di struttura coarse per lo scan (range asimmetrico favorito, come il vincitore) +SCAN_RD = [0.08, 0.12, 0.16, 0.20] +SCAN_RU = [0.04, 0.06] +SCAN_LEVELS = [3, 4, 6] +SCAN_SLB = [0.12] +SCAN_TP = 0.05 +MAXBARS_TF = {"15m": 2880, "30m": 1440, "1h": 720} # ~30 giorni di episodio + + +def scan(asset, tf, regime="none", trend_max=2.0, top=10) -> list: + """Sweep coarse della struttura (rd x ru x levels) col gate PORT06, baseline + cachata (una load per processo). Ritorna le top-`top` celle per qualita' di gate.""" + mb = MAXBARS_TF.get(tf, 720) + rows = [] + for rd in SCAN_RD: + for ru in SCAN_RU: + for lv in SCAN_LEVELS: + for slb in SCAN_SLB: + r = evaluate(asset, tf, rd, ru, lv, slb, SCAN_TP, mb, + regime=regime, trend_max=trend_max, + do_gate=True, do_fee2x=False) + if "error" not in r: + rows.append(r) + # score: PROMOSSO half/full premiati; poi OOS migliorato col candidato; penalita' FULL DD del portafoglio + def score(r): + promo = (r.get("verdict_half") == "PROMOSSO") + (r.get("verdict_full") == "PROMOSSO") + return (promo, r.get("half_oos_sh", 0) - 0.1 * r.get("full_full_dd", 99)) + rows.sort(key=score, reverse=True) + return rows[:top] + + +def main(): + a = sys.argv + if len(a) >= 2 and a[1] == "scan": + asset, tf = a[2], a[3] + regime = a[4] if len(a) > 4 else "none" + trend_max = float(a[5]) if len(a) > 5 else 2.0 + print(json.dumps(scan(asset, tf, regime=regime, trend_max=trend_max))) + return + if len(a) < 2 or a[1] != "eval": + print(__doc__); return + asset, tf = a[2], a[3] + rd, ru = float(a[4]), float(a[5]) + levels = int(a[6]); sl_buf, tp_buf = float(a[7]), float(a[8]); max_bars = int(a[9]) + regime = a[10] if len(a) > 10 else "none" + trend_max = float(a[11]) if len(a) > 11 else 2.0 + print(json.dumps(evaluate(asset, tf, rd, ru, levels, sl_buf, tp_buf, max_bars, + regime=regime, trend_max=trend_max))) + + +if __name__ == "__main__": + main()