PythagorasGoal/scripts/research/r0726_dvolspread_gate.py

#!/usr/bin/env python
"""r0726_dvolspread_gate — T2 dell'ondata 2026-07-26: chiudere il lead DVOLSPREAD, in limbo
dal 2026-06-21.

CHI E'. `agent_14_dvol_spread` (onda ortho, 2026-06-21) e' **l'unico sopravvissuto** al marginal
scorer indurito: dei 18 book relative-value che facevano "ADDS" con lo scorer vecchio, l'indurito
ne ha lasciato in piedi **1**, questo — "l'unico con edge in-sample reale". Fu lasciato in
forward-monitor con due riserve dichiarate: *multiple-testing* e *storia DVOL corta*. Da allora
non e' mai stato ripreso: non e' nel book, non e' in un paper monitor, non e' stato rifiutato.
E' semplicemente rimasto li'.

PERCHE' RIPRENDERLO ORA. Perche' nel frattempo il progetto ha codificato **due gate che il 21/06
non esistevano**, e sono esattamente quelli che colpiscono la sua riserva:

  1. **`deflated_sharpe`** (29/06). Il docstring dell'agente dichiara *da se'* uno **sweep di 72
     celle** su ZWIN/LW/ZW/TGT/SVW. 72 trial sono precisamente cio' che il DSR sconta. Nel giugno
     "72/72 celle ADDS" veniva letto come prova di robustezza; il DSR dice che e' anche 72 tentativi.
  2. **`select_cell_insample` / `study_family_honest`** (29/06). Il plateau dell'agente e' descritto
     in termini di **`uplift_hold`** ("ogni vicino resta ADDS con uplift_hold in [0.34,0.39]"),
     cioe' la cella e' stata guardata SULL'HOLD-OUT. E' la firma che il gate SELECTION-ON-HOLDOUT
     e' stato scritto per catturare — lo stesso difetto che uccise il lead ERM intraday.

DOMANDA SECCA: scegliendo la cella con il solo in-sample, e scontando i 72 trial, DVOLSPREAD
resta uno sleeve o era un falso positivo sopravvissuto solo perche' i gate non c'erano ancora?

Terza verifica, indipendente dalle prime due: la **storia DVOL** parte 2021-03, quindi il book e'
FLAT per meta' del campione. Si misura quanto e' corto il campione ATTIVO (non quello nominale),
perche' e' su quello che vanno letti tutti i numeri.

Uso: `uv run python scripts/research/r0726_dvolspread_gate.py`
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt"))
sys.path.insert(0, str(ROOT / "scripts" / "research" / "ortho"))

import altlib as al  # noqa: E402
import ortholib as ol  # noqa: E402

# griglia DICHIARATA dall'agente stesso (docstring PLATEAU): 72 celle
GRID = dict(zwin=[120, 150, 180], tanh_k=[1.5, 1.8, 2.0],
            lw=[0.3, 0.45, 0.6], zw=[0.9, 1.0, 1.1],
            tgt=[0.11, 0.13, 0.17], svw=[40, 50, 60])
CHOSEN = dict(zwin=150, tanh_k=1.8, lw=0.45, zw=1.0, tgt=0.13, svw=50)
LEG_CAP = 0.5


def make_book(zwin, tanh_k, lw, zw, tgt, svw):
    """Fabbrica il book dell'agente a parametri arbitrari (stessa formula, nessuna variante)."""
    def _book(btc, eth):
        import warnings
        bc = btc["close"].values.astype(float)
        ec = eth["close"].values.astype(float)
        rb = ol.simple_returns(bc)
        re = ol.simple_returns(ec)
        db = al.dvol(btc, "BTC")
        de = al.dvol(eth, "ETH")
        with np.errstate(invalid="ignore", divide="ignore"):
            ldvr = np.log(db) - np.log(de)
        ldvr = np.where(np.isfinite(ldvr), ldvr, np.nan)
        level = np.tanh(tanh_k * np.nan_to_num(ldvr, nan=0.0))
        zdev = np.tanh(tanh_k * np.nan_to_num(ol.zscore(ldvr, zwin), nan=0.0))
        g_dir = np.where(np.isfinite(ldvr), lw * level + zw * zdev, 0.0)
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", category=RuntimeWarning)
            spvol = ol.realized_vol(re - rb, svw, 365.25)
        scal = np.where((spvol > 0) & np.isfinite(spvol), tgt / spvol, 0.0)
        g = np.nan_to_num(np.clip(g_dir * scal, -LEG_CAP, LEG_CAP), nan=0.0)
        return g, -g
    return _book


