PythagorasGoal/tests/test_cb_chain_vrp.py

"""Test della catena opzioni reale (cerbero-bite) e della misura di f per VRP01.

Due famiglie:
  A. RILEVATORI di difetto del dato (quote vuote / book incrociato / non monotonia). Ognuno ha il
     suo CONTROLLO POSITIVO: un rilevatore che non segnala mai e' indistinguibile da uno rotto.
  B. MISURA di f. Il punto non e' "calcola un rapporto" ma che il rapporto sia quello GIUSTO:
     f dello spread, non della gamba corta — e' la distinzione che ha ribaltato la conclusione
     del 30/07 rispetto alla calibrazione del 20/06.

Nessun test tocca il DB o la rete: tutte le fixture sono sintetiche.
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd
import pytest

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

import cblib  # noqa: E402
from cblib import bs_put, causale, f_factors, pick_legs  # noqa: E402
from certify_cb_chain import (  # noqa: E402
    crossed_rate, hollow_rate, monotonicity_violations, worst_day_hollow,
)

TS = pd.Timestamp("2026-07-24T08:00:00Z")
EXP = pd.Timestamp("2026-07-31T08:00:00Z")


def _riga(strike, bid, ask, delta, mid=None, iv=50.0, ts=TS, tipo="P"):
    return {
        "ts": ts, "asset": "ETH", "instrument_name": f"ETH-31JUL26-{int(strike)}-{tipo}",
        "strike": float(strike), "exp": EXP, "option_type": tipo,
        "bid": bid, "ask": ask, "mid": (bid + ask) / 2 if mid is None and bid is not None else mid,
        "iv": iv, "delta": delta, "book_depth_top3": 100, "source": "research",
    }


def _catena() -> pd.DataFrame:
    """Catena put plausibile: premio crescente nello strike, delta piu' negativo vicino allo spot."""
    return pd.DataFrame([
        _riga(1500, 0.0020, 0.0025, -0.05),
        _riga(1600, 0.0050, 0.0055, -0.10),
        _riga(1700, 0.0110, 0.0120, -0.20),
        _riga(1750, 0.0160, 0.0175, -0.28),
        _riga(1800, 0.0240, 0.0260, -0.40),
    ])


# --------------------------------------------------------------- A. rilevatori

def test_quote_vuote_contate_e_non_confuse_con_righe_mancanti():
    df = _catena()
    df.loc[[0, 1], ["bid", "ask", "mid", "iv", "delta"]] = np.nan   # ticker fallito: la riga RESTA
    assert len(df) == 5, "una quota vuota non riduce il numero di righe — e' il punto del difetto"
    assert hollow_rate(df) == pytest.approx(2 / 5)


def test_controllo_positivo_catena_sana_non_segnala():
    df = _catena()
    assert hollow_rate(df) == 0.0
    assert crossed_rate(df) == 0.0
    assert monotonicity_violations(df) == 0.0


def test_book_incrociato_segnalato():
    df = _catena()
    df.loc[2, "bid"] = df.loc[2, "ask"] + 0.01
    assert crossed_rate(df) == pytest.approx(1 / 5)


def test_premio_non_monotono_nello_strike_segnalato():
    """Il premio di una put e' non-decrescente nello strike: e' il payoff, non un modello."""
    df = _catena()
    df.loc[3, "mid"] = 0.0001                       # quota stantia/corrotta
    v = monotonicity_violations(df)
    assert v > 0, "una put che costa MENO di una piu' OTM e' un difetto, va segnalato"


def test_giorno_peggiore_non_si_lascia_diluire_dalla_media():
    """Il guasto del 29/07 mediato su 7 giorni si legge 13%: la statistica giusta e' il giorno."""
    righe = []
    for g in range(7):
        ts = TS + pd.Timedelta(days=g)
        vuote = 8 if g == 6 else 0                  # guasto solo nell'ultimo giorno
        for i in range(10):
            r = _riga(1500 + 50 * i, 0.01, 0.011, -0.2, ts=ts)
            if i < vuote:
                r["bid"] = r["ask"] = r["mid"] = None
            righe.append(r)
    df = pd.DataFrame(righe)
    media = hollow_rate(df)
    peggiore, giorno = worst_day_hollow(df, days=7)
    assert media == pytest.approx(8 / 70)           # ~11%: il guasto sembra piccolo
    assert peggiore == pytest.approx(0.8)           # 80%: il guasto com'e' davvero
    assert giorno == (TS + pd.Timedelta(days=6)).date()


# --------------------------------------------------------------- B. struttura e f

def test_pick_legs_prende_le_gambe_piu_vicine_ai_delta_bersaglio():
    legs = pick_legs(_catena())
    assert legs is not None
    assert legs["k_short"] == 1750 and legs["d_short"] == pytest.approx(-0.28)
    assert legs["k_long"] == 1600 and legs["d_long"] == pytest.approx(-0.10)


