PythagorasGoal/tests/test_monitor_health.py

"""Test della sorveglianza dei forward-monitor (src/live/monitor_health.py).

Tre gate pre-registrati (STATARB 27/09, XSR01 23/10, DVOLSPREAD 24/10) si decidono leggendo serie
che nessuno sorvegliava. Questi test coprono le due modalita' di guasto (coda ferma, buchi
interni) e — obbligatoriamente — i **controlli positivi**: un rilevatore che non ha mai segnalato
nulla e' indistinguibile da uno rotto finche' non si prova che sa segnalare.

Il test che conta piu' di tutti e' `test_una_serie_bucata_non_passa_per_fresca`: e' l'unico guasto
che una guardia di sola freschezza lascerebbe passare, ed e' quello che falsifica un gate senza
farsi notare.
"""
from __future__ import annotations

import json
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

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

from src.live.monitor_health import (  # noqa: E402
    MIN_COVERAGE,
    MonitorSpec,
    alerts_from,
    assess,
    check_all,
    expected_bars,
    read_stamps,
)

NOW = datetime(2026, 7, 27, 12, 0, tzinfo=timezone.utc)
DAILY = MonitorSpec("t", "t", "returns.jsonl", 24.0)
GATED = MonitorSpec("t_gate", "t_gate", "returns.jsonl", 24.0, gate="2026-10-23 — gate XSR01")
EQ = MonitorSpec("t_eq", "t_eq", "equity.csv", 24.0, calendar="equity", max_age_h=120.0)


def _daily(n: int, end: datetime = NOW, step_h: float = 24.0, skip: set[int] | None = None):
    """n timestamp a passo `step_h` che finiscono a `end`, opzionalmente con buchi."""
    skip = skip or set()
    return [end - timedelta(hours=step_h * (n - 1 - i)) for i in range(n) if i not in skip]


# ===========================================================================
# stato normale
# ===========================================================================
def test_una_serie_completa_e_fresca_e_ok():
    r = assess(DAILY, _daily(30), NOW)
    assert r["status"] == "OK" and r["coverage"] == 1.0


def test_ok_non_produce_allarmi():
    assert alerts_from([assess(DAILY, _daily(30), NOW)]) == []


# ===========================================================================
# CONTROLLI POSITIVI — la guardia sa segnalare?
# ===========================================================================
def test_un_monitor_fermo_viene_visto():
    """Il guasto piu' semplice: il cron non gira piu'. Ultima barra di 5 giorni fa."""
    r = assess(DAILY, _daily(30, end=NOW - timedelta(days=5)), NOW)
    assert r["status"] == "FERMO"
    assert alerts_from([r])


def test_una_serie_bucata_non_passa_per_fresca():
    """IL test: il monitor gira, l'ultima barra e' di stanotte, ma ha perso il 30% delle barre di
    mezzo. Una guardia di sola freschezza direbbe OK, e il gate verrebbe deciso su meta' serie."""
    stamps = _daily(30, skip=set(range(5, 15)))     # 20 barre su 30 attese
    r = assess(DAILY, stamps, NOW)
    assert r["status"] == "BUCATO"
    assert r["coverage"] < MIN_COVERAGE
    assert r["age_h"] < 24.0                        # ...ed e' fresca: la freschezza non basta


def test_una_serie_assente_e_un_allarme_non_un_silenzio():
    """'Non vedo' non e' 'va tutto bene' — se lo stato e' perso, e' un evento."""
    r = assess(DAILY, [], NOW)
    assert r["status"] == "ASSENTE"
    assert alerts_from([r])


def test_la_soglia_di_copertura_e_quella_del_veto_dvolspread():
    """Riusata di proposito: una soglia diversa per monitor renderebbe i gate non confrontabili."""
    assert MIN_COVERAGE == 0.80
    assert assess(DAILY, _daily(30, skip={5, 6, 7, 8, 9}), NOW)["status"] == "OK"      # 83%
    assert assess(DAILY, _daily(30, skip=set(range(5, 12))), NOW)["status"] == "BUCATO"  # 77%


# ===========================================================================
# giovinezza, che non e' salute
# ===========================================================================
def test_un_monitor_con_una_sola_barra_non_e_giudicabile():
    r = assess(DAILY, _daily(1), NOW)
    assert r["status"] == "NUOVO" and r["coverage"] is None