_EVAL_CACHE: dict = {}


def evaluate(**kw) -> dict:
    """Valutazione col MOTORE ORIGINALE dell'agente (ortholib.eval_book): stesso shift, stesse
    fee su entrambe le gambe, stessa serie daily. Riusarlo — invece di riscriverlo — e' cio' che
    rende questo un RE-TEST dello stesso oggetto e non di una sua approssimazione."""
    key = tuple(sorted(kw.items()))
    if key not in _EVAL_CACHE:
        _EVAL_CACHE[key] = ol.eval_book(make_book(**kw))
    return _EVAL_CACHE[key]


def daily_returns(**kw) -> pd.Series:
    return evaluate(**kw)["daily"].dropna()


def sh(s: pd.Series, lo=None, hi=None) -> float:
    x = s.dropna()
    if lo is not None:
        x = x[x.index >= lo]
    if hi is not None:
        x = x[x.index < hi]
    return float(x.mean() / x.std() * np.sqrt(365.25)) if len(x) > 30 and x.std() > 0 else 0.0


HOLDOUT = "2025-01-01"


def main() -> None:
    print("=" * 100)
    print("  T2 — DVOLSPREAD: il lead del 21/06 passa i gate del 29/06?")
    print("=" * 100)

    base = daily_returns(**CHOSEN)
    act = base[base != 0.0]
    print(f"\n  campione NOMINALE: {len(base)} giorni  ({base.index[0].date()} -> {base.index[-1].date()})")
    print(f"  campione ATTIVO  : {len(act)} giorni  ({len(act)/len(base):.0%}) — "
          f"prima del 2021-03 non c'e' DVOL e il book e' FLAT")
    print(f"  attivo pre-holdout {len(act[act.index < HOLDOUT])} g | "
          f"attivo hold-out {len(act[act.index >= HOLDOUT])} g "
          f"({len(act[act.index >= HOLDOUT])/365:.1f} anni)")

    # ------------------------------------------------------- griglia dichiarata dall'agente
    print("\n" + "-" * 100)
    print("  A. LA GRIGLIA CHE L'AGENTE DICHIARA DA SE' (72 celle: ZWIN x K x LW x ZW x TGT x SVW)")
    print("-" * 100)
    import itertools
    keys = list(GRID)
    cells = []
    for combo in itertools.product(*[GRID[k] for k in keys]):
        kw = dict(zip(keys, combo))
        r = daily_returns(**kw)
        cells.append(dict(kw=kw, full=sh(r), is_=sh(r, hi=HOLDOUT), hold=sh(r, lo=HOLDOUT)))
    df = pd.DataFrame([{**c["kw"], "full": c["full"], "is": c["is_"], "hold": c["hold"]}
                       for c in cells])
    print(f"  celle valutate: {len(df)}")
    print(f"  Sharpe FULL : min {df['full'].min():.2f} | mediana {df['full'].median():.2f} | "
          f"max {df['full'].max():.2f}")
    print(f"  Sharpe HOLD : min {df['hold'].min():.2f} | mediana {df['hold'].median():.2f} | "
          f"max {df['hold'].max():.2f}")
    print(f"  celle con HOLD > 0: {int((df['hold'] > 0).sum())}/{len(df)}")

    # ------------------------------------------------------- selezione IN-SAMPLE vs hold-out
    print("\n" + "-" * 100)
    print("  B. SELECTION-ON-HOLDOUT — la cella scelta in-sample e' la stessa scelta sull'hold-out?")
    print("-" * 100)
    i_is = int(df["is"].idxmax())
    i_ho = int(df["hold"].idxmax())
    i_ch = int(df[[*CHOSEN]].eq(pd.Series(CHOSEN)).all(axis=1).idxmax())
    for lab, i in (("cella scelta IN-SAMPLE (onesta)", i_is),
                   ("cella scelta sull'HOLD-OUT (proibita)", i_ho),
                   ("cella PUBBLICATA dall'agente", i_ch)):
        row = df.loc[i]
        pars = {k: row[k] for k in CHOSEN}
        print(f"  {lab:>40}: FULL {row['full']:>5.2f}  IS {row['is']:>5.2f}  "
              f"HOLD {row['hold']:>5.2f}   {pars}")
    rank_ch_hold = int((df["hold"] > df.loc[i_ch, "hold"]).sum()) + 1
    rank_ch_is = int((df["is"] > df.loc[i_ch, "is"]).sum()) + 1
    print(f"\n  rango della cella PUBBLICATA: {rank_ch_hold}/{len(df)} sull'HOLD-OUT, "
          f"{rank_ch_is}/{len(df)} sull'IN-SAMPLE")
    print("  (se e' molto meglio classificata sull'hold-out che in-sample -> selezione-sull'hold-out)")

    # ------------------------------------------------------- deflated Sharpe sui 72 trial
    print("\n" + "-" * 100)
    print("  C. DEFLATED SHARPE sui trial DICHIARATI DALL'AGENTE STESSO")
    print("-" * 100)
    r_is = daily_returns(**{k: df.loc[i_is, k] for k in CHOSEN})
    for lab, i, r in (("cella IN-SAMPLE", i_is, r_is),
                      ("cella PUBBLICATA", i_ch, base)):
        dsr, sr0 = al.deflated_sharpe(sh(r), list(df["full"].values), r.dropna().values)
        verdict = "PASS" if (np.isfinite(dsr) and dsr >= 0.95) else "FAIL"
        print(f"  {lab:>20}: Sharpe FULL {sh(r):>5.2f} | max atteso sotto il null "
              f"{sr0:>5.2f} | DSR {dsr:>6.3f}  -> {verdict}")

    # ------------------------------------------------------- marginale vs TP01
    print("\n" + "-" * 100)
    print("  D. MARGINALE vs TP01 (scorer indurito) — sulla cella scelta IN-SAMPLE")
    print("-" * 100)
    tp = al.tp01_baseline_daily()
    for lab, r, kwc in (("cella IN-SAMPLE", r_is, {k: df.loc[i_is, k] for k in CHOSEN}),
                        ("cella PUBBLICATA", base, CHOSEN)):
        J = pd.concat({"tp": tp, "c": r}, axis=1, join="inner").fillna(0.0)
        corr = float(J["tp"].corr(J["c"]))
        for w in (0.15, 0.25):
            bl = (1 - w) * J["tp"] + w * J["c"]
            print(f"  {lab:>20} w={w:.0%}: corr {corr:>+5.2f} | "
                  f"dSharpe FULL {sh(bl) - sh(J['tp']):>+5.2f} | "
                  f"dSharpe HOLD {sh(bl, lo=HOLDOUT) - sh(J['tp'], lo=HOLDOUT):>+5.2f}")
        mg = al.marginal_vs_tp01(r)
        print(f"  {'':>20}   verdetto {mg['marginal_verdict']} | robust_oos={mg['robust_oos']} | "
              f"multicut={mg['multicut_persistent']} | is_hedge={mg['is_hedge']} | "
              f"insample_edge={mg['has_insample_edge']} | beats_noise={mg['beats_noise_null']}")
        print(f"  {'':>20}   jackknife_min_uplift {mg['jackknife_min_uplift']:+.3f} | "
              f"alpha_ann {mg['alpha_ann']:+.1%} | beta_TP01 {mg['beta_to_tp01']:+.3f} | "
              f"uplift TP01-up {mg['uplift_tp01_up']:+.3f} / TP01-down {mg['uplift_tp01_down']:+.3f}")

    print("\n" + "=" * 100)


if __name__ == "__main__":
    main()