PythagorasGoal/tests/test_eq_crosscheck.py

"""Test del cross-check equity contro il gemello (src/data/eq_crosscheck.py).

Il rilevatore nasce da un difetto VERO trovato il 2026-07-26: CSPX 2012-01-13 con open/high in
USD e low/close in EUR. La certificazione esistente non lo vedeva perche' la sua unica guardia
sui movimenti era `maxret > 50% -> SPIKE?` e una contaminazione EUR/USD vale ~22-28%.

I test che contano davvero sono i CONTROLLI su dati reali:
  * il difetto noto viene segnalato (senza questo il rilevatore potrebbe essere semplicemente rotto);
  * gli eventi di mercato VERI non vengono segnalati — IDTL marzo 2020 (liquidazione dei treasury)
    e IGLN aprile 2013 (crollo dell'oro) sono andate-e-ritorni identici nella FORMA a una stampa
    sbagliata, e una scansione fatta sui prezzi invece che sul rapporto col gemello li segnalava.
E il test `test_limite_dichiarato_*` congela il LIMITE noto: se qualcuno abbassa la soglia, il
test dice esattamente cosa e' cambiato invece di lasciar credere che il rilevatore sia completo.
"""
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))

from src.data.eq_crosscheck import (  # noqa: E402
    MIN_DEV, MIN_UNEXPLAINED, REVERT_BARS, TWIN_MOVE_FLOOR,
    detect_twin_outliers, repair_twin_outliers,
)

RAW = ROOT / "data" / "raw"


def _load(prefix: str, sym: str) -> pd.DataFrame:
    p = RAW / f"{prefix}_{sym.lower()}_1d.parquet"
    if not p.exists():
        pytest.skip(f"{p.name} assente (data/raw e' gitignored)")
    d = pd.read_parquet(p)
    d.index = pd.to_datetime(d["timestamp"], unit="ms", utc=True)
    return d.sort_index()


def _close(prefix: str, sym: str) -> pd.Series:
    d = _load(prefix, sym)
    return pd.Series(d["close"].astype(float).values, index=d.index)


def _serie(vals, start="2020-01-01"):
    return pd.Series(np.asarray(vals, float),
                     index=pd.bdate_range(start, periods=len(vals)))


# ===========================================================================
# controlli su DATI REALI — un rilevatore che non segnala nulla puo' essere rotto
# ===========================================================================
def test_controllo_positivo_il_difetto_noto_di_cspx_viene_segnalato():
    """IL test. Se questo fallisce, il rilevatore non serve a niente."""
    bad = detect_twin_outliers(_close("eqx", "CSPX"), _close("eq", "SPY"))
    date = {pd.Timestamp(b["date"]).date().isoformat() for b in bad}
    assert "2012-01-13" in date, f"difetto noto non rilevato (segnalate: {sorted(date)})"


def test_il_difetto_ha_la_firma_valutaria_attesa():
    bad = [b for b in detect_twin_outliers(_close("eqx", "CSPX"), _close("eq", "SPY"))
           if pd.Timestamp(b["date"]).date().isoformat() == "2012-01-13"]
    assert len(bad) == 1
    b = bad[0]
    assert b["dev"] < -0.15, "la deviazione non ha la scala di una contaminazione valutaria"
    assert abs(b["twin_move"]) < 0.01, "il gemello si era mosso: allora non e' inspiegato"
    assert b["unexplained"] > 3 * MIN_UNEXPLAINED / 2


def test_controllo_negativo_gli_eventi_di_mercato_veri_non_sono_segnalati():
    """Marzo 2020 sui treasury e aprile 2013 sull'oro sono andate-e-ritorni VERI: nella forma
    identici a una stampa sbagliata. Li distingue solo il fatto che li fa anche il gemello."""
    for veic, gem, quando in [("IDTL", "TLT", "2020-03"), ("IGLN", "GLD", "2013-04")]:
        bad = detect_twin_outliers(_close("eqx", veic), _close("eq", gem))
        colpiti = [pd.Timestamp(b["date"]).strftime("%Y-%m") for b in bad]
        assert quando not in colpiti, f"{veic}: evento di mercato vero scambiato per difetto"


def test_nessun_falso_positivo_sui_veicoli_puliti():
    for veic, gem in [("EQQQ", "QQQ"), ("XRSU", "IWM"), ("R2US", "IWM"),
                      ("IDTL", "TLT"), ("IGLN", "GLD"), ("IHYU", "HYG")]:
        bad = detect_twin_outliers(_close("eqx", veic), _close("eq", gem))
        assert bad == [], f"{veic}/{gem}: falsi positivi {[str(b['date']) for b in bad]}"


# ===========================================================================
# le tre condizioni, una alla volta
# ===========================================================================
def test_una_deviazione_che_il_gemello_SPIEGA_non_e_segnalata():
    """La condizione decisiva: un disallineamento d'orario non puo' superare il movimento del
    mercato. Se il gemello si e' mosso quanto il veicolo, non c'e' niente da segnalare."""
    n = 80
    twin = _serie(np.full(n, 100.0))
    twin.iloc[40] = 80.0          # il gemello CROLLA del 20%...
    twin.iloc[41] = 100.0
    veh = twin * 1.0              # ...e il veicolo fa lo stesso
    veh.iloc[40] = 78.0           # con un piccolo scarto d'orario
    assert detect_twin_outliers(veh, twin) == []


