PythagorasGoal/tests/test_venue_watch.py

"""Test del tripwire di venue (src/live/venue_watch.py + scripts/research/r0726_venue_tripwire.py).

Il test che conta di piu' e' `test_controllo_positivo_*`: un rilevatore che non segnala mai nulla
e' indistinguibile da uno rotto, e questo qui e' TARATO per non segnalare. Senza un controllo
positivo, "non e' mai scattato" non e' una buona notizia.

⚠️ CORREZIONE 2026-08-21 (`scripts/research/r0821_venue_refs.py`): la taratura NON e' "zero falsi
allarmi in 8 anni" sull'insieme di referenze che gira in PRODUZIONE. Quel numero fu misurato con
`bitfinex` nel consenso, che il sorvegliante live non ha mai avuto; sul consenso reale (Coinbase +
Bitstamp, e da oggi + Kraken) i falsi allarmi in 8 anni sono **1**, il 2020-03-13 (crash COVID,
4 ore a -418 bps di picco su BTC) — cioe' proprio l'evento che la memoria citava come esempio di
cio' su cui NON scattava. Il punto (100 bps, 4h) resta invariato per economia, non per assenza di
falsi allarmi: 1 ogni 8 anni costa ~0.031%/anno di equity attesa contro il 100% che evita.
"""
from __future__ import annotations

import json
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"))

from src.live.venue_watch import (BLIND_ALERT_HOURS, PERSIST_HOURS,  # noqa: E402
                                  THRESHOLD_BPS, AssetState, WatchState,
                                  dislocation_bps, load_state, save_state, step)
import src.live.venue_watch as VW  # noqa: E402  — i test sul trasporto patchano `observe`


# ===========================================================================
# consenso: quando NON si puo' misurare
# ===========================================================================
def test_una_sola_referenza_non_e_un_consenso():
    """Con una referenza sola non si distingue 'Deribit e' fuori' da 'la referenza e' rotta'."""
    bps, n, spread = dislocation_bps(100.0, [100.0])
    assert bps is None and n == 1


def test_referenze_in_disaccordo_producono_non_misurabile():
    """Se le referenze litigano fra loro il problema e' loro: scartare, non allertare."""
    bps, n, spread = dislocation_bps(100.0, [100.0, 103.0])   # 300 bps di spread
    assert bps is None
    assert spread > 100.0


def test_consenso_normale_da_scarto_piccolo():
    bps, n, spread = dislocation_bps(100.05, [100.0, 100.02])
    assert n == 2
    assert 0 < bps < 10


def test_lo_scarto_e_firmato_in_entrambe_le_direzioni():
    """Premio e sconto sono ENTRAMBI segnale: Mt.Gox andava a premio, un venue in fuga a sconto."""
    up, _, _ = dislocation_bps(101.0, [100.0, 100.0])
    dn, _, _ = dislocation_bps(99.0, [100.0, 100.0])
    assert up > 0 and dn < 0
    assert abs(up + dn) < abs(up) * 0.05     # simmetrici a meno del secondo ordine


# ===========================================================================
# macchina a stati
# ===========================================================================
def test_sotto_soglia_resta_ok():
    st, lvl = step(AssetState(), THRESHOLD_BPS - 1)
    assert lvl == "OK" and st.streak_hours == 0


def test_servono_persist_ore_consecutive_per_allertare():
    st = AssetState()
    for i in range(PERSIST_HOURS - 1):
        st, lvl = step(st, THRESHOLD_BPS + 50)
        assert lvl == "WATCH", f"ora {i+1}: allarme troppo presto"
    st, lvl = step(st, THRESHOLD_BPS + 50)
    assert lvl == "ALERT"


def test_il_cambio_di_segno_azzera_lo_streak():
    """E' il cuore della discriminazione: uno scarto che rimbalza non e' un arbitraggio rotto."""
    st = AssetState()
    for _ in range(PERSIST_HOURS - 1):
        st, _ = step(st, THRESHOLD_BPS + 50)
    st, lvl = step(st, -(THRESHOLD_BPS + 50))     # stesso modulo, segno opposto
    assert lvl == "WATCH" and st.streak_hours == 1


