#!/usr/bin/env python3 """r0822_ortho_screen.py — ONDATA 2026-08-22, filone ORTHO-SCREEN (la rete larga, onesta). NON e' un'ipotesi: e' un SETACCIO. Sette famiglie di segnale mai (o solo parzialmente) coperte dalle ondate precedenti, passate TUTTE dallo stesso protocollo, senza eccezioni: 1. famiglia DICHIARATA prima (parametri + celle), trial contati AL RIALZO; 2. A.study_family_honest -> cella scelta IN-SAMPLE-ONLY + deflated-Sharpe di famiglia; 3. A.marginal_vs_tp01 -> ADDS / NEUTRAL / DILUTES / HEDGE / NOISE / REDUNDANT; 4. A.causality_ok (+ A.day_boundary_robust dove ha senso, vedi NOTA CONFINE); 5. banda d'ancora su 8 offset orari (bar 1d ricostruite da 1h) -> la stima onesta e' la MEDIANA; 6. null del DE-LEVERING su ogni claim di drawdown; 7. A.eval_weights_smallcap($600) + A.implausible_sharpe sulla cella scelta; 8. **DSR di SCREEN**: oltre al deflated-Sharpe di famiglia, il deflated-Sharpe calcolato sul POOL di TUTTE le celle di TUTTE le famiglie. Se scelgo il migliore fra 7 famiglie, i trial sono 168, non 24: e' l'unico conto onesto per un setaccio. PRIOR DICHIARATO PRIMA DI MISURARE: il soffitto direzionale BTC/ETH misurato e' Sharpe ~1.3 e in 104 ipotesi + 153 agenti (giugno 2026) NESSUNA e' sopravvissuta alla verifica avversariale. Mi aspetto 0 CANDIDATI. Il valore consegnato e' **quale gate uccide cosa**, perche' questo restringe lo spazio della prossima ondata. LE 7 FAMIGLIE (24 celle ciascuna = 168 trial dichiarati) RSKEW skew realizzata (da 1h) come PREDITTORE direzionale, non come gate [nuovo vs blind/agent_40_skewgate, che la usava come FILTRO su un trend] TACC accelerazione del trend (derivata seconda del log-prezzo) come direzione standalone VPX relazione volume-prezzo su barre 1d (conferma / divergenza a volume alto e basso) XDISP dispersione cross-sezionale dei 51 alt HL usata come segnale TIME-SERIES sul mercato [nuovo vs XS01, dove la dispersione e' un gate sul paniere cross-sectional] XCORR correlazione realizzata BTC-ETH come STATO di mercato direzionale [nuovo vs ortho/agent_09 e XAS08, dove la corr gata un book relative-value] XTAIL struttura dei ritorni dopo uno shock a k-sigma (continuazione vs inversione) VRAT variance-ratio (persistenza) come CONVINZIONE continua sul TSMOM [nuovo vs blind/agent_37_hurst, che faceva uno switch BINARIO di modo] NOTA CONFINE (perche' day_boundary_robust non gira su tutte): il test sposta le ETICHETTE del calendario lasciando i prezzi fermi. Per un segnale di solo prezzo (TACC/VPX/XTAIL/VRAT) e' esattamente il controllo giusto e deve dare INVARIANT. Per un segnale che fa merge_asof su una serie AUSILIARIA (RSKEW da 1h, XDISP da HL, XCORR da 1h BTC/ETH) spostare l'etichetta in AVANTI sposta il punto di merge nel FUTURO: sarebbe un LEAK indotto dallo strumento, non un controllo. Per quelle famiglie il test d'ancora corretto e' la RICOSTRUZIONE delle barre a offset orario (banda d'ancora, punto 5), che gira su tutte e sette. USO: nice -n 19 timeout 900 uv run python scripts/research/r0822_ortho_screen.py """ from __future__ import annotations import os import sys import time from functools import lru_cache import numpy as np import pandas as pd sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "alt")) import altlib as A # noqa: E402 HL_DIR = A.DATA_DIR ANCHOR_OFFSETS = (0, 3, 6, 9, 12, 15, 18, 21) W_BLEND = 0.25 # peso del candidato nel blend con TP01 (convenzione altlib) # =========================================================================== # INFRASTRUTTURA — barre a offset, merge causale di serie ausiliarie # =========================================================================== @lru_cache(maxsize=64) def bars_at(asset: str, off: int) -> pd.DataFrame: """Barre 1d ancorate all'ora `off`, ricostruite dal feed 1h certificato. off=0 deve riprodurre esattamente A.get(asset,'1d') (controllo in _sanity).""" h = A.get(asset, "1h").copy() idx = pd.to_datetime(h["timestamp"], unit="ms", utc=True) - pd.Timedelta(hours=off) h.index = idx o = (h.resample("1D", label="left", closed="left") .agg({"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}) .dropna(subset=["open"])) o["datetime"] = o.index + pd.Timedelta(hours=off) epoch = pd.Timestamp("1970-01-01", tz="UTC") o["timestamp"] = ((o["datetime"] - epoch) // pd.Timedelta(milliseconds=1)).astype("int64") return o.reset_index(drop=True)[["timestamp", "open", "high", "low", "close", "volume", "datetime"]] def _bar_ms(df: pd.DataFrame) -> int: d = pd.Series(df["timestamp"].astype("int64")).diff().median() return int(d) if np.isfinite(d) and d > 0 else 86_400_000 def merge_aux(df: pd.DataFrame, aux_ts: np.ndarray, aux_val: np.ndarray) -> np.ndarray: """Valore di una serie ausiliaria NOTO alla CHIUSURA di ogni barra di df. `aux_ts` e' gia' il timestamp in cui il valore diventa noto (fine della sua barra). merge_asof backward sul close della barra -> mai un dato del futuro: eval_weights tiene target[i] durante la barra i+1, quindi close[i] e' l'ultimo istante lecito.""" close_ts = df["timestamp"].astype("int64").values + _bar_ms(df) left = pd.DataFrame({"timestamp": close_ts}) right = (pd.DataFrame({"timestamp": np.asarray(aux_ts, dtype="int64"), "v": np.asarray(aux_val, dtype=float)}) .dropna().sort_values("timestamp").reset_index(drop=True)) if right.empty: return np.full(len(df), np.nan) return pd.merge_asof(left, right, on="timestamp", direction="backward")["v"].values def roll_z(x: np.ndarray, win: int) -> np.ndarray: s = pd.Series(np.asarray(x, float)) mp = max(20, win // 4) m = s.rolling(win, min_periods=mp).mean() sd = s.rolling(win, min_periods=mp).std() return ((s - m) / sd.where(sd > 0)).values def exp_pct(x: np.ndarray, min_obs: int = 180) -> np.ndarray: """Percentile ESPANDENTE causale (rank fra tutti i valori visti finora, incluso l'attuale).""" return pd.Series(np.asarray(x, float)).expanding(min_periods=min_obs).rank(pct=True).values def _shape(u: np.ndarray, shape: str) -> np.ndarray: u = np.nan_to_num(np.asarray(u, float), nan=0.0) return np.sign(u) if shape == "sign" else np.clip(u, -1.0, 1.0) def _sized(direction: np.ndarray, df: pd.DataFrame) -> np.ndarray: return A.vol_target(np.nan_to_num(direction, nan=0.0), df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0) # =========================================================================== # SERIE AUSILIARIE (calcolate una volta, riusate da tutte le celle) # =========================================================================== @lru_cache(maxsize=16) def skew_aux(asset: str, w_days: int): """Skew realizzata dei ritorni 1h su finestra trascinata di w_days giorni. Valore al bar 1h i -> noto alla sua chiusura (ts_i + 1h).""" h = A.get(asset, "1h") r = A.log_returns(h["close"].values.astype(float)) win = w_days * 24 sk = pd.Series(r).rolling(win, min_periods=win // 2).skew().values return h["timestamp"].astype("int64").values + 3_600_000, sk @lru_cache(maxsize=4) def disp_aux(): """Dispersione cross-sezionale dei 51 alt Hyperliquid (1d, 2024+): deviazione standard cross-section dei log-ritorni giornalieri. Valore del giorno d noto alla chiusura di d.""" import glob files = sorted(glob.glob(str(HL_DIR / "hl_*_1d.parquet"))) cols = {} for f in files: x = pd.read_parquet(f, columns=["timestamp", "close"]) s = pd.Series(x["close"].values.astype(float), index=x["timestamp"].astype("int64").values) cols[os.path.basename(f)] = np.log(s).diff() M = pd.DataFrame(cols).sort_index() n_ok = M.notna().sum(axis=1) disp = M.std(axis=1, ddof=1).where(n_ok >= 15) ts = M.index.values.astype("int64") + 86_400_000 # noto a fine giornata return ts, disp.values, int(len(files)) @lru_cache(maxsize=16) def xcorr_aux(w_days: int): """Correlazione realizzata BTC-ETH sui ritorni 1h, finestra trascinata w_days giorni.""" b = A.get("BTC", "1h")[["timestamp", "close"]].rename(columns={"close": "b"}) e = A.get("ETH", "1h")[["timestamp", "close"]].rename(columns={"close": "e"}) J = pd.merge(b, e, on="timestamp", how="inner").sort_values("timestamp") rb = pd.Series(A.log_returns(J["b"].values.astype(float))) re = pd.Series(A.log_returns(J["e"].values.astype(float))) win = w_days * 24 c = rb.rolling(win, min_periods=win // 2).corr(re).values return J["timestamp"].astype("int64").values + 3_600_000, c # =========================================================================== # LE 7 FAMIGLIE — factory(tf=..., **params) -> target(df, asset) # =========================================================================== def mk_rskew(tf="1d", w_days=30, z_days=365, shape="lin", sgn=-1): """RSKEW: la skew realizzata recente come PREDITTORE del ritorno successivo. A-priori (Amaya et al., azioni): skew realizzata alta -> ritorno futuro BASSO (sgn=-1).""" def target(df, asset): ts, sk = skew_aux(asset, w_days) v = merge_aux(df, ts, sk) u = np.clip(roll_z(v, z_days) / 2.0, -1.0, 1.0) return _sized(sgn * _shape(u, shape), df) return target def mk_tacc(tf="1d", H=60, L=20, shape="lin", sgn=1): """TACC: accelerazione = variazione della PENDENZA log-prezzo su H barre, misurata a L barre di distanza. Direzione standalone (non e' il LIVELLO del trend). A-priori sgn=+1.""" def target(df, asset): c = np.log(df["close"].values.astype(float)) n = len(c) slope = np.full(n, np.nan) slope[H:] = (c[H:] - c[:-H]) / H acc = np.full(n, np.nan) acc[L:] = slope[L:] - slope[:-L] u = np.clip(roll_z(acc, 365) / 2.0, -1.0, 1.0) return _sized(sgn * _shape(u, shape), df) return target def mk_vpx(tf="1d", W=30, k=5, mode="confirm", thr=0.5): """VPX: volume-prezzo su barre 1d (a 1d il volume Deribit e' informativo, a 5m no). confirm -> segui il movimento SOLO se il volume conferma (z >= thr) fade_low -> fai il contrario del movimento avvenuto a volume BASSO (z <= -thr) fade_high -> fai il contrario del movimento avvenuto a volume ALTO (climax, z >= thr)""" def target(df, asset): c = df["close"].values.astype(float) vol = df["volume"].values.astype(float) n = len(c) vz = roll_z(np.log(np.where(vol > 0, vol, np.nan)), W) rk = np.full(n, np.nan) rk[k:] = c[k:] / c[:-k] - 1.0 s = np.sign(np.nan_to_num(rk, nan=0.0)) vzf = np.nan_to_num(vz, nan=0.0) if mode == "confirm": d = np.where(vzf >= thr, s, 0.0) elif mode == "fade_low": d = np.where(vzf <= -thr, -s, 0.0) else: # fade_high d = np.where(vzf >= thr, -s, 0.0) return _sized(d, df) return target def mk_xdisp(tf="1d", S=5, trans="z252", shape="lin", sgn=1): """XDISP: dispersione cross-sezionale dei 51 alt HL come STATO del mercato (time-series), non come gate di un paniere cross-sectional. Nessun prior forte sul segno -> entrambi in griglia.""" def target(df, asset): ts, dsp, _ = disp_aux() sm = pd.Series(dsp).rolling(S, min_periods=1).mean().values u = (np.clip(roll_z(sm, 252) / 2.0, -1, 1) if trans == "z252" else 2.0 * exp_pct(sm, 180) - 1.0) v = merge_aux(df, ts, u) return _sized(sgn * _shape(v, shape), df) return target def mk_xcorr(tf="1d", W=60, trans="z252", shape="lin", sgn=-1): """XCORR: co-movimento BTC-ETH (corr realizzata) come stato direzionale del mercato. A-priori debole: corr alta = 'tutto il crypto e' un trade solo' = regime tardo -> sgn=-1.""" def target(df, asset): ts, cr = xcorr_aux(W) u_aux = (np.clip(roll_z(cr, 252 * 24) / 2.0, -1, 1) if trans == "z252" else 2.0 * exp_pct(cr, 180 * 24) - 1.0) v = merge_aux(df, ts, u_aux) return _sized(sgn * _shape(v, shape), df) return target def mk_xtail(tf="1d", ksig=3.0, N=3, sgn=1, sigwin=30): """XTAIL: struttura dei ritorni ESTREMI. Dopo |r| > ksig*sigma trascurata, tieni per N barre la direzione (sgn=+1 continuazione) o il suo opposto (sgn=-1 inversione). ATTENZIONE dichiarata: sgn=-1 e' mean-reversion, famiglia gia' MORTA -> se vince li', si chiude.""" def target(df, asset): c = df["close"].values.astype(float) r = A.simple_returns(c) sd = pd.Series(r).rolling(sigwin, min_periods=sigwin // 2).std().shift(1).values n = len(c) d = np.zeros(n) hit = np.where(np.isfinite(sd) & (sd > 0) & (np.abs(r) > ksig * sd))[0] for i in hit: d[i:min(n, i + N)] = sgn * np.sign(r[i]) return _sized(d, df) return target def mk_vrat(tf="1d", q=10, LB=365, conv="gate", lf=True): """VRAT: variance-ratio VR(q)=Var(r_q)/(q*Var(r_1)) su finestra LB come CONVINZIONE CONTINUA sul TSMOM (non uno switch binario di modo come blind/agent_37_hurst). gate -> pesa il trend solo dove VR>1 (persistente), altrimenti flat flip -> inverte il trend dove VR<1 (anti-persistente)""" HZ = (30, 90, 180) def target(df, asset): c = df["close"].values.astype(float) lc = np.log(c) n = len(c) r1 = pd.Series(A.log_returns(c)) rq = pd.Series(np.concatenate([np.full(q, np.nan), lc[q:] - lc[:-q]])) v1 = r1.rolling(LB, min_periods=LB // 2).var() vq = rq.rolling(LB, min_periods=LB // 2).var() vr = (vq / (q * v1.where(v1 > 0))).values dirs = np.zeros(n) for H in HZ: m = np.zeros(n) m[H:] = np.sign(c[H:] / c[:-H] - 1.0) dirs += m / len(HZ) if lf: dirs = np.maximum(dirs, 0.0) k = np.nan_to_num(vr, nan=1.0) - 1.0 w = np.clip(k * 5.0, 0.0, 1.0) if conv == "gate" else np.clip(k * 5.0, -1.0, 1.0) return _sized(dirs * w, df) return target # ---- griglie DICHIARATE (24 celle ciascuna, 7 famiglie = 168 trial) ---------- def _grid(**axes): keys = list(axes) out = [{}] for k in keys: out = [dict(o, **{k: v}) for o in out for v in axes[k]] return out FAMILIES = [ dict(code="RSKEW", factory=mk_rskew, price_only=False, grid=_grid(w_days=[14, 30, 60], z_days=[180, 365], shape=["lin", "sign"], sgn=[1, -1])), dict(code="TACC", factory=mk_tacc, price_only=True, grid=_grid(H=[20, 60, 120], L=[5, 20], shape=["lin", "sign"], sgn=[1, -1])), dict(code="VPX", factory=mk_vpx, price_only=True, grid=_grid(W=[30, 90], k=[5, 20], mode=["confirm", "fade_low", "fade_high"], thr=[0.5, 1.0])), dict(code="XDISP", factory=mk_xdisp, price_only=False, grid=_grid(S=[1, 5, 20], trans=["z252", "pct"], shape=["lin", "sign"], sgn=[1, -1])), dict(code="XCORR", factory=mk_xcorr, price_only=False, grid=_grid(W=[20, 60, 120], trans=["z252", "pct"], shape=["lin", "sign"], sgn=[1, -1])), dict(code="XTAIL", factory=mk_xtail, price_only=True, grid=_grid(ksig=[2.5, 3.0], N=[1, 3, 10], sgn=[1, -1], sigwin=[30, 90])), dict(code="VRAT", factory=mk_vrat, price_only=True, grid=_grid(q=[5, 10, 20], LB=[180, 365], conv=["gate", "flip"], lf=[True, False])), ] # =========================================================================== # GATE AGGIUNTIVI # =========================================================================== def cand_daily_at(fn, off: int) -> pd.Series: """Serie giornaliera 50/50 BTC+ETH del candidato con barre ancorate all'ora `off`.""" ser = {} for a in A.CERTIFIED: df = bars_at(a, off) ev = A.eval_weights(df, A._call_target(fn, df, a)) ser[a] = pd.Series(ev["net"], index=ev["idx"]) J = pd.concat(ser, axis=1, join="inner").fillna(0.0) return A._to_daily(0.5 * J[A.CERTIFIED[0]] + 0.5 * J[A.CERTIFIED[1]]) def anchor_report(fn) -> dict: """Banda d'ancora su 8 offset orari: Sharpe standalone E uplift del blend. La stima ONESTA e' la MEDIANA della banda, non l'ancora 00:00 (lezione 02/07).""" band = A.anchor_luck_band(lambda o: cand_daily_at(fn, o), ANCHOR_OFFSETS) B = A.tp01_baseline_daily() ups = {} for o in ANCHOR_OFFSETS: C = cand_daily_at(fn, o) J = pd.concat({"B": B, "C": C}, axis=1, join="inner").dropna() if len(J) > 30: ups[o] = round(A._sh((1 - W_BLEND) * J["B"] + W_BLEND * J["C"]) - A._sh(J["B"]), 3) vals = np.array(list(ups.values()), float) if ups else np.array([np.nan]) band["uplift_per_offset"] = ups band["uplift_canonical"] = ups.get(0) band["uplift_median"] = round(float(np.median(vals)), 3) band["uplift_frac_positive"] = round(float((vals > 0).mean()), 3) return band def delever_null(C: pd.Series, w: float = W_BLEND) -> dict: """NULL DEL DE-LEVERING (obbligatorio su ogni claim di meno drawdown, 5 occorrenze nel progetto): esiste k<1 che, applicato al SOLO TP01, da' lo STESSO maxDD con Sharpe MIGLIORE? Se si', l'overlay non serve: bastava meno leva.""" B = A.tp01_baseline_daily() J = pd.concat({"B": B, "C": C}, axis=1, join="inner").dropna() if len(J) < 60: return dict(run=False, reason="overlap insufficiente") blend = (1 - w) * J["B"] + w * J["C"] dd_bl, sh_bl = A._dd_ret(blend), A._sh(blend) dd_b, sh_b = A._dd_ret(J["B"]), A._sh(J["B"]) if dd_bl >= dd_b: return dict(run=True, claims_less_dd=False, dd_blend=round(dd_bl, 4), dd_tp01=round(dd_b, 4), sharpe_blend=round(sh_bl, 3), sharpe_tp01=round(sh_b, 3), note="nessun claim di DD: il blend NON riduce il drawdown -> null non pertinente") ks = np.linspace(0.05, 1.0, 96) dds = np.array([A._dd_ret(k * J["B"]) for k in ks]) k = float(ks[int(np.argmin(np.abs(dds - dd_bl)))]) sh_k = A._sh(k * J["B"]) return dict(run=True, claims_less_dd=True, dd_blend=round(dd_bl, 4), dd_tp01=round(dd_b, 4), sharpe_blend=round(sh_bl, 3), k_equal_dd=round(k, 3), dd_k=round(A._dd_ret(k * J["B"]), 4), sharpe_k_tp01=round(sh_k, 3), passes=bool(sh_bl > sh_k), note=("il blend batte il de-levering" if sh_bl > sh_k else "REFUTED: k*TP01 da' lo stesso DD con Sharpe migliore")) def smallcap_report(fn) -> dict: out = {} for a in A.CERTIFIED: df = A.get(a, "1d") sc = A.eval_weights_smallcap(df, A._call_target(fn, df, a), capital=600.0, min_order=5.0) out[a] = dict(modeled=sc["modeled"]["sharpe"], realistic=sc["realistic"]["sharpe"], haircut=sc["sharpe_haircut"], n_trades=sc["n_executed_trades"]) return out def active_window_metrics(C: pd.Series) -> dict: """Trappola #9 (barre attive vs calendario): una serie che e' 0 per meta' della storia (XDISP parte nel 2024) ha Sharpe DILUITO di sqrt(frazione attiva). Si riporta anche il numero sulla sola finestra ATTIVA, senza usarlo per decidere.""" nz = C[C != 0.0] if len(nz) < 30: return dict(active_days=int(len(nz))) first = nz.index[0] sub = C[C.index >= first] return dict(active_days=int(len(nz)), frac_active=round(float(len(nz) / max(1, len(C))), 3), first_active=str(first.date()), sharpe_active_window=round(A._sh(sub), 3), maxdd_active_window=round(A._dd_ret(sub), 4)) # =========================================================================== # SANITY — un setaccio che non si controlla non sta setacciando niente # =========================================================================== def sanity() -> None: print("### SANITY") for a in A.CERTIFIED: b0, g = bars_at(a, 0), A.get(a, "1d") n = min(len(b0), len(g)) d = float(np.max(np.abs(b0["close"].values[:n] - g["close"].values[:n]))) dts = float(np.max(np.abs(b0["timestamp"].values[:n] - g["timestamp"].values[:n]))) print(f" bars_at({a},0) vs A.get({a},'1d'): n={n}/{len(g)} max|dclose|={d:.6g} max|dts|={dts:.0f}") assert d < 1e-9 and dts == 0, "ricostruzione barre a offset 0 NON identica al feed certificato" ts, dsp, nfiles = disp_aux() ok = np.isfinite(dsp) print(f" disp_aux: {nfiles} file HL, {int(ok.sum())} giorni con >=15 asset, " f"da {pd.Timestamp(ts[ok][0], unit='ms', tz='UTC').date()} " f"a {pd.Timestamp(ts[ok][-1], unit='ms', tz='UTC').date()}") B = A.tp01_baseline_daily() print(f" TP01 baseline: n={len(B)} ShFULL={A._sh(B):.3f} ShHOLD={A._sh(B[B.index >= A.HOLDOUT]):.3f}") print() # =========================================================================== # MAIN # =========================================================================== def main() -> None: t0 = time.time() print(__doc__.split("USO:")[0]) sanity() pooled_sh: list[float] = [] results = [] for fam in FAMILIES: code, fac, grid = fam["code"], fam["factory"], fam["grid"] t1 = time.time() print(f"### {code} — {len(grid)} celle dichiarate") sel = A.select_cell_insample(fac, grid, ("1d",)) pooled_sh.extend([s for s in sel["all_full_sharpe"] if np.isfinite(s)]) rep = A.study_family_honest(code, fac, grid, ("1d",)) ch = rep["chosen"] if ch is None: print(f" nessuna cella valida ({rep.get('reason')})\n") results.append(dict(code=code, n_cells=len(grid), chosen=None, killed_by="nessuna cella valutabile")) continue fn = fac(tf=ch["tf"], **ch["params"]) C = A.candidate_daily(fn, tf=ch["tf"]) m = rep["marginal"]["marginal"] absr = rep["marginal"]["absolute"]["cells"][0] caus = A.causality_ok(fn, tf="1d") dayb = A.day_boundary_robust(fn, tf="1d") if fam["price_only"] else \ dict(verdict="NON GIRATO", reason="segnale con merge su serie ausiliaria: " "spostare l'etichetta in avanti sposta il merge nel FUTURO (leak indotto), " "non e' un controllo -> vale la banda d'ancora") anch = anchor_report(fn) deln = delever_null(C) smal = smallcap_report(fn) impl = A.implausible_sharpe(C) actw = active_window_metrics(C) full_sh = A._sh(C) hold_sh = A._sh(C[C.index >= A.HOLDOUT]) mm = A._metrics_from_net(C.values, C.index) results.append(dict( code=code, n_cells=len(grid), chosen=ch, fn=fn, C=C, full_sharpe=round(full_sh, 3), hold_sharpe=round(hold_sh, 3), maxdd=mm["maxdd"], cagr=mm["cagr"], corr=m.get("corr_full"), corr_hold=m.get("corr_hold"), verdict=rep["marginal"]["marginal_verdict"], earns_slot=rep["earns_slot_marginal"], honest=rep["earns_slot_honest"], dsr_family=rep["deflated_sharpe"], dsr_pass=rep["dsr_pass"], insample_sharpe=m.get("cand_insample_sharpe"), has_insample_edge=m.get("has_insample_edge"), robust_oos=m.get("robust_oos"), is_hedge=m.get("is_hedge"), uplift_full=m["blends"]["w25"]["uplift_full"], uplift_hold=m["blends"]["w25"]["uplift_hold"], multicut=m.get("multicut_uplift"), abs_grade=rep["marginal"]["abs_grade"], _m=m, fee_survives=absr.get("fee_survives"), causality=caus, dayb=dayb, anchor=anch, delever=deln, smallcap=smal, implausible=impl, active=actw, )) print(f" cella IN-SAMPLE-ONLY: {ch['params']} (IS Sh {ch['insample_sharpe']}, " f"rango 1/{len(sel['rows'])} in-sample; FULL Sh di quella cella {ch['full_sharpe']})") print(f" standalone FULL {full_sh:+.3f} | HOLD {hold_sh:+.3f} | maxDD {mm['maxdd']:.1%} | " f"CAGR {mm['cagr']:+.2%} | IS {m.get('cand_insample_sharpe')}") print(f" marginale {rep['marginal']['marginal_verdict']:<9} corr->TP01 {m.get('corr_full')} " f"(hold {m.get('corr_hold')}) uplift w25 full {m['blends']['w25']['uplift_full']:+.3f} " f"hold {m['blends']['w25']['uplift_hold']}") print(f" robust_oos={m.get('robust_oos')} insample_edge={m.get('has_insample_edge')} " f"is_hedge={m.get('is_hedge')} abs={rep['marginal']['abs_grade']} " f"fee_survives={absr.get('fee_survives')}") print(f" DSR famiglia {rep['deflated_sharpe']} (atteso null max {rep['expected_null_max']}) " f"pass={rep['dsr_pass']} earns_slot_honest={rep['earns_slot_honest']}") print(f" causality ok={caus['ok']} max_tail_diff={caus['max_tail_diff']}") print(f" confine {dayb.get('verdict')} " + (f"spread {dayb.get('spread')} per_offset {dayb.get('per_offset')}" if fam["price_only"] else f"({dayb.get('reason','')[:60]}...)")) _nan = float("nan") print(f" ancora Sh canonica {anch.get('canonical', _nan):+.3f} " f"(pctl {anch.get('canonical_pctl')}) " f"MEDIANA {anch.get('median', _nan):+.3f} " f"banda [{anch.get('lo', _nan):+.3f},{anch.get('hi', _nan):+.3f}] " f"frac>0 {anch.get('frac_positive')}") print(f" uplift canonico {anch.get('uplift_canonical')} MEDIANA " f"{anch.get('uplift_median')} frac>0 {anch.get('uplift_frac_positive')}") print(f" de-levering {deln.get('note')}") print(f" $600 " + " ".join( f"{a}: mod {v['modeled']:+.2f} -> real {v['realistic']:+.2f} (haircut {v['haircut']:+.2f}, " f"{v['n_trades']} trade)" for a, v in smal.items())) print(f" implausible {impl.get('implausible')} {impl.get('reasons')}") if actw.get("frac_active", 1.0) < 0.9: print(f" attive {actw}") print(f" [{time.time() - t1:.1f}s]\n") # ---- DSR DI SCREEN: i trial sono TUTTE le celle di TUTTE le famiglie ------------- print("### DSR DI SCREEN (il conto onesto per un setaccio: scelgo il meglio fra " f"{len(pooled_sh)} celle, non fra 24)") for r in results: if r.get("chosen") is None: continue d, s0 = A.deflated_sharpe(r["full_sharpe"], pooled_sh, r["C"]) r["dsr_screen"] = round(d, 3) if np.isfinite(d) else None print(f" {r['code']:<6} Sh {r['full_sharpe']:+.3f} DSR famiglia {r['dsr_family']} " f"DSR screen {r['dsr_screen']} (null max atteso {s0:.3f})") print() deep_dive(results) # ---- TABELLA FINALE -------------------------------------------------------------- def killer(r) -> str: if r.get("chosen") is None: return "nessuna cella valutabile" why = [] if not r["causality"]["ok"]: why.append("CAUSALITY") if r["verdict"] != "ADDS": why.append(f"marginal={r['verdict']}") if not r["has_insample_edge"]: why.append("no in-sample edge (<0.5)") if not r["robust_oos"]: why.append("robust_oos") if r["is_hedge"]: why.append("is_hedge") if not r["dsr_pass"]: why.append(f"DSR fam {r['dsr_family']}") if r.get("dsr_screen") is not None and r["dsr_screen"] < 0.95: why.append(f"DSR screen {r['dsr_screen']}") if r["dayb"].get("verdict") == "ARTIFACT-RISK": why.append("confine ARTIFACT-RISK") um = r["anchor"].get("uplift_median") if um is not None and np.isfinite(um) and um <= 0: why.append("ancora: uplift mediano <=0") if r["delever"].get("claims_less_dd") and not r["delever"].get("passes"): why.append("null de-levering") if r["implausible"].get("implausible"): why.append("implausible_sharpe") return " + ".join(why) if why else "-- nessun gate lo uccide --" print("### TABELLA — ORTHO-SCREEN 2026-08-22 (candidato | trial | ShFULL | hold-out | " "corr->TP01 | marginale | DSR fam / screen | esito)") hdr = (f"{'cand':<7}{'trial':>6}{'ShFULL':>8}{'hold':>7}{'maxDD':>8}{'CAGR':>8}" f"{'corr':>7} {'marginale':<10}{'DSRfam':>7}{'DSRscr':>8} esito / gate che l'ha ucciso") print(hdr); print("-" * len(hdr)) for r in results: if r.get("chosen") is None: print(f"{r['code']:<7}{r['n_cells']:>6}{'--':>8}{'--':>7}{'--':>8}{'--':>8}{'--':>7} " f"{'--':<10}{'--':>7}{'--':>8} SCARTATO — {killer(r)}") continue ok = (r["verdict"] == "ADDS" and r["dsr_pass"] and (r.get("dsr_screen") or 0) >= 0.95 and r["causality"]["ok"]) esito = "CANDIDATO" if ok else "SCARTATO" print(f"{r['code']:<7}{r['n_cells']:>6}{r['full_sharpe']:>+8.3f}{r['hold_sharpe']:>+7.2f}" f"{r['maxdd']:>8.1%}{r['cagr']:>+8.1%}{r['corr']:>7.2f} {r['verdict']:<10}" f"{str(r['dsr_family']):>7}{str(r.get('dsr_screen')):>8} {esito} — {killer(r)}") print() print("### FAMIGLIE NON TESTATE (l'elenco di cio' che non ho guardato fa parte del risultato)") for line in NOT_TESTED: print(" - " + line) print(f"\n[totale {time.time() - t0:.1f}s]") def mk_mom(tf="1d", H=5, shape="lin"): """CONTROLLO per TACC: il LIVELLO del momentum a orizzonte H, stessa pipeline (z-score 365g -> clip -> vol_target). Serve a rispondere alla domanda della famiglia: l'ACCELERAZIONE aggiunge qualcosa al LIVELLO, o e' un momentum corto travestito?""" def target(df, asset): c = df["close"].values.astype(float) n = len(c) m = np.full(n, np.nan) m[H:] = c[H:] / c[:-H] - 1.0 u = np.clip(roll_z(m, 365) / 2.0, -1.0, 1.0) return _sized(_shape(u, shape), df) return target MOM_CONTROL_GRID = _grid(H=[5, 20, 60], shape=["lin", "sign"]) # 6 celle di CONTROLLO def deep_dive(results) -> None: """APPROFONDIMENTO sull'unica famiglia arrivata a marginal=ADDS. Due domande, entrambe misurate: (a) QUALE sotto-gate di robust_oos ha fallito (clean-year? jackknife? multi-cut?); (b) l'accelerazione e' distinta dal LIVELLO del momentum corto, o e' quello travestito? -> 6 celle di CONTROLLO dichiarate (H x shape), che si SOMMANO ai 168 trial.""" adds = [r for r in results if r.get("chosen") is not None and r["verdict"] == "ADDS"] if not adds: print("### APPROFONDIMENTO — nessuna famiglia a marginal=ADDS, niente da approfondire\n") return r = max(adds, key=lambda x: x["uplift_hold"] or -9) print(f"### APPROFONDIMENTO — {r['code']} (l'unica a marginal=ADDS)") print(f" perche' robust_oos=False: clean_year_uplift={r['_m'].get('clean_year_uplift')} " f"(serve >0.02) jackknife_min_uplift={r['_m'].get('jackknife_min_uplift')} (serve >0)") print(f" multicut (uplift a ogni taglio d'anno): {r['multicut']}") print(f" uplift TP01-up {r['_m'].get('uplift_tp01_up')} / TP01-down {r['_m'].get('uplift_tp01_down')} " f"| hedge_yearly_corr {r['_m'].get('hedge_yearly_corr')} | alpha/anno {r['_m'].get('alpha_ann')}") print(" --- accelerazione vs LIVELLO del momentum (6 celle di controllo) ---") B = A.tp01_baseline_daily() C = r["C"] best = None for p in MOM_CONTROL_GRID: fnm = mk_mom(tf="1d", **p) Cm = A.candidate_daily(fnm, tf="1d") J = pd.concat({"A": C, "M": Cm}, axis=1, join="inner").dropna() JB = pd.concat({"B": B, "M": Cm}, axis=1, join="inner").dropna() up = A._sh((1 - W_BLEND) * JB["B"] + W_BLEND * JB["M"]) - A._sh(JB["B"]) row = dict(params=p, sharpe=round(A._sh(Cm), 3), hold=round(A._sh(Cm[Cm.index >= A.HOLDOUT]), 3), corr_to_acc=round(float(J["A"].corr(J["M"])), 3), corr_to_tp01=round(float(JB["B"].corr(JB["M"])), 3), uplift_w25_full=round(up, 3)) print(f" MOM {str(p):<28} Sh {row['sharpe']:+.3f} hold {row['hold']:+.3f} " f"corr->{r['code']} {row['corr_to_acc']:+.3f} corr->TP01 {row['corr_to_tp01']:+.3f} " f"uplift {row['uplift_w25_full']:+.3f}") if best is None or row["corr_to_acc"] > best["corr_to_acc"]: best = row print(f" livello piu' vicino all'accelerazione: {best['params']} corr {best['corr_to_acc']:+.3f}") print(" lettura: corr alta col LIVELLO => l'accelerazione non e' una dimensione nuova; " "corr bassa => e' distinta ma resta da spiegare perche' non sopravvive al DSR.\n") NOT_TESTED = [ "catena opzioni / vol term-structure: 3-5 mesi di storia -> non esiste hold-out, al massimo " "un LEAD; e altri agenti dell'ondata ci stanno sopra.", "funding (carry, time-series, cross-sectional): filone dichiarato CHIUSO su 3 lati.", "calendario in ogni forma (weekend 375 trial, expiry, stagionalita', event-clock, " "ora-del-giorno): famiglia dichiarata SATURA.", "mean-reversion pura / fade: morta anche a fee zero. Qui rientra SOLO come branch dichiarato " "di XTAIL e VPX, per misurarla dentro un conditioner nuovo.", "microstruttura 5m/15m: morta per fee, e una sweep su 5m sfonda il budget macchina.", "cross-sectional sui 51 alt HL: territorio XS01, e l'espansione d'universo e' gia' refutata. " "Qui i 51 alt entrano solo come AGGREGATO time-series (XDISP).", "equity / GTAA / MAT01: GTAA01 non e' deployabile (PRIIPs) e MAT01 e' refutato.", "macro regime gate: ridondante col trend (corr 0.989).", "CRT / ICT / Elliott / Fibonacci / Albimarini: 3 ondate di refutazione.", "gamma scalping long-vol e overlay DD su VRP01 (5/5 refutati).", "SOL come terza gamba: decisione dell'operatore del 22/08.", ] if __name__ == "__main__": main()