#!/usr/bin/env python """venue_watch.py — runner ORARIO del tripwire di venue. Allerta su Telegram, NON blocca nulla. Vedi `src/live/venue_watch.py` per la taratura (100 bps / 4h a segno costante, zero falsi allarmi su 8 anni, controllo positivo su Bitfinex 2018-19) e per il RUNBOOK a un allarme. uv run python scripts/live/venue_watch.py # un giro, stampa il report uv run python scripts/live/venue_watch.py --quiet # solo in caso di allarme (per il cron) """ from __future__ import annotations import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) from src.live.notifier import notify # noqa: E402 from src.live.venue_watch import (PERSIST_HOURS, THRESHOLD_BPS, # noqa: E402 run_once) def main() -> int: quiet = "--quiet" in sys.argv rep = run_once() if not quiet: print("=" * 78) print(" VENUE WATCH — scarto Deribit vs consenso USD indipendente") print("=" * 78) print(f" soglia: {THRESHOLD_BPS:.0f} bps persistenti {PERSIST_HOURS}h a segno costante") lock = rep["platform_locked"] print(f" public/status locked: {lock if lock is not None else 'non leggibile'}") for a, lvl in rep["levels"].items(): d = rep["detail"][a] bps = f"{d['bps']:+7.1f} bps" if d["bps"] is not None else " n/d " print(f" {a:<4} {lvl:<6} {bps} ({d['n_refs']} referenze, " f"spread {d['ref_spread_bps']:.1f} bps)" if d["ref_spread_bps"] is not None else f" {a:<4} {lvl:<6} {bps} ({d['n_refs']} referenze)") if rep["alerts"]: # ⚠️ Il titolo dice cosa FARE, non solo cosa e' successo: chi lo legge sul telefono deve # sapere il primo passo senza aprire il repo (runbook completo in src/live/venue_watch.py). notify("🚨 VENUE WATCH — possibile stress su Deribit. PRIMO PASSO: prelievo di prova", {f"alert {i+1}": a for i, a in enumerate(rep["alerts"])}) for a in rep["alerts"]: print(f" ALERT: {a}") return 2 return 0 if __name__ == "__main__": raise SystemExit(main())