def test_un_rientro_sotto_soglia_azzera_lo_streak():
    st = AssetState()
    for _ in range(PERSIST_HOURS - 1):
        st, _ = step(st, THRESHOLD_BPS + 50)
    st, lvl = step(st, 5.0)
    assert lvl == "OK" and st.streak_hours == 0


def test_non_rispamma_ogni_ora_durante_lo_stesso_streak():
    st = AssetState()
    levels = []
    for _ in range(PERSIST_HOURS + 5):
        st, lvl = step(st, THRESHOLD_BPS + 50)
        levels.append(lvl)
    assert levels.count("ALERT") == 1, f"allarmi ripetuti: {levels}"


def test_dopo_un_alert_un_nuovo_streak_di_segno_opposto_riallerta():
    """Il de-spam non deve nascondere un evento NUOVO."""
    st = AssetState()
    for _ in range(PERSIST_HOURS):
        st, _ = step(st, THRESHOLD_BPS + 50)
    st, _ = step(st, 0.0)                       # rientro
    levels = []
    for _ in range(PERSIST_HOURS):
        st, lvl = step(st, -(THRESHOLD_BPS + 50))
        levels.append(lvl)
    assert levels[-1] == "ALERT"


# ===========================================================================
# lo stato BLIND: "non vedo" non e' "va bene"
# ===========================================================================
def test_dati_mancanti_azzerano_lo_streak_non_lo_accumulano():
    st = AssetState()
    for _ in range(PERSIST_HOURS - 1):
        st, _ = step(st, THRESHOLD_BPS + 50)
    st, lvl = step(st, None)
    assert st.streak_hours == 0, "evidenza accumulata su dati che non parlano"


def test_cecita_prolungata_diventa_essa_stessa_un_allarme():
    st = AssetState()
    for i in range(BLIND_ALERT_HOURS - 1):
        st, lvl = step(st, None)
        assert lvl == "OK"
    st, lvl = step(st, None)
    assert lvl == "BLIND"


def test_una_misura_valida_azzera_il_contatore_di_cecita():
    st = AssetState()
    for _ in range(BLIND_ALERT_HOURS):
        st, _ = step(st, None)
    st, lvl = step(st, 1.0)
    assert st.blind_hours == 0 and lvl == "OK"


# ===========================================================================
# persistenza
# ===========================================================================
def test_stato_salvato_e_riletto(tmp_path):
    p = tmp_path / "state.json"
    ws = WatchState(assets={"BTC": AssetState(streak_hours=3, sign=-1)}, last_ts=123)
    save_state(ws, p)
    back = load_state(p)
    assert back.assets["BTC"].streak_hours == 3
    assert back.assets["BTC"].sign == -1
    assert back.last_ts == 123


def test_stato_corrotto_non_fa_crashare_il_cron(tmp_path):
    """Un JSON rotto non deve rompere il giro orario del book."""
    p = tmp_path / "state.json"
    p.write_text("{ questo non e' json")
    st = load_state(p)
    assert isinstance(st, WatchState) and st.last_ts == 0


# ===========================================================================
# nucleo di ricerca: episodi e outer-join
# ===========================================================================
def _tw():
    import r0726_venue_tripwire as T
    return T


def test_il_consenso_usa_outer_join_non_intersezione():
    """Il bug della prima corsa: una referenza corta troncava il campione da 8 anni a 29 giorni."""
    T = _tw()
    d = pd.Series([100.0] * 10, index=range(10))
    lunga = pd.Series([100.0] * 10, index=range(10))
    corta = pd.Series([100.0] * 3, index=range(3))
    f = T.dislocation(d, [lunga, corta])
    assert len(f) == 10, "inner join: la referenza corta decide il campione"


def test_episodi_richiedono_segno_costante():
    T = _tw()
    idx = range(10)
    alternato = pd.Series([200.0, -200.0] * 5, index=idx)
    usable = pd.Series([True] * 10, index=idx)
    assert T.episodes(alternato, usable, 100, 3) == []
    costante = pd.Series([200.0] * 10, index=idx)
    assert len(T.episodes(costante, usable, 100, 3)) == 1


