"""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") # =========================================================================== # 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}, }) == []