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