def test_le_ore_non_utilizzabili_rompono_il_run():
    T = _tw()
    idx = range(10)
    bps = pd.Series([200.0] * 10, index=idx)
    usable = pd.Series([True, True, False, True, True, True, True, True, True, True], index=idx)
    eps = T.episodes(bps, usable, 100, 6)
    assert len(eps) == 1 and eps[0]["hours"] == 7, "un buco di dati non deve saldare due run"


# ===========================================================================
# CONTROLLI POSITIVI — senza questi "non scatta mai" non e' una buona notizia
# ===========================================================================
def test_controllo_positivo_firma_di_venue_gated():
    """Dislocazione grande, persistente e a segno costante = deve scattare."""
    T = _tw()
    n = 200
    idx = range(n)
    bps = pd.Series([450.0] * n, index=idx)      # premio tipo Bitfinex 2018
    usable = pd.Series([True] * n, index=idx)
    eps = T.episodes(bps, usable, THRESHOLD_BPS, PERSIST_HOURS)
    assert len(eps) == 1 and eps[0]["hours"] == n and eps[0]["sign"] == 1


def test_controllo_positivo_la_macchina_a_stati_scatta_sulla_stessa_firma():
    """Lo stesso episodio, passato dalla macchina a stati di produzione, deve produrre ALERT."""
    st = AssetState()
    out = []
    for _ in range(24):
        st, lvl = step(st, 450.0)
        out.append(lvl)
    assert "ALERT" in out


def test_controllo_negativo_un_crash_breve_non_scatta():
    """Marzo 2020 su Deribit: 12h a 158 bps di picco ma sotto i 100 bps per meno di 4h di fila
    alla soglia scelta. Qui la versione stilizzata: uno spike di 3 ore non deve allertare."""
    st = AssetState()
    out = []
    for bps in (150.0, 150.0, 150.0, 10.0, 5.0):
        st, lvl = step(st, bps)
        out.append(lvl)
    assert "ALERT" not in out, f"spike breve ha allertato: {out}"


def test_la_taratura_congelata_non_cambia_per_sbaglio():
    """La taratura VALE 'zero falsi allarmi su 8 anni': se qualcuno la cambia, il claim decade
    e va rimisurato con r0726_venue_tripwire.py."""
    assert THRESHOLD_BPS == 100.0
    assert PERSIST_HOURS == 4


def test_il_watch_e_cablato_nel_cron_orario():
    sh = (ROOT / "scripts" / "cron_book.sh").read_text()
    assert "venue_watch.py" in sh
    assert sh.index("venue_watch.py") < sh.index("book_execute.py"), (
        "il watch deve girare PRIMA dell'esecuzione: se Deribit e' in stress l'allarme deve "
        "partire anche quando book_execute fallisce per la stessa ragione")


# ===========================================================================
# Disciplina del blocco piattaforma (2026-08-19). Il 18/08 sono usciti QUATTRO
# 🚨 identici con "PRIMO PASSO: prelievo di prova" per una manutenzione
# annunciata, e tutti e quattro DOPO che il book aveva gia' ripreso a eseguire.
# ===========================================================================
from src.live.venue_watch import LockState, is_maintenance, lock_step  # noqa: E402


def _replay(sequenza, grace=2):
    st, out = LockState(), []
    for locked, maint in sequenza:
        st, lvl = lock_step(st, locked, maint, grace=grace)
        out.append(lvl)
    return out


def test_manutenzione_breve_e_un_avviso_morbido_non_il_runbook_del_prelievo():
    assert _replay([(True, True), (False, False)]) == ["MAINT", "RIENTRATO"]


def test_non_si_ripete_ogni_ora_mentre_la_condizione_dura():
    """Il difetto vero del 18/08: nessuna memoria, quindi un messaggio identico ogni giro."""
    livelli = _replay([(True, True)] * 5 + [(False, False)])
    assert livelli.count("MAINT") == 1
    assert livelli.count("ALERT") == 1
    assert livelli == ["MAINT", "MUTO", "ALERT", "MUTO", "MUTO", "RIENTRATO"]


