#!/usr/bin/env python """venue_news.py — legge cosa il venue ANNUNCIA. SOLA LETTURA, non blocca mai nulla. PERCHE' ESISTE (2026-08-31). Il debito #8 dice, testuale: *«Resta scoperto il Rulebook vero e proprio: la sonda legge se il venue RISPONDE, non cosa il venue ANNUNCIA.»* Questa e' quella meta'. In una settimana quattro cambi di policy Deribit materiali sono arrivati **a mano**, perche' l'operatore ha incollato dei link — non perche' qualcosa li avesse visti: * tetto sull'USDE (~31,2% dell'equity) * esclusione dell'Italia dai reward USDC * permesso mancante su BUIDL * fine della Proof of Reserves quotidiana E il caso che decide la questione: le **specifiche dei perpetual lineari USDC** sono cambiate il 18/08, annunciate il **14/08**. `check_specs()` rileva la deriva DOPO che e' avvenuta ed e' nato proprio da quella svista; il feed l'avrebbe detto **quattro giorni prima**. Rilevare a posteriori e sapere in anticipo non sono la stessa cosa: la prima volta e' andata bene solo perche' i cambi erano RIDUZIONI (un valore piu' grosso resta conforme). Il giorno che Deribit ALZA un minimo, gli ordini vengono rifiutati. COSA E' E COSA NON E'. E' un promemoria che legge un feed RSS pubblico e dice *«e' uscito questo, guardalo»*. **Non interpreta e non decide**: nessun automatismo su un testo di marketing (P13 — una guardia sui NUMERI non copre il RAGIONAMENTO, e la prosa si legge come opinione di un lettore fallibile). L'unica cosa che classifica e' l'URGENZA, e lo fa su parole DERIVATE dal codice sorvegliato, non ridichiarate qui (P1): gli strumenti vengono da `deribit._CONTRACT`, le valute di collaterale da `config/live.json`. Chi aggiunge uno strumento o una valuta allarga da solo la sorveglianza, senza toccare questo file. uv run python scripts/live/venue_news.py # report + allerta sui nuovi uv run python scripts/live/venue_news.py --quiet # solo allerta (per il cron) """ from __future__ import annotations import argparse import html import json import re import sys from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) from src.live.notifier import notify # noqa: E402 FEED = "https://insights.deribit.com/exchange-updates/feed/" STATE = ROOT / "data" / "live" / "venue_news.jsonl" # Parole che alzano un annuncio a 🚨. Le prime sono FISSE (categorie di rischio che ci toccano # per costruzione); le altre si DERIVANO dal codice, sotto. _CHIAVI_FISSE = ( "contract specification", "margin", "collateral", "fee", "delist", "discontinu", "reserve", "maintenance", "liquidation", "settlement", "withdrawal", "reward", "jurisdiction", "eligib", "insurance fund", "socialized", "adl", "rulebook", ) def chiavi() -> list[str]: """Parole-chiave DERIVATE dal codice sorvegliato (P1), piu' quelle fisse.""" out = set(_CHIAVI_FISSE) try: from src.live.deribit import _CONTRACT for nome in _CONTRACT: # BTC_USDC-PERPETUAL -> "btc", "usdc" for pezzo in re.split(r"[_\-]", nome.lower()): if len(pezzo) >= 3 and pezzo != "perpetual": out.add(pezzo) except Exception: pass try: cfg = json.loads((ROOT / "config" / "live.json").read_text()) out.add(str(cfg.get("usde", {}).get("index_name", "")).split("_")[0].lower()) except Exception: pass return sorted(k for k in out if k) def scarica(url: str = FEED, timeout: float = 20.0) -> str | None: """Feed RSS pubblico (tokenless). None se non leggibile. Mai solleva.""" try: import requests r = requests.get(url, timeout=timeout, headers={"User-Agent": "PythagorasGoal/venue_news (+read-only)"}) return r.text if r.status_code == 200 else None except Exception: return None def voci(xml: str) -> list[dict]: """PURA. Estrae (guid, titolo, data, link) dal RSS. Mai solleva.""" out = [] for blocco in re.findall(r"(.*?)", xml or "", re.S): def campo(tag): m = re.search(rf"<{tag}[^>]*>(?:)?", blocco, re.S) return html.unescape(m.group(1).strip()) if m else "" g = campo("guid") or campo("link") if g: out.append({"guid": g, "titolo": campo("title"), "data": campo("pubDate"), "link": campo("link")}) return out def rilevante(titolo: str, ks: list[str]) -> list[str]: """PURA. Quali parole-chiave compaiono nel titolo.""" t = titolo.lower() return [k for k in ks if k in t] def gia_visti() -> set[str]: if not STATE.exists(): return set() out = set() for riga in STATE.read_text().splitlines(): try: out.add(json.loads(riga)["guid"]) except Exception: pass return out def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--quiet", action="store_true") a = ap.parse_args() xml = scarica() if xml is None: if not a.quiet: print(" ✗ feed non leggibile — 'non vedo' non e' 'niente di nuovo' (P5)") return 2 ks, viste, nuove = chiavi(), gia_visti(), [] primo_giro = not STATE.exists() tutte = voci(xml) STATE.parent.mkdir(parents=True, exist_ok=True) with STATE.open("a") as f: for v in tutte: if v["guid"] in viste: continue v["match"] = rilevante(v["titolo"], ks) v["visto_il"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") v["seed"] = primo_giro f.write(json.dumps(v) + "\n") nuove.append(v) # Primo giro: si semina lo storico SENZA allertare (P9: l'allarme massimo non si spende # per un arretrato di 10 voci, o non verra' letto il giorno che e' vero). if not primo_giro: for v in nuove: if v["match"]: notify("🚨 annuncio Deribit sul NOSTRO percorso", {"titolo": v["titolo"], "tocca": ", ".join(v["match"][:6]), "data": v["data"], "link": v["link"], "azione": "leggerlo: questo sorvegliante NON interpreta"}) else: notify("📌 nuovo annuncio Deribit", {"titolo": v["titolo"], "data": v["data"], "link": v["link"]}) if not a.quiet: print("=" * 78) print(f" VENUE NEWS — {len(tutte)} voci nel feed, {len(nuove)} nuove" + (" (PRIMO GIRO: storico seminato, nessun allarme)" if primo_giro else "")) print("=" * 78) for v in tutte: m = rilevante(v["titolo"], ks) segno = "🚨" if m else ("·" if v["guid"] in viste else "📌") print(f" {segno} {v['data'][:16]:17s} {v['titolo'][:70]}") if m: print(f" tocca: {', '.join(m[:8])}") print(f"\n parole derivate dal codice: {', '.join(ks)}") return 0 if __name__ == "__main__": raise SystemExit(main())