PythagorasGoal/tests/test_vrp_f_watch.py

"""Test del sorvegliante di f sulle strutture VRP (scripts/live/vrp_f_watch.py).

Cosa va congelato:

* **Il criterio di sufficienza**, perche' e' pre-registrato: se un domani non si arriva alla
  soglia, la tentazione e' abbassarla. Il test la lega a una costante, e cambiarla e' un atto
  visibile in un diff.
* **La statistica appaiata**: la domanda e' una DIFFERENZA fra due strutture sulla stessa
  scadenza, non due intervalli guardati a occhio.
* **Il fatto che 'non abbastanza' NON e' 'nessuna differenza'** — con l'IC che contiene lo zero
  si aspetta, non si conclude.
* **I controlli positivi**: un criterio che non scatta mai e uno rotto si somigliano troppo.
"""
from __future__ import annotations

import importlib.util
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"))

_spec = importlib.util.spec_from_file_location(
    "vrp_f_watch", ROOT / "scripts" / "live" / "vrp_f_watch.py")
W = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(W)


def _frame(pairs: int, diff: float, spread: float = 0.0, seed: int = 0) -> pd.DataFrame:
    """Coppie sintetiche: `pairs` scadenze con canonico ~0.72 e candidato ~0.72+diff."""
    r = np.random.default_rng(seed)
    rows = []
    for i in range(pairs):
        base = 0.72 + r.normal(0, spread)
        for name, v in (("canonico", base), ("candidato", base + diff + r.normal(0, spread))):
            rows.append(dict(asset="BTC", struct=name, exp=pd.Timestamp("2026-01-01") + pd.Timedelta(days=i),
                             ts=pd.Timestamp("2026-01-01"), dte=7.0, f_short=1.0, f_long=2.0,
                             f_net=v, f_net_mid=v, quota_lunga=0.18))
    return pd.DataFrame(rows)


# ===========================================================================
# il criterio, che e' pre-registrato
# ===========================================================================
def test_il_criterio_di_sufficienza_e_quello_dichiarato_il_30_luglio():
    assert W.MIN_PAIRS == 40
    assert W.MAX_CI_WIDTH == 0.12


def test_le_strutture_confrontate_sono_quelle_congelate():
    """Il canonico e' VRP01 in produzione; il candidato e' la cella scelta al buio dal gate."""
    assert W.STRUCTS["canonico"][3:] == (-0.28, -0.10)
    assert W.STRUCTS["candidato"][3:] == (-0.28, -0.05)
    assert W.STRUCTS["candidato"][2] == 10.0            # il tenore sotto esame


# ===========================================================================
# la statistica
# ===========================================================================
def test_la_differenza_e_appaiata_per_scadenza():
    """Con una differenza costante di +0.10 su ogni coppia, la mediana appaiata deve essere
    esattamente +0.10 e l'IC strettissimo — cosa che due IC separati non garantirebbero."""
    a = W.assess(_frame(50, diff=0.10, spread=0.05, seed=1))
    assert a["pairs"] == 50
    assert a["diff"] == pytest.approx(0.10, abs=0.03)
    assert a["ci_width"] < 0.10


def test_poche_coppie_non_bastano_anche_se_la_differenza_e_grande():
    """Il criterio ha DUE gambe: un effetto grande su 5 coppie non e' una misura."""
    a = W.assess(_frame(5, diff=0.30, spread=0.05, seed=2))
    assert not a["ready"] and a["pairs"] < W.MIN_PAIRS


def test_molte_coppie_con_ic_largo_non_bastano():
    """L'altra gamba: tante osservazioni ma rumorose non chiudono la domanda."""
    a = W.assess(_frame(60, diff=0.10, spread=0.60, seed=3))
    assert a["pairs"] >= W.MIN_PAIRS
    assert a["ci_width"] > W.MAX_CI_WIDTH and not a["ready"]


# ===========================================================================
# CONTROLLI POSITIVI
# ===========================================================================
def test_controllo_positivo_il_criterio_sa_dire_di_si():
    a = W.assess(_frame(60, diff=0.15, spread=0.05, seed=4))
    assert a["ready"] and a["esclude_zero"]


def test_una_differenza_nulla_e_misurata_come_nulla_non_come_ignota():
    """A campione pieno e differenza vera zero la domanda E' risolta (`ready`): 'misurato
    uguale' e' un risultato, 'non misurato' no."""
    a = W.assess(_frame(60, diff=0.0, spread=0.05, seed=5))
    assert a["ready"]


def test_l_intervallo_bootstrap_e_calibrato():
    """La soglia pre-registrata sull'AMPIEZZA dell'IC vale solo se l'IC e' calibrato: se
    fosse troppo stretto, 'ready' arriverebbe presto e con un falso positivo in mano.
    Misurato: con differenza vera nulla lo zero viene escluso nel ~5% dei casi, come deve.

    (Serve un test sulla PROPRIETA' e non su un seed: la prima stesura fissava seed=5 e
    falliva perche' quel seed era uno dei ~5% legittimi.)"""
    n_seeds = 100
    esclusi = sum(W.assess(_frame(40, diff=0.0, spread=0.05, seed=s))["esclude_zero"]
                  for s in range(n_seeds))
    assert esclusi / n_seeds <= 0.12          # 5% atteso, +3 sd binomiali su 100 estrazioni


# ===========================================================================
# stato reale del progetto
# ===========================================================================
@pytest.fixture(scope="module")
def reale():
    return W.assess(W.measure())


def test_oggi_il_campione_non_basta_e_lo_script_lo_dice(reale):
    """Congelato: al 2026-07-30 sono 16 coppie con IC95 che CONTIENE lo zero. Se un domani
    questo test fallisce e' perche' il campione e' cresciuto — ed e' il momento di guardare."""
    assert reale["pairs"] >= 16
    if reale["pairs"] < W.MIN_PAIRS:
        assert not reale["ready"]


def test_il_f_del_canonico_replica_la_misura_del_30_luglio(reale):
    """Percorso indipendente da `r0730_vrp_real_quotes` (finestra DTE e pairing diversi):
    deve ritrovare ~0.73. Se diverge, una delle due misure e' rotta."""
    assert 0.65 <= reale["f_canonico"] <= 0.80


def test_l_ala_piu_lontana_e_piu_mispriced_ma_pesa_meno():
    """Il meccanismo, misurato: f_long molto peggiore sul candidato, ma quota sul premio corto
    molto minore -> effetto NETTO minore. E' il fatto che ha refutato il mio sospetto."""
    R = W.measure()
    med = R.groupby("struct")[["f_long", "quota_lunga", "f_net"]].median()
    assert med.loc["candidato", "f_long"] > med.loc["canonico", "f_long"]
    assert med.loc["candidato", "quota_lunga"] < med.loc["canonico", "quota_lunga"]
    assert med.loc["candidato", "f_net"] > med.loc["canonico", "f_net"]