def test_la_manutenzione_che_sfora_RIALZA_invece_di_restare_morbida():
    """Deribit aveva annunciato 15-30 minuti e la piattaforma e' rimasta bloccata per ore:
    una manutenzione che dura sei volte l'annuncio torna a essere una notizia."""
    assert _replay([(True, True)] * 3)[-1] == "ALERT"


def test_un_blocco_senza_manutenzione_dichiarata_e_subito_il_caso_serio():
    assert _replay([(True, False)])[0] == "ALERT"


def test_status_illeggibile_non_e_un_rientro():
    """Dichiarare 'e' rientrato' perche' non si e' riusciti a guardare sarebbe la bugia peggiore."""
    st = LockState()
    st, _ = lock_step(st, True, False)
    prima = st.hours
    st, lvl = lock_step(st, None, False)
    assert lvl == "MUTO" and st.hours == prima


def test_dopo_un_rientro_un_nuovo_blocco_riallerta():
    """L'ammutolimento vale per QUESTO episodio, non per sempre."""
    livelli = _replay([(True, False), (False, False), (True, False)])
    assert livelli == ["ALERT", "RIENTRATO", "ALERT"]


def test_riconosce_la_firma_della_manutenzione_deribit():
    assert is_maintenance(['OnMaintenance: deribit {"error":{"message":"system_maintenance",'
                           '"code":11051}}'])
    assert not is_maintenance(["HTTPError: 502 Bad Gateway"])
    assert not is_maintenance([])


def test_lo_stato_del_lock_sopravvive_al_giro_successivo(tmp_path):
    """Senza persistenza la memoria si perde a ogni cron e si ricomincia a gridare."""
    from src.live.venue_watch import WatchState, load_state, save_state
    p = tmp_path / "s.json"
    st = WatchState()
    st.lock, _ = lock_step(st.lock, True, True)
    save_state(st, p)
    assert load_state(p).lock.hours == 1 and load_state(p).lock.alerted_soft


# ===========================================================================
# Specifiche contratto vs venue (2026-08-19)
# ===========================================================================
from src.live.deribit import compare_specs  # noqa: E402

_VERO = {"ETH_USDC-PERPETUAL": {"tick_size": 0.01, "min_trade_amount": 0.0001,
                                "contract_size": 0.0001}}


def test_dichiarato_piu_grosso_e_solo_granularita():
    """La situazione del 18/08: conforme, si perde precisione. Non e' un'emergenza."""
    d = compare_specs({"ETH_USDC-PERPETUAL": {"tick": 0.05, "min": 0.001, "step": 0.001}}, _VERO)
    assert d and all(x["rischio"] == "granularita'" for x in d)


def test_dichiarato_piu_fine_significa_ordini_RIFIUTATI():
    """L'altra direzione, quella che costa: il venue alza un minimo e noi non lo sappiamo."""
    d = compare_specs({"ETH_USDC-PERPETUAL": {"tick": 0.001, "min": 1e-5, "step": 1e-5}}, _VERO)
    assert d and all(x["rischio"] == "rifiuto" for x in d)


def test_uno_strumento_non_letto_non_e_una_divergenza():
    """Silenzio e uguaglianza non sono la stessa cosa: cio' che non si e' letto va dichiarato
    a parte (`non_letti`), non fatto passare per 'combacia'."""
    assert compare_specs({"X": {"tick": 1.0}}, {}) == []


def test_la_tabella_dichiarata_combacia_col_venue_oggi():
    """Verificato a mano il 19/08. Se questo rompe, o Deribit ha cambiato le specifiche o
    qualcuno ha toccato _CONTRACT: in entrambi i casi va guardato, non silenziato."""
    from src.live.deribit import _CONTRACT
    assert compare_specs(_CONTRACT, {
        "BTC_USDC-PERPETUAL": {"tick_size": 0.1, "min_trade_amount": 0.0001,
                               "contract_size": 0.0001},
        "ETH_USDC-PERPETUAL": {"tick_size": 0.01, "min_trade_amount": 0.0001,
                               "contract_size": 0.0001},
        "BTC-PERPETUAL": {"tick_size": 0.5, "min_trade_amount": 10.0, "contract_size": 10.0},
        "ETH-PERPETUAL": {"tick_size": 0.05, "min_trade_amount": 1.0, "contract_size": 1.0},
    }) == []