def test_una_divergenza_che_PERSISTE_non_e_una_stampa():
    """Un cambio di livello permanente (fusione, cambio di indice, ridenominazione) va tenuto:
    scartarlo cancellerebbe un fatto invece di un errore."""
    n = 80
    twin = _serie(np.full(n, 100.0))
    veh = _serie(np.concatenate([np.full(40, 100.0), np.full(n - 40, 78.0)]))
    assert detect_twin_outliers(veh, twin) == []


def test_una_deviazione_troppo_piccola_non_e_segnalata():
    n = 80
    twin = _serie(np.full(n, 100.0))
    veh = twin.copy()
    veh.iloc[40] = 100.0 * (1 - MIN_DEV / 2)
    assert detect_twin_outliers(veh, twin) == []


# ===========================================================================
# iniezione: cosa si cattura e cosa NO (il limite dichiarato, congelato in un test)
# ===========================================================================
def _inietta(fx: float):
    n = 200
    rng = np.random.default_rng(7)
    base = 100 * np.exp(np.cumsum(rng.normal(0, 0.01, n)))
    twin = _serie(base)
    veh = _serie(base * 1.0)
    veh.iloc[120] = veh.iloc[120] / fx
    return veh, twin


def test_contaminazione_a_cambio_alto_gbp_viene_catturata():
    veh, twin = _inietta(1.27)
    bad = detect_twin_outliers(veh, twin)
    assert len(bad) == 1 and pd.Timestamp(bad[0]["date"]) == veh.index[120]


def test_limite_dichiarato_contaminazione_a_cambio_basso_eur_sfugge_a_mercato_mosso():
    """⚠️ Questo test NON descrive un comportamento desiderato: congela un LIMITE noto.

    EURUSD ~1.09 produce ~8% di deviazione, che sta dentro il rumore legittimo dei giorni
    violenti (misurato 9.9% il 2025-04-09). Abbassare la soglia per catturarlo genererebbe
    falsi positivi sui giorni veri. Il punto e' chiuso altrove misurando il DANNO
    (r0726_gtaa_ucits.py sez. 5: dSharpe mediano -0.003, peggiore -0.06).
    Se qualcuno cambia le soglie, questo test dice esattamente cosa e' cambiato.
    """
    n = 200
    rng = np.random.default_rng(11)
    base = 100 * np.exp(np.cumsum(rng.normal(0, 0.01, n)))
    twin = _serie(base)
    veh = _serie(base.copy())
    veh.iloc[120] /= 1.09
    twin.iloc[120] *= 1.02        # il gemello quel giorno si muove: la deviazione diventa 'spiegabile'
    assert detect_twin_outliers(veh, twin) == [], \
        "il limite dichiarato non vale piu': aggiornare la docstring del modulo e il diario"


# ===========================================================================
# riparazione
# ===========================================================================
def test_la_riparazione_scarta_la_barra_e_non_ne_inventa_una():
    d = _load("eqx", "CSPX")
    pulito, bad = repair_twin_outliers(d, _close("eq", "SPY"))
    assert len(bad) == 1
    assert len(pulito) == len(d) - 1
    rimasta = pd.to_datetime(pulito.index).tz_localize(None).normalize()
    assert pd.Timestamp("2012-01-13") not in set(rimasta)
    # nessun prezzo ricostruito: i close superstiti sono identici agli originali
    orig = d["close"]
    assert (pulito["close"].values == orig.loc[pulito.index].values).all()


def test_la_riparazione_rende_sano_il_rendimento_a_cavallo():
    d = _load("eqx", "CSPX")
    pulito, _ = repair_twin_outliers(d, _close("eq", "SPY"))
    r = pulito["close"].pct_change()
    w = r[(r.index >= "2012-01-10") & (r.index <= "2012-01-20")]
    assert w.abs().max() < 0.05, f"resta un movimento assurdo: {w.abs().max():.1%}"


def test_riparazione_idempotente():
    d = _load("eqx", "CSPX")
    p1, b1 = repair_twin_outliers(d, _close("eq", "SPY"))
    p2, b2 = repair_twin_outliers(p1, _close("eq", "SPY"))
    assert b2 == [] and len(p2) == len(p1)


def test_serie_vuota_o_cortissima_non_esplode():
    vuoto = pd.DataFrame({"close": []}, index=pd.DatetimeIndex([]))
    assert repair_twin_outliers(vuoto, _serie([1.0, 2.0])) == (vuoto, [])
    assert detect_twin_outliers(_serie([1.0] * 5), _serie([1.0] * 5)) == []


# ===========================================================================
# la taratura e' un valore, non un dettaglio
# ===========================================================================
def test_le_soglie_sono_congelate():
    """Tarate su misura: rumore massimo 8.1, difetto 42.7 -> soglia 18.0 equidistante in scala
    logaritmica. Cambiarle senza rifare la taratura invalida ogni conclusione del 26/07."""
    assert MIN_UNEXPLAINED == 18.0
    assert MIN_DEV == 0.05
    assert TWIN_MOVE_FLOOR == 0.005
    assert REVERT_BARS == 5