Adriano
1a7f670966
src/live/giri.registra_giro: una riga jsonl {ts ms, campi} dopo una misura completata,
mai un'eccezione. edge_watch, fee_watch, scale_watch, vrp_f_watch, venue_news (cron_daily)
e venue_watch (cron_book, orario) scrivono data/<dir>/runs.jsonl; niente riga senza misura.
monitor_health: sei MonitorSpec (30h i giornalieri, 3h venue_watch), open_labeled=False.
Stessa riparazione di cc01 (#10, #12), generalizzata. Primo giro notturno: NUOVO, non allerta.
Test: incrocio P1 percorso/RUNS, misura -> una riga, non-misura -> nessuna. Suite: 1098 passed.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
78 lines
3.2 KiB
Python
78 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Sorveglianza giornaliera della CHIAVE DI SCALA del libro (SPEC-scale-key §6.4).
|
|
|
|
uv run python scripts/live/scale_watch.py # osserva e stampa, invia se ALLARME
|
|
uv run python scripts/live/scale_watch.py --secco # osserva e stampa, NON invia nulla
|
|
uv run python scripts/live/scale_watch.py --quiet # una riga sola (per il cron)
|
|
|
|
Tre domande, tre azioni diverse, tre stati (OK / ALLARME / NON MISURABILE).
|
|
Non impedisce la modifica: la rende visibile entro 24 ore e attribuibile.
|
|
"""
|
|
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.giri import registra_giro # noqa: E402
|
|
from src.live.notifier import notify # noqa: E402
|
|
from src.live.scale_watch import ALLARME, OK, run_once # noqa: E402
|
|
|
|
RUNS = ROOT / "data" / "scale_watch" / "runs.jsonl"
|
|
|
|
|
|
def _sender(testo: str):
|
|
try:
|
|
notify(testo)
|
|
return True, "telegram"
|
|
except Exception as e: # P3: si registra dove lo si ingoia
|
|
return False, f"{type(e).__name__}: {e}"
|
|
|
|
|
|
def main() -> None:
|
|
secco = "--secco" in sys.argv
|
|
quiet = "--quiet" in sys.argv
|
|
rep = run_once(sender=None if secco else _sender)
|
|
registra_giro(RUNS, stato=rep["stato"])
|
|
icona = {OK: "✅", ALLARME: "🚨"}.get(rep["stato"], "⚠️")
|
|
if quiet:
|
|
fb = rep["fallback"]
|
|
q = f"{100*fb['quota']:.2f}%" if "quota" in fb else "n/d"
|
|
print(f" scale_watch: {icona} {rep['stato']} · scala {rep['scala_dichiarata']:.2f}x · "
|
|
f"fallback {q} · {rep['invio']}")
|
|
return
|
|
print("=" * 88)
|
|
print(f" SCALE WATCH — la chiave di scala del libro {icona} {rep['stato']}")
|
|
print("=" * 88)
|
|
print(f" scala dichiarata : {rep['scala_dichiarata']:.2f}x "
|
|
f"(scaletta {rep['ladder']}, tetto {rep['tetto']:.2f}x — costanti di CODICE)")
|
|
for nome, d in rep["domande"].items():
|
|
print(f"\n {nome} [{d['stato']}] {d['motivo']}")
|
|
if d.get("azione") and d["azione"] != "nessuna":
|
|
print(f" → azione: {d['azione']}")
|
|
fb = rep["fallback"]
|
|
if fb.get("stato") == "NON MISURABILE":
|
|
print(f"\n fallback [{fb['stato']}] {fb.get('motivo')}")
|
|
else:
|
|
print(f"\n fallback [{fb['stato']}] equity illeggibile in {fb['fallback_veri']}/{fb['giri']} "
|
|
f"giri ({100*fb['quota']:.2f}%, soglia {100*fb['soglia']:.0f}%) — "
|
|
f"{fb['paper']} giri pre-finanziamento esclusi, li' il libro non invia affatto")
|
|
print(f"\n lettore di produzione [{rep['lettore']['stato']}] {rep['lettore']['motivo'][:150]}")
|
|
print(f"\n invio: {rep['invio']}")
|
|
|
|
|
|
USO = """uso: scale_watch.py [--quiet] [--secco]
|
|
|
|
uv run python scripts/live/scale_watch.py # osserva e stampa, invia se ALLARME
|
|
uv run python scripts/live/scale_watch.py --secco # osserva e stampa, NON invia nulla
|
|
uv run python scripts/live/scale_watch.py --quiet # una riga sola (per il cron)
|
|
|
|
Dettaglio nel docstring in testa al file."""
|
|
|
|
if __name__ == "__main__":
|
|
from src.live.cli import valida
|
|
valida("scale_watch.py", USO, flag=("--quiet", "--secco"), con_valore=())
|
|
main()
|