# ===========================================================================
# MECCANICA scoperta il 2026-08-21 ri-misurando la taratura a tre referenze.
# Sono i due fatti PURI che spiegano i numeri di r0821_venue_refs.py; i numeri
# empirici stanno nel diario, riprodotti da quello script.
# ===========================================================================
def test_una_terza_referenza_non_puo_aumentare_le_ore_utilizzabili():
    """Aggiungere una referenza puo' solo ALLARGARE lo spread max-min, mai stringerlo ->
    la quota di ore utilizzabili e' monotona NON crescente nel numero di referenze.

    E' la ragione strutturale per cui "piu' referenze" non e' gratis: la modifica del 19/08
    sposta il sistema verso BLIND, che e' lo stato morbido (allerta di MENO, non di piu').
    """
    base = [100.0, 100.4]                       # spread 40 bps: concordi
    ok, n, spread = dislocation_bps(100.2, base)
    assert ok is not None and n == 2 and spread < THRESHOLD_BPS

    # una terza referenza lontana allarga lo spread oltre la soglia -> BLIND, non allarme
    fuori, n3, spread3 = dislocation_bps(100.2, base + [102.0])
    assert n3 == 3 and spread3 > spread, "lo spread deve essere monotono nel numero di referenze"
    assert fuori is None, "referenze in disaccordo -> si SCARTA il campione, non si allerta"

    # e una terza referenza CONCORDE non toglie nulla: il caso Kraken misurato il 21/08
    dentro, n3b, spread3b = dislocation_bps(100.2, base + [100.2])
    assert dentro is not None and n3b == 3 and spread3b >= spread


def test_una_sola_ora_BLIND_spezza_lo_streak_e_impedisce_l_allarme():
    """⚠️ Il fatto che rende ambiguo un "zero falsi allarmi" misurato con piu' referenze.

    Lo streak si AZZERA su un'ora BLIND (non si accumula evidenza su dati che non parlano).
    Quindi un insieme di referenze che manda in BLIND una sola ora dentro una finestra di
    dislocazione fa SPARIRE l'episodio senza che la dislocazione sia diminuita: e' esattamente
    cio' che e' successo al falso allarme del 2020-03-13, presente a 2 referenze e assente a 3
    perche' 1 delle 4 ore diventava BLIND (dislocazione mediana ancora -343 bps).

    Morale: "zero falsi allarmi" puo' voler dire "piu' accurato" oppure "piu' cieco", e le due
    si distinguono solo guardando la quota di ore utilizzabili.
    """
    st = AssetState()
    # 4 ore consecutive oltre soglia a segno costante -> ALERT alla quarta
    for _ in range(PERSIST_HOURS - 1):
        st, lvl = step(st, -400.0)
        assert lvl == "WATCH"
    st_alert, lvl = step(st, -400.0)
    assert lvl == "ALERT", "il caso di controllo deve scattare, altrimenti il test non prova nulla"

    # stessa dislocazione, ma con UNA ora cieca in mezzo -> nessun allarme
    st2 = AssetState()
    for bps in (-400.0, -400.0, None, -400.0):
        st2, lvl2 = step(st2, bps)
    assert lvl2 != "ALERT"
    assert st2.streak_hours < PERSIST_HOURS, "l'ora BLIND deve azzerare lo streak"


# =============================================================================================
# IL TRASPORTO — un 🚨 perso non deve costare l'EPISODIO INTERO
# =============================================================================================
# Fino al 2026-08-28 `run_once` salvava lo stato coi marcatori `alerted` gia' a True e l'invio
# lo faceva il chiamante DOPO. Col 6,9% di invii falliti misurato (2 su 29), un allarme perso
# restava perso finche' durava l'episodio — e gli episodi storici durano 200-2.324 ore.
# Questi test guardano l'unica cosa che conta: dopo un invio FALLITO, l'ora dopo ri-allerta?

def _obs_finto(monkeypatch, bps, locked=False, maint=False):
    def fake():
        return {"ts": 1, "platform_locked": locked, "locked_raw": str(locked).lower(),
                "maintenance": maint,
                "assets": {a: dict(deribit=1.0, n_refs=3, ref_spread_bps=1.0, bps=bps)
                           for a in VW.ASSETS},
                "deribit_errors": []}
    monkeypatch.setattr(VW, "observe", fake)