def test_una_gamba_sola_quotata_non_e_meta_struttura():
    """Con una gamba sola non esiste un credit spread: va scartato, non prezzato a meta'."""
    df = _catena()
    df.loc[df["strike"] != 1750, ["bid", "ask"]] = np.nan
    assert pick_legs(df) is None


def test_gambe_incrociate_rifiutate():
    """Se la 'corta' ha strike sotto la 'lunga' la struttura e' invertita: None, non un credito."""
    df = _catena().copy()
    df["delta"] = [-0.28, -0.20, -0.10, -0.05, -0.02]   # delta bersaglio sugli strike sbagliati
    assert pick_legs(df) is None


def test_f_uguale_a_uno_quando_il_mercato_quota_esattamente_il_modello():
    """Controllo di round-trip: quote costruite DAL modello devono dare f = 1 su tutto."""
    spot, dvol, dte = 1800.0, 0.50, 7.0
    T = dte / 365.25
    ks, kl = 1750.0, 1600.0
    ps, pl = bs_put(spot, ks, T, dvol), bs_put(spot, kl, T, dvol)
    legs = {"k_short": ks, "k_long": kl,
            "bid_short": ps / spot, "ask_short": ps / spot,
            "bid_long": pl / spot, "ask_long": pl / spot,
            "mid_short": ps / spot, "mid_long": pl / spot}
    f = f_factors(legs, spot, dvol, dte)
    assert f["f_short"] == pytest.approx(1.0, rel=1e-9)
    assert f["f_long"] == pytest.approx(1.0, rel=1e-9)
    assert f["f_net"] == pytest.approx(1.0, rel=1e-9)


@pytest.mark.parametrize("k_long", [1700.0, 1650.0, 1600.0])
def test_il_f_della_gamba_corta_non_e_il_f_dello_spread(k_long):
    """IL test che vale il filone: con la sola ala comprata piu' cara del modello (skew), la gamba
    corta resta a f~1 e il CREDITO NETTO si comprime. Misurare una gamba sola da' la risposta
    sbagliata con segno rassicurante — e' cio' che era successo con la calibrazione del 20/06.

    La compressione segue l'identita'  f_net = (1 - f_long*k) / (1 - k),  k = premio_lungo/premio_corto:
    quanto piu' l'ala comprata pesa nel credito, tanto piu' il credito netto crolla a parita' di
    f_long. Per questo il test gira su tre ali: una soglia sola sarebbe stata una fixture fortunata.
    """
    spot, dvol, dte, f_long = 1800.0, 0.50, 7.0, 2.3
    T = dte / 365.25
    ks = 1750.0
    ps, pl = bs_put(spot, ks, T, dvol), bs_put(spot, k_long, T, dvol)
    legs = {"k_short": ks, "k_long": k_long,
            "bid_short": ps / spot, "ask_short": ps / spot,
            "bid_long": f_long * pl / spot, "ask_long": f_long * pl / spot,
            "mid_short": ps / spot, "mid_long": f_long * pl / spot}
    f = f_factors(legs, spot, dvol, dte)
    k = pl / ps
    assert f["f_short"] == pytest.approx(1.0, rel=1e-9), "la gamba venduta e' prezzata bene"
    assert f["f_long"] == pytest.approx(f_long, rel=1e-9)
    assert f["f_net"] == pytest.approx((1 - f_long * k) / (1 - k), rel=1e-9)
    assert f["f_net"] < f["f_short"], "il f dello spread e' SEMPRE sotto quello della gamba corta"
    assert f["f_net"] < 1.0


def test_fill_conservativo_non_e_mai_migliore_del_mid():
    """Attraversare lo spread non puo' aumentare il credito: guardia di verso, non di valore."""
    legs = pick_legs(_catena())
    f = f_factors(legs, spot=1800.0, dvol_frac=0.50, dte_days=7.0)
    assert f["cred_real"] <= f["cred_mid"] + 1e-12
    assert f["f_net"] <= f["f_net_mid"] + 1e-12


# --------------------------------------------------------------- C. guardia sulla convenzione

def test_nel_sleeve_f_moltiplica_il_credito_NETTO():
    """Tutta l'aritmetica del 30/07 poggia su questa convenzione: `f` scala il credito NETTO
    (`net_prem = (bs(Ks) - bs(Kl)) * f`). Se qualcuno cambia dove entra `f`, il f misurato sulle
    quote reali non e' piu' il parametro giusto da sostituire e le conclusioni del diario
    2026-07-30 vanno rifatte. Questo test deve rompersi PRIMA che questo accada."""
    import inspect

    from src.portfolio import sleeves as SL

    src = inspect.getsource(SL._vrp_weekly_asset)
    righe = [r.strip() for r in src.splitlines() if "net_prem" in r and "cfg[\"f\"]" in r]
    assert righe, "non trovo piu' `net_prem = (...) * cfg['f']` in _vrp_weekly_asset"
    r = righe[0]
    assert r.count("_bs_put") == 2 and r.index("cfg[\"f\"]") > r.rindex("_bs_put"), (
        "`f` deve moltiplicare la DIFFERENZA dei due premi, non una gamba sola"
    )
    assert SL.VRP_CFG["f"] == 1.0, (
        "il sleeve gira ancora a f=1.0: se cambia, aggiornare il diario 2026-07-30 e CLAUDE.md"
    )


