PythagorasGoal/scripts/live/vrp_f_watch.py

"""VRP-f WATCH — misura f sulle strutture VRP dalle quote REALI, e avvisa quando basta.

PERCHE' ESISTE. Il gate del tenore (30/07, `r0730_vrp_tenor_gate`) ha bocciato il candidato
**10g / delta -0.28 / -0.05** sul deflated-Sharpe (0.948 < 0.95), e fra le ragioni di cautela
avevo scritto che quel candidato *"sta dove il modello sbaglia di piu'"* — compra l'ala piu'
lontana, che il 30/07 aveva misurato sottoprezzata ~2.3x.

⚠️ **Quel sospetto e' stato MISURATO e la conclusione era sbagliata.** L'ala lontana e' molto
piu' mispriced in RELATIVO (f_long 5.85 contro 2.23) ma pesa il **3.2%** del premio corto invece
del **18.4%**, quindi sul credito NETTO l'effetto e' minore: f_net **0.852** contro **0.718**.
Meccanismo giusto, segno della conclusione sbagliato.
(Numeri del 30/07, misurati con `cblib` che leggeva spot e DVOL di UN'ORA / fino a 24 ORE dopo
l'ingresso — debito §5.18, riparato la sera del 09/09: da allora la tabella stampata da questo
script e' la fonte, e il f canonico e' passato da 0,732 a 0,706 sulle stesse 28 coppie.)

E' pero' una misura su **16 coppie**: la differenza appaiata e' **+0.109 con IC95
[-0.047, +0.193]**, cioe' **compatibile con zero**. Serve piu' campione, e "quando ce ne sara'
abbastanza" e' un criterio, non un promemoria — un lead senza sorvegliante e senza data e' un
lead perso (lezione DVOLSPREAD, 26/07). Questo script fa la misura a ogni giro e **avvisa da
solo** quando il campione basta.

CRITERIO DI SUFFICIENZA — PRE-REGISTRATO IL 2026-07-30, PRIMA DI AVERE IL CAMPIONE:
    (a) >= 40 coppie appaiate (asset x scadenza), contro le 16 di oggi
    (b) ampiezza dell'IC95 della differenza appaiata <= 0.12 (oggi 0.240)
Con ~2 coppie/settimana (scadenze settimanali x 2 asset) (a) cade verso **fine ottobre 2026**.
Quando entrambe sono soddisfatte lo script manda una notifica UNA volta e si zittisce.

⚠️ **COSA QUESTA MISURA NON FA.** Non promuove niente. Il candidato e' bocciato sul
deflated-Sharpe, che non dipende da f; e la regola "niente short-vol da modello in deploy" resta.
Un f favorevole toglie UN argomento di cautela, non aggiunge un edge. Il valore operativo vero e'
sull'altra riga: il f del **canonico** e' cio' che rende onesti i numeri di uno sleeve che sta
nel book.

    uv run python scripts/live/vrp_f_watch.py
    uv run python scripts/live/vrp_f_watch.py --quiet     # per il cron
"""
from __future__ import annotations

import argparse
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))
sys.path.insert(0, str(PROJECT_ROOT / "scripts" / "research"))

from cblib import (                                                        # noqa: E402
    dvol_series, f_factors, load_chain, pick_legs, puts, spot_series,
)
from src.live.notifier import notify                                       # noqa: E402

STATE = PROJECT_ROOT / "data" / "vrp_f_watch" / "state.json"

# --- config CONGELATA il 2026-07-30 ---------------------------------------
STRUCTS = {
    # nome        (dte_lo, dte_hi, dte_target, short_delta, long_delta)
    "canonico":  (4.0, 10.0, 7.0, -0.28, -0.10),      # VRP01 in produzione
    "candidato": (8.0, 13.0, 10.0, -0.28, -0.05),     # cella scelta al buio dal gate 30/07
}
MIN_PAIRS = 40            # criterio (a), pre-registrato
MAX_CI_WIDTH = 0.12       # criterio (b), pre-registrato
BOOT = 20000
SEED = 7
ASSETS = ("BTC", "ETH")


def _boot_median_ci(x: np.ndarray, seed: int = SEED) -> tuple[float, float]:
    if len(x) < 3:
        return float("nan"), float("nan")
    r = np.random.default_rng(seed)
    b = np.median(r.choice(x, (BOOT, len(x)), replace=True), axis=1)
    return float(np.percentile(b, 2.5)), float(np.percentile(b, 97.5))


def measure(chain: pd.DataFrame | None = None) -> pd.DataFrame:
    """Un'osservazione per (asset, struttura, scadenza): lo snapshot col DTE piu' vicino al
    bersaglio in cui ENTRAMBE le gambe sono quotate."""
    ch = load_chain() if chain is None else chain
    rows = []
    for asset in ASSETS:
        p = puts(asset, ch)
        S, V = spot_series(asset), dvol_series(asset)
        for name, (lo, hi, tgt, sd, ld) in STRUCTS.items():
            w = p[(p["dte"] >= lo) & (p["dte"] <= hi)]
            for exp, g in w.groupby("exp"):
                order = sorted(g["ts"].unique(),
                               key=lambda t: abs((exp - pd.Timestamp(t)).total_seconds() / 86400.0 - tgt))
                for ts in order:
                    legs = pick_legs(g[g["ts"] == ts], sd, ld)
                    if legs is None:
                        continue
                    t = pd.Timestamp(ts).as_unit("ns")
                    dte = (exp - pd.Timestamp(ts)).total_seconds() / 86400.0
                    ff = f_factors(legs, float(S.asof(t)), float(V.asof(t)) / 100.0, dte)
                    rows.append(dict(asset=asset, struct=name, exp=exp, ts=t, dte=dte,
                                     f_short=ff["f_short"], f_long=ff["f_long"],
                                     f_net=ff["f_net"], f_net_mid=ff["f_net_mid"],
                                     quota_lunga=ff["mod_long"] / ff["mod_short"]
                                     if ff["mod_short"] > 0 else np.nan))
                    break
    return pd.DataFrame(rows)