def _porta_ad_alert(monkeypatch, tmp_path, sender):
    """Fa salire lo streak fino all'ALERT e ritorna il report del giro che allerta."""
    sp = tmp_path / "vw.json"
    _obs_finto(monkeypatch, bps=VW.THRESHOLD_BPS * 3)
    rep = None
    for _ in range(VW.PERSIST_HOURS):
        rep = VW.run_once(state_path=sp, sender=sender)
    return sp, rep


def test_invio_fallito_NON_consuma_l_allarme(monkeypatch, tmp_path):
    ko = lambda rep: (False, "ConnectionError: rete giu'")
    sp, rep = _porta_ad_alert(monkeypatch, tmp_path, ko)
    assert any(l == "ALERT" for l in rep["levels"].values())
    assert rep["invio"].startswith("FALLITO")
    # il marcatore 'gia' detto' NON deve essere sul disco
    st = VW.load_state(sp)
    assert all(not s.alerted for s in st.assets.values()), "un invio fallito ha consumato l'allarme"
    # l'ora dopo, stessa dislocazione: deve RI-allertare
    rep2 = VW.run_once(state_path=sp, sender=ko)
    assert any(l == "ALERT" for l in rep2["levels"].values()), "l'allarme non e' stato ritentato"


def test_invio_riuscito_consuma_l_allarme_una_volta_sola(monkeypatch, tmp_path):
    """Il controllo positivo: senza questo, 'ri-allerta sempre' passerebbe il test di sopra."""
    ok = lambda rep: (True, "ok")
    sp, rep = _porta_ad_alert(monkeypatch, tmp_path, ok)
    assert any(l == "ALERT" for l in rep["levels"].values())
    assert rep["invio"] == "inviato"
    rep2 = VW.run_once(state_path=sp, sender=ok)
    assert not any(l == "ALERT" for l in rep2["levels"].values()), "allarme ripetuto ogni ora"
    assert all(l == "WATCH" for l in rep2["levels"].values())


def test_lock_maint_fallito_ritenta_ma_le_ORE_continuano_a_correre(monkeypatch, tmp_path):
    """Disfare anche le ore congelerebbe l'escalation proprio mentre non si riesce a parlare:
    una manutenzione che sfora la grazia deve salire ad ALERT anche col trasporto giu'."""
    sp = tmp_path / "vw.json"
    ko = lambda rep: (False, "rete giu'")
    _obs_finto(monkeypatch, bps=0.0, locked=True, maint=True)
    livelli = [VW.run_once(state_path=sp, sender=ko)["lock_level"]
               for _ in range(VW.MAINT_GRACE_HOURS + 1)]
    assert livelli[0] == "MAINT"
    assert livelli[1] == "MAINT", "il MAINT perso non e' stato ritentato"
    assert livelli[-1] == "ALERT", "le ore non sono avanzate: l'escalation si e' congelata"


def test_rientro_perso_si_riannuncia(monkeypatch, tmp_path):
    """Il rientro si dice UNA volta e azzera le ore: senza ripristinare lo stato precedente
    non ci sarebbe piu' un rientro da annunciare."""
    sp = tmp_path / "vw.json"
    _obs_finto(monkeypatch, bps=0.0, locked=True, maint=False)
    VW.run_once(state_path=sp, sender=lambda r: (True, "ok"))
    _obs_finto(monkeypatch, bps=0.0, locked=False)
    assert VW.run_once(state_path=sp, sender=lambda r: (False, "giu'"))["lock_level"] == "RIENTRATO"
    assert VW.run_once(state_path=sp, sender=lambda r: (True, "ok"))["lock_level"] == "RIENTRATO"


def test_senza_sender_il_report_lo_DICE(monkeypatch, tmp_path):
    """'non tentato' non e' 'inviato': il silenzio non deve leggersi come un invio riuscito."""
    sp, rep = _porta_ad_alert(monkeypatch, tmp_path, None)
    assert "non tentato" in rep["invio"]