# ===========================================================================================
# C. CAUSALITA' delle serie di contesto (debito §5.18, 09/09): `asof(ts)` non guarda avanti
# ===========================================================================================

def test_causale_rietichetta_alla_chiusura_e_non_tocca_i_valori():
    """Il feed e' etichettato all'APERTURA: la chiusura della barra T e' nota a T+cadenza.
    DEVE FALLIRE SE: qualcuno "semplifica" togliendo lo spostamento, o sposta i VALORI."""
    idx = pd.date_range("2026-06-01", periods=5, freq="h", tz="UTC")
    S = pd.Series([10.0, 11.0, 12.0, 13.0, 14.0], index=idx)
    Sc = causale(S, "1h")
    assert list(Sc.index) == list(idx + pd.Timedelta(hours=1))
    assert Sc.asof(idx[2]) == 11.0, "a 02:00 e' nota la chiusura della barra 01:00"
    assert S.asof(idx[2]) == 12.0, "il difetto che la funzione ripara (controllo positivo)"
    assert np.isnan(Sc.asof(idx[0])), "prima della prima chiusura nota non c'e' spot"
    assert list(Sc.values) == list(S.values), "i VALORI non si toccano, solo le etichette"


def test_spot_series_e_causale_sul_feed_1h(monkeypatch):
    """`cblib.spot_series(a).asof(ts)` = ultima chiusura ORARIA nota a ts. Controllo positivo:
    la serie grezza del feed (etichettata all'apertura) a ts da' la chiusura di ts+1h."""
    import scripts.analysis.research_lab as rl
    idx = pd.date_range("2026-06-01", periods=6, freq="h", tz="UTC")
    epoch = pd.Timestamp("1970-01-01", tz="UTC")     # NON `asi8 // 10**6`: in pandas 3 e' in `us` (D6)
    px = pd.DataFrame({"timestamp": ((idx - epoch) // pd.Timedelta(milliseconds=1)).astype("int64"),
                       "close": [100.0, 101.0, 102.0, 103.0, 104.0, 105.0]})
    monkeypatch.setattr(rl, "load_tf", lambda asset, tf: px)
    cblib.spot_series.cache_clear()
    try:
        S = cblib.spot_series("ZZZ")
        ts = idx[3] + pd.Timedelta(minutes=17)          # 03:17: la barra 03:00 e' ancora aperta
        assert S.asof(ts) == 102.0, "a 03:17 l'ultima chiusura nota e' quella della barra 02:00"
        assert S.asof(idx[3]) == 102.0, "a 03:00 esatte la barra 02:00 e' appena chiusa"
        assert S.asof(idx[5] + pd.Timedelta(hours=1)) == 105.0, "l'ultima chiusura arriva un'ora dopo"
        grezza = pd.Series(px["close"].values, index=idx)
        assert grezza.asof(ts) == 103.0, "il difetto riparato: la serie del feed a 03:17 da' la chiusura delle 04:00"
    finally:
        cblib.spot_series.cache_clear()


def test_dvol_series_e_causale_sul_feed_giornaliero(monkeypatch, tmp_path):
    """La riga DVOL del giorno D e' la chiusura delle 23:00 di D: dentro D non e' ancora nota.
    `cblib.dvol_series(a).asof(ts in D)` deve dare la chiusura di D−1."""
    days = pd.date_range("2026-06-01", periods=4, freq="D", tz="UTC")
    epoch = pd.Timestamp("1970-01-01", tz="UTC")
    pd.DataFrame({"timestamp": ((days - epoch) // pd.Timedelta(milliseconds=1)).astype("int64"),
                  "close": [36.0, 37.0, 43.0, 47.0]}).to_parquet(tmp_path / "dvol_zzz.parquet")
    monkeypatch.setattr(cblib, "RAW", tmp_path)
    cblib.dvol_series.cache_clear()
    try:
        V = cblib.dvol_series("ZZZ")
        ts = days[2] + pd.Timedelta(hours=10)            # 03/06 10:00
        assert V.asof(ts) == 37.0, "dentro il 03/06 l'ultimo DVOL noto e' la chiusura del 02/06"
        assert V.asof(days[3]) == 43.0, "alle 00:00 del 04/06 e' nota la chiusura del 03/06"
        assert np.isnan(V.asof(days[0] + pd.Timedelta(hours=5))), "nel primo giorno non c'e' storia"
    finally:
        cblib.dvol_series.cache_clear()