feat(analysis): ladder_search — harness caccia Price Ladder che passi il gate PORT06

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  <asset tf rd ru levels sl_buf tp_buf max_bars [regime] [trend_max]>
  - scan  <asset tf [regime] [trend_max]>  (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) <noreply@anthropic.com>

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)

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()