def test_nuovo_non_allerta_ma_non_e_ok():
    """Un monitor appena partito non deve suonare, ma non deve nemmeno risultare sano:
    XSR01 e DVOLSPREAD hanno 2 barre e mancano mesi al loro gate."""
    r = assess(DAILY, _daily(1), NOW)
    assert alerts_from([r]) == [] and r["status"] != "OK"


# ===========================================================================
# cadenze: sbagliarle vuol dire un falso allarme a settimana
# ===========================================================================
def test_un_monitor_orario_non_viene_scambiato_per_giornaliero():
    """paper_prevday registra a barra oraria: 864 barre in 36 giorni sono complete, non un
    eccesso. Con la cadenza sbagliata la copertura uscirebbe al 2400%."""
    spec = MonitorSpec("h", "h", "returns.jsonl", 1.0)
    r = assess(spec, _daily(864, step_h=1.0), NOW)
    assert r["status"] == "OK" and r["expected"] == 864


def test_il_weekend_non_ferma_un_monitor_di_borsa():
    """paper_combo dipende dalle gambe IB: venerdi' e' l'ultima barra fino a lunedi'. Contare
    il weekend come eta' produrrebbe un allarme ogni lunedi' mattina."""
    ven = datetime(2026, 7, 24, 0, 0, tzinfo=timezone.utc)
    lun = datetime(2026, 7, 27, 12, 0, tzinfo=timezone.utc)
    stamps = [ven - timedelta(days=k) for k in range(20, 0, -1)] + [ven]
    assert assess(EQ, stamps, lun)["status"] != "FERMO"


def test_le_barre_attese_su_calendario_di_borsa_escludono_il_weekend():
    lun = datetime(2026, 7, 20, tzinfo=timezone.utc)
    ven = datetime(2026, 7, 24, tzinfo=timezone.utc)
    assert expected_bars(lun, ven, 24.0, "equity") == 5     # non 5 giorni solari a caso: 5 sedute
    assert expected_bars(lun, ven, 24.0, "crypto") == 5


# ===========================================================================
# escalation: il danno non e' il monitor, e' il gate
# ===========================================================================
def test_un_monitor_che_alimenta_un_gate_lo_dice_nell_allarme():
    a = alerts_from([assess(GATED, _daily(30, end=NOW - timedelta(days=5)), NOW)])
    assert a and "gate" in a[0].lower() and "XSR01" in a[0]


# ===========================================================================
# lettura dei formati reali
# ===========================================================================
def test_legge_jsonl_con_ts_in_millisecondi(tmp_path):
    p = tmp_path / "returns.jsonl"
    p.write_text("\n".join(json.dumps({"ts": 1785024000000 + i * 86_400_000, "net": 0.0})
                           for i in range(3)))
    s = read_stamps(p)
    assert len(s) == 3 and s[0] == datetime(2026, 7, 26, tzinfo=timezone.utc)


def test_legge_equity_csv_con_data_in_prima_colonna(tmp_path):
    p = tmp_path / "equity.csv"
    p.write_text("date,equity\n2026-07-24 00:00:00+00:00,2000.0\n2026-07-27 00:00:00+00:00,2010.0\n")
    s = read_stamps(p)
    assert len(s) == 2 and s[-1].day == 27


def test_una_riga_corrotta_non_fa_esplodere_la_guardia(tmp_path):
    """Una guardia che muore su una riga malformata smette di guardare proprio quando
    qualcosa e' andato storto."""
    p = tmp_path / "returns.jsonl"
    p.write_text('{"ts": 1785024000000}\nnon-json\n{"ts": 1785110400000}\n')
    assert len(read_stamps(p)) == 2


# ===========================================================================
# integrazione sullo stato REALE del progetto
# ===========================================================================
def test_i_monitor_reali_sono_leggibili_e_giudicati():
    """Non asserisce che siano sani (dipende da quando gira il test): asserisce che la guardia
    li trova e si pronuncia su ognuno — il fallimento silenzioso sarebbe una lista vuota."""
    rows = check_all(ROOT, now=NOW)
    assert len(rows) == 6
    assert all(r["status"] in ("OK", "NUOVO", "FERMO", "BUCATO", "ASSENTE") for r in rows)
    assert {"paper_xsr", "paper_statarb", "paper_dvolspread"} <= {r["name"] for r in rows}