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 (zero falsi allarmi su
8 anni). Senza un controllo positivo, "non e' mai scattato" non e' una buona notizia.
"""
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)


# ===========================================================================
# 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")