"""cuscino_watch: il cuscino USDC di regolamento e' sorvegliato, e sotto zero si riconverte. Blindano: (a) il cuscino e' DERIVATO da config + book (P1), la stessa formula di usde_convert; (b) i tre stati e la soglia di preavviso; (c) la transizione non si consuma se l'invio fallisce (debito §5.2); (d) le guardie della riconversione automatica — interruttore del libro, depeg, cadenza — e la quota di ripristino che lascia margine.""" from __future__ import annotations import importlib.util import json import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from src.live import usde as U # noqa: E402 def _carica(nome): sp = importlib.util.spec_from_file_location(nome, ROOT / "scripts" / "live" / f"{nome}.py") m = importlib.util.module_from_spec(sp) sp.loader.exec_module(m) return m W = _carica("cuscino_watch") # ------------------------------ il cuscino e' derivato, non ridichiarato ------------------------------ def test_cuscino_derivato_da_config_e_book(): cfg = json.loads((ROOT / "config" / "live.json").read_text()) from src.live import book as B f, come = U.frazione_cuscino() assert abs(f - 2 * float(B.WEIGHT) * float(cfg["disaster_sl_pct"])) < 1e-12 usd, _ = U.cuscino_richiesto_usd(1000.0) assert abs(usd - 1000.0 * f) < 1e-9 assert "SL" in come def test_usde_convert_usa_la_stessa_formula(): C = _carica("usde_convert") assert C.cuscino_richiesto_usd(4469.46) == U.cuscino_richiesto_usd(4469.46) def test_quota_ripristino_lascia_margine_sul_cuscino(): f, _ = U.frazione_cuscino() q0, q10 = U.quota_ripristino(0.0), U.quota_ripristino(0.10) assert abs(q0 - (1 - f)) < 1e-12 # a margine zero e' il complemento del cuscino assert q10 < q0 # col margine si vende UN PO' di piu' # dopo la riconversione a q10 lo slack e' esattamente il 10% del cuscino E = 4000.0 usdc_dopo = E * (1 - q10) assert abs((usdc_dopo - E * f) - 0.10 * E * f) < 1e-9 # ------------------------------ i tre stati ------------------------------ def test_stati_ok_preavviso_scoperto(): assert W.giudica(1500.0, 4000.0, 1200.0)["stato"] == "OK" # slack 300 > 120 g = W.giudica(1300.0, 4000.0, 1200.0) assert g["stato"] == "PREAVVISO" and abs(g["slack"] - 100.0) < 1e-9 and g["preavviso_usd"] == 120.0 assert W.giudica(1199.0, 4000.0, 1200.0)["stato"] == "SCOPERTO" def test_non_vedo_e_blind_non_ok(): assert W.giudica(None, 4000.0, 1200.0)["stato"] == "BLIND" assert W.giudica(1500.0, None, None)["stato"] == "BLIND" # ------------------------------ transizione e trasporto ------------------------------ def test_ok_non_allerta_mai(): assert not W.transizione(None, "OK") and not W.transizione({"stato": "SCOPERTO"}, "OK") def test_transizione_alla_prima_e_al_cambio_di_stato(): assert W.transizione(None, "PREAVVISO") assert W.transizione({"stato": "OK", "allertato": None}, "PREAVVISO") assert W.transizione({"stato": "PREAVVISO", "allertato": True}, "SCOPERTO") assert not W.transizione({"stato": "PREAVVISO", "allertato": True}, "PREAVVISO") def test_un_invio_fallito_non_consuma_la_transizione(): """Debito §5.2: il marcatore 'gia' detto' e' l'esito dell'invio.""" assert W.transizione({"stato": "SCOPERTO", "allertato": False}, "SCOPERTO") assert W.transizione({"stato": "SCOPERTO", "allertato": None}, "SCOPERTO") # ------------------------------ guardie della riconversione ------------------------------ def test_disarmare_il_libro_disarma_la_riconversione(): ok, m = W.puo_riconvertire([], 0, False, 1.0, 0.99) assert not ok and "execution_enabled" in m def test_sotto_depeg_non_si_vende_al_buio(): ok, m = W.puo_riconvertire([], 0, True, 0.985, 0.99) assert not ok and "depeg" in m ok, m = W.puo_riconvertire([], 0, True, None, 0.99) assert not ok and "leggibile" in m def test_un_tentativo_ogni_sei_ore(): now = 1_789_000_000_000 prev = [{"ts": now - 2 * 3_600_000, "riconversione": {"tentata": True}}] ok, m = W.puo_riconvertire(prev, now, True, 1.0, 0.99) assert not ok and "non si insiste" in m prev = [{"ts": now - 7 * 3_600_000, "riconversione": {"tentata": True}}] assert W.puo_riconvertire(prev, now, True, 1.0, 0.99)[0] # un record SENZA riconversione non conta come tentativo assert W.puo_riconvertire([{"ts": now - 1000, "riconversione": None}], now, True, 1.0, 0.99)[0] def test_riconverti_lancia_lo_script_vero_con_esegui(monkeypatch): chiamate = [] class R: returncode, stdout, stderr = 0, " → piano valido\n ESITO: 10 USDE\n", "" def finto_run(cmd, **kw): chiamate.append(cmd) return R() monkeypatch.setattr(W.subprocess, "run", finto_run) e = W.riconverti(0.67) assert e["ok"] and chiamate and chiamate[0][1].endswith("usde_convert.py") assert "--esegui" in chiamate[0] and "0.6700" in chiamate[0] assert e["coda"][-1].startswith(" ESITO") def test_il_monitor_e_registrato_in_monitor_health(): from src.live.monitor_health import MONITORS spec = [m for m in MONITORS if m.name == "cuscino_watch"] assert spec and spec[0].cadence_h == 1.0 and spec[0].open_labeled is False def test_un_non_tentativo_non_consuma_il_budget(): """Prezzo illeggibile, --secco, interruttore spento: `tentata=False` non blocca il giro dopo.""" now = 1_789_000_000_000 prev = [{"ts": now - 3_600_000, "riconversione": {"tentata": False, "motivo": "--secco"}}] assert W.puo_riconvertire(prev, now, True, 1.0, 0.99)[0] def test_dopo_il_ripristino_lo_stato_e_OK_non_preavviso(): """Il margine di ripristino deve superare la soglia di preavviso, o ogni 🚨 e' seguito da un ⚠️ per costruzione (P14). Include il floor di 1 USDE e la tolleranza di quota di usde_convert.""" assert W.MARGINE_RIPRISTINO > W.PREAVVISO_FRAC f, _ = U.frazione_cuscino() E = 4450.0 q = U.quota_ripristino(W.MARGINE_RIPRISTINO) + 0.005 # usde_convert si ferma entro TOLL usdc_dopo = E * (1 - q) - 1.0 # e vende fino a 1 USDE in meno assert W.giudica(usdc_dopo, E, E * f)["stato"] == "OK" def test_cron_installato_e_dichiarato_fuori_dal_tondo_e_dopo_il_book(): """Legge la crontab VIVA (come test_book_cadenza), non un commento nello .sh (P1).""" import shutil import subprocess s = (ROOT / "scripts" / "cron_cuscino.sh").read_text() assert "cuscino_watch.py --quiet" in s if not shutil.which("crontab"): return # SALTATO: distinto da 'assente' out = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=10) if out.returncode != 0 or "PythagorasGoal" not in out.stdout: return # altra macchina / altro utente righe = [ln for ln in out.stdout.splitlines() if "cron_cuscino.sh" in ln and not ln.startswith("#")] assert len(righe) == 1, "cron_cuscino.sh deve essere installato UNA volta" minuto, ora = righe[0].split()[:2] assert minuto == "53" and ora == "*" # orario, dopo il book (:47)