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