PythagorasGoal/tests/test_skh_feed_freshness.py

"""Test della misura di FRESCHEZZA del feed 5m usato dal segnale SKH01 (2026-07-26).

Perche' esiste questo file: `src.live.livefeed.fresh_5m` ricade sul feed certificato **in
silenzio** se il fetch pubblico Deribit fallisce — non solleva. Il feed certificato si
rigenera una volta al giorno, quindi in quel caso la latenza d'uscita di SKH01 passa da ~1 ora a
~1 giorno senza che nulla lo segnali. Ed e' proprio la latenza d'uscita dove vive la qualita' del
path live (misura del 2026-07-26: modellarla male sottostimava il book di +0.081 Sharpe FULL).

I test blindano che la misura ci sia, sia corretta nei casi limite, e arrivi fino al report.

AGGIUNTA 2026-07-29 — il fallback resta silenzioso come COMPORTAMENTO, ma non piu' come CAUSA.
Quel giorno il feed e' ricaduto sul certificato in 6 giri orari su 8 (fino a 685 min di eta') e
l'allerta incolpava il "fetch pubblico KO" solo perche' quella nota era CABLATA in book_execute:
la causa vera non e' mai stata stabilita, perche' l'eccezione era gia' stata ingoiata e a
guasto rientrato non c'era piu' niente da interrogare. I test in coda blindano che la ragione
venga REGISTRATA, che distingua guasti diversi, e che non sopravviva a una chiamata riuscita
(senno' l'allerta riporterebbe un guasto gia' rientrato).
"""
from __future__ import annotations

import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

pytest.importorskip("pandas")
import pandas as pd  # noqa: E402

import src.live.livefeed as lf  # noqa: E402
from src.live.livefeed import SCHEMA, feed_age_minutes, merge_tail  # noqa: E402


def _df(ts_list):
    return pd.DataFrame([[t, 1.0, 1.0, 1.0, 1.0, 1.0] for t in ts_list], columns=SCHEMA)


NOW = 1_800_000_000_000          # riferimento fisso: i test non devono dipendere dall'orologio


def test_feed_fresco_ha_eta_zero():
    """Barra chiusa proprio adesso -> eta' 0 (l'eta' si misura dalla CHIUSURA, non dall'apertura)."""
    assert feed_age_minutes(_df([NOW - 5 * 60_000]), now_ms=NOW) == pytest.approx(0.0)


def test_eta_misurata_dalla_chiusura_della_barra():
    """Una barra 5m aperta 35 min fa e' chiusa da 30 -> eta' 30, non 35."""
    assert feed_age_minutes(_df([NOW - 35 * 60_000]), now_ms=NOW) == pytest.approx(30.0)


def test_feed_di_un_giorno_supera_di_gran_lunga_la_soglia():
    """E' lo scenario da intercettare: fallback silenzioso sul certificato (rebuild 1x/giorno)."""
    age = feed_age_minutes(_df([NOW - 24 * 60 * 60_000]), now_ms=NOW)
    assert age == pytest.approx(24 * 60 - 5)
    assert age > 30.0, "il caso che l'allerta deve prendere"


def test_eta_non_negativa_con_barra_nel_futuro():
    """Uno skew d'orologio non deve produrre un'eta' negativa che passerebbe ogni soglia."""
    assert feed_age_minutes(_df([NOW + 60 * 60_000]), now_ms=NOW) == 0.0


@pytest.mark.parametrize("bad", [None, pd.DataFrame(), pd.DataFrame({"x": [1]})])
def test_input_non_interpretabile_da_None_non_zero(bad):
    """None = 'non misurata' e va trattata come sospetta; se tornasse 0.0 sembrerebbe fresca."""
    assert feed_age_minutes(bad, now_ms=NOW) is None


def test_merge_tail_vince_sui_duplicati_e_ringiovanisce_il_feed():
    """La coda fresca deve davvero spostare in avanti l'ultima barra: e' il meccanismo che
    tiene bassa la latenza d'uscita."""
    base = _df([NOW - 3 * 86_400_000, NOW - 2 * 86_400_000])
    tail = _df([NOW - 2 * 86_400_000, NOW - 5 * 60_000])
    m = merge_tail(base, tail)
    assert len(m) == 3, "il duplicato dev'essere collassato"
    assert feed_age_minutes(m, now_ms=NOW) == pytest.approx(0.0)
    assert feed_age_minutes(base, now_ms=NOW) > 30.0


def test_merge_tail_vuota_lascia_il_base_intatto():
    """Fetch fallito -> coda vuota -> il feed resta il certificato: e' il fallback silenzioso,
    e il test lo fissa come comportamento ATTESO (non e' un bug, e' il motivo della misura)."""
    base = _df([NOW - 86_400_000])
    assert merge_tail(base, pd.DataFrame(columns=SCHEMA)).equals(base)
    assert merge_tail(base, None).equals(base)


def test_book_report_espone_la_chiave(monkeypatch):
    """La misura deve arrivare fino al report, altrimenti l'esecutore non puo' allertare.
    Senza live_feed la chiave esiste ma vale None ('non misurata')."""
    from src.live import book as bk
    r = bk.book_report(offline=True, equity_override=600.0)
    assert "skh_feed_age_min" in r
    assert r["skh_feed_age_min"] is None


def _stub_certificato(monkeypatch):
    """Feed 'certificato' finto e vecchio di 10 ore. Evita di leggere i 22MB reali da disco:
    questi test riguardano la CAUSA del fallback, non i dati."""
    old = NOW - 10 * 60 * 60_000
    monkeypatch.setattr(lf, "load_data", lambda a, tf: _df([old - 300_000, old]))