def assess(R: pd.DataFrame) -> dict:
    """Statistica appaiata per (asset, scadenza): stessa scadenza, due strutture."""
    if R.empty:
        return dict(pairs=0, ready=False, reason="nessuna osservazione")
    piv = R.pivot_table(index=["asset", "exp"], columns="struct", values="f_net").dropna()
    if len(piv) < 3:
        return dict(pairs=int(len(piv)), ready=False, reason="troppo poche coppie")
    d = (piv["candidato"] - piv["canonico"]).to_numpy()
    lo, hi = _boot_median_ci(d)
    c_lo, c_hi = _boot_median_ci(piv["canonico"].to_numpy())
    k_lo, k_hi = _boot_median_ci(piv["candidato"].to_numpy())
    width = hi - lo
    out = dict(
        pairs=int(len(piv)),
        f_canonico=float(np.median(piv["canonico"])), f_canonico_ci=[c_lo, c_hi],
        f_candidato=float(np.median(piv["candidato"])), f_candidato_ci=[k_lo, k_hi],
        diff=float(np.median(d)), diff_ci=[lo, hi], ci_width=float(width),
        n_positive=int((d > 0).sum()),
        esclude_zero=bool(lo > 0 or hi < 0),
    )
    out["ready"] = bool(out["pairs"] >= MIN_PAIRS and width <= MAX_CI_WIDTH)
    return out


def _load_state() -> dict:
    try:
        return json.loads(STATE.read_text())
    except Exception:
        return {}


def _save_state(d: dict) -> None:
    STATE.parent.mkdir(parents=True, exist_ok=True)
    STATE.write_text(json.dumps(d, indent=2, default=str))


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--quiet", action="store_true", help="stampa solo se c'e' qualcosa da dire")
    args = ap.parse_args()

    R = measure()
    a = assess(R)
    prev = _load_state()

    if not args.quiet:
        print("=" * 88)
        print("  VRP-f WATCH — f dalle quote reali, canonico vs candidato del gate 30/07")
        print("=" * 88)
        if R.empty:
            print("\n  nessuna osservazione nella catena.")
        else:
            print(f"\n  {'struttura':<12}{'oss.':>6}{'f_short':>10}{'f_long':>10}"
                  f"{'f_net':>10}{'quota ala/corta':>18}")
            for name, g in R.groupby("struct"):
                print(f"  {name:<12}{len(g):>6}{g['f_short'].median():>10.3f}"
                      f"{g['f_long'].median():>10.3f}{g['f_net'].median():>10.3f}"
                      f"{g['quota_lunga'].median()*100:>17.1f}%")
            print(f"\n  coppie appaiate (stesso asset, stessa scadenza): {a['pairs']}")
            print(f"    f canonico  {a['f_canonico']:.3f}  IC95 [{a['f_canonico_ci'][0]:.3f}, {a['f_canonico_ci'][1]:.3f}]")
            print(f"    f candidato {a['f_candidato']:.3f}  IC95 [{a['f_candidato_ci'][0]:.3f}, {a['f_candidato_ci'][1]:.3f}]")
            print(f"    DIFFERENZA  {a['diff']:+.3f}  IC95 [{a['diff_ci'][0]:+.3f}, {a['diff_ci'][1]:+.3f}]"
                  f"   ampiezza {a['ci_width']:.3f}   ({a['n_positive']}/{a['pairs']} positive)")
            print(f"\n  criterio pre-registrato 2026-07-30: >= {MIN_PAIRS} coppie E ampiezza IC95 <= {MAX_CI_WIDTH}")
            print(f"    coppie   {a['pairs']:>3} / {MIN_PAIRS}   {'OK' if a['pairs'] >= MIN_PAIRS else 'non ancora'}")
            print(f"    ampiezza {a['ci_width']:.3f} / {MAX_CI_WIDTH}   "
                  f"{'OK' if a['ci_width'] <= MAX_CI_WIDTH else 'non ancora'}")
            print(f"\n  -> {'CAMPIONE SUFFICIENTE' if a['ready'] else 'campione ancora insufficiente: si ASPETTA, non si decide'}")
            print("\n  ⚠ Anche a campione pieno questa misura NON promuove il candidato: e' bocciato")
            print("    sul deflated-Sharpe, che non dipende da f. Toglie un argomento di cautela.")
            print()

    if a.get("ready") and not prev.get("notified"):
        notify("📐 VRP-f WATCH — campione sufficiente",
               f"Le coppie appaiate sono {a['pairs']} (soglia {MIN_PAIRS}) e l'IC95 della "
               f"differenza è ampio {a['ci_width']:.3f} (soglia {MAX_CI_WIDTH}).\n\n"
               f"f canonico {a['f_canonico']:.3f} · f candidato {a['f_candidato']:.3f} · "
               f"differenza {a['diff']:+.3f} IC95 [{a['diff_ci'][0]:+.3f}, {a['diff_ci'][1]:+.3f}]"
               f"{' — ESCLUDE lo zero' if a['esclude_zero'] else ' — compatibile con zero'}.\n\n"
               f"Non promuove nulla: il candidato resta bocciato sul deflated-Sharpe.")
        a["notified"] = True
    else:
        a["notified"] = bool(prev.get("notified", False))

    _save_state(a)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())