PythagorasGoal/scripts/live/venue_news.py

#!/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"<item>(.*?)</item>", xml or "", re.S):
        def campo(tag):
            m = re.search(rf"<{tag}[^>]*>(?:<!\[CDATA\[)?(.*?)(?:\]\]>)?</{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())