def test_causa_registrata_quando_il_fetch_solleva(monkeypatch):
    """La forma del guasto del 29/07: l'errore cade sulla PRIMA pagina. `_fetch_recent_5m` lo
    ingoia con `break` e ritorna un frame vuoto, quindi `fresh_5m` non vede nessuna eccezione ->
    senza registrazione esplicita la ragione sparisce e resta solo 'il feed e' vecchio'.
    (Il 429 qui e' un errore d'esempio: la causa reale di quel giorno e' rimasta ignota.)"""
    ccxt = pytest.importorskip("ccxt")
    _stub_certificato(monkeypatch)

    class Boom:
        def fetch_ohlcv(self, *a, **k):
            raise RuntimeError("429 Too Many Requests")

    monkeypatch.setattr(ccxt, "deribit", lambda *a, **k: Boom())
    lf.fresh_5m("BTC")
    err = lf.last_fetch_error()
    assert err is not None, "la causa del fallback dev'essere registrata, non presunta"
    assert "429" in err and "BTC/USD:BTC" in err, err


def test_causa_distingue_risposta_vuota_da_eccezione(monkeypatch):
    """Deribit che risponde SENZA barre e Deribit che solleva sono guasti diversi (il primo non
    e' un errore di rete). Se la nota fosse la stessa, la diagnosi ripartirebbe da zero."""
    ccxt = pytest.importorskip("ccxt")
    _stub_certificato(monkeypatch)

    class Vuoto:
        def fetch_ohlcv(self, *a, **k):
            return []

    monkeypatch.setattr(ccxt, "deribit", lambda *a, **k: Vuoto())
    lf.fresh_5m("BTC")
    err = lf.last_fetch_error()
    assert err is not None and "429" not in err, err
    assert "vuota" in err or "nessuna barra" in err, err


def test_causa_registrata_anche_con_coda_TRONCATA(monkeypatch):
    """Il caso intermedio, quello che un test pigro salta: l'errore cade a META' paginazione.
    La coda parziale VIENE attaccata (meglio di niente) -> il feed non e' 'caduto', e' solo
    piu' vecchio del dovuto. Siccome la paginazione va IN AVANTI, cio' che manca sono le barre
    PIU' RECENTI: e' esattamente il caso in cui l'eta' da sola direbbe poco e la causa serve."""
    ccxt = pytest.importorskip("ccxt")
    _stub_certificato(monkeypatch)
    primo = NOW - 3 * 60 * 60_000                 # coda attaccata, ma ferma a 3 ore fa

    class MezzaStrada:
        def __init__(self):
            self._servito = False

        def fetch_ohlcv(self, *a, **k):
            if self._servito:
                raise RuntimeError("ConnectionResetError alla seconda pagina")
            self._servito = True
            return [[primo, 1.0, 1.0, 1.0, 1.0, 1.0]]

    monkeypatch.setattr(ccxt, "deribit", lambda *a, **k: MezzaStrada())
    df = lf.fresh_5m("BTC")
    err = lf.last_fetch_error()
    assert err is not None and "pagina 2" in err, err
    # la coda parziale c'e' davvero (l'ultima barra e' quella della pagina 1, non del certificato)
    assert int(df["timestamp"].iloc[-1]) == primo
    # ...e l'eta' resta fuori soglia: la causa spiega un feed vecchio, non ne salva uno.
    assert feed_age_minutes(df, now_ms=NOW) == pytest.approx(175.0)


def test_causa_non_sopravvive_a_una_chiamata_riuscita(monkeypatch):
    """Coda attaccata -> nessuna causa residua: un errore vecchio che resta appiccicato farebbe
    allertare su un guasto gia' rientrato (rumore che spegne l'attenzione sull'allerta vera)."""
    ccxt = pytest.importorskip("ccxt")
    _stub_certificato(monkeypatch)
    recente = NOW - 5 * 60_000

    class Ok:
        def __init__(self):
            self._servito = False

        def fetch_ohlcv(self, *a, **k):
            if self._servito:
                return []
            self._servito = True
            return [[recente, 1.0, 1.0, 1.0, 1.0, 1.0]]

    monkeypatch.setattr(ccxt, "deribit", lambda *a, **k: Ok())
    lf._LAST_ERROR = "residuo di una chiamata precedente"
    df = lf.fresh_5m("BTC")
    assert lf.last_fetch_error() is None, lf.last_fetch_error()
    assert feed_age_minutes(df, now_ms=NOW) == pytest.approx(0.0)


def test_book_report_espone_le_cause():
    """La causa dev'essere nel report: senza, `book_execute` non puo' allertare con la ragione
    vera e ricade sulla nota cablata — il difetto che ha tenuto nascosto il guasto del 29/07."""
    from src.live import book as bk
    r = bk.book_report(offline=True, equity_override=600.0)
    assert "skh_feed_errors" in r
    assert r["skh_feed_errors"] == {}, "senza live_feed non c'e' nessun fallback da spiegare"


def test_soglia_di_config_presente_e_sensata():
    """La soglia dev'essere piu' larga della cadenza del cron (orario) e molto piu' stretta
    di un giorno, sennò non distingue 'fresco' da 'ricaduto sul certificato'."""
    import json
    cfg = json.loads((ROOT / "config" / "live.json").read_text())
    thr = float(cfg["skh_feed_max_age_min"])
    assert 10.0 <= thr <= 120.0, thr