"""cuscino_watch: il cuscino USDC e' sorvegliato; sotto zero si vende USDE, sopra la banda si ricompra. 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 quattro stati (issue #7) ------------------------------ B = dict(preavviso=0.10, margine=0.20, riacquisto=0.40) def test_stati_scoperto_preavviso_ok_eccedente(): assert W.giudica(1199.0, 4000.0, 1200.0, B)["stato"] == "SCOPERTO" g = W.giudica(1300.0, 4000.0, 1200.0, B) assert g["stato"] == "PREAVVISO" and abs(g["slack"] - 100.0) < 1e-9 and g["preavviso_usd"] == 120.0 assert W.giudica(1500.0, 4000.0, 1200.0, B)["stato"] == "OK" # slack 300: dentro [120, 480] assert W.giudica(1680.0, 4000.0, 1200.0, B)["stato"] == "OK" # slack 480: sul bordo, non compra g = W.giudica(1700.0, 4000.0, 1200.0, B) assert g["stato"] == "ECCEDENTE" and g["riacquisto_usd"] == 480.0 def test_le_bande_vengono_dalla_config_e_sono_ordinate(): b = U.bande_cuscino() assert 0 < b["preavviso"] < b["margine"] < b["riacquisto"] assert W.giudica(1500.0, 4000.0, 1200.0)["stato"] == W.giudica(1500.0, 4000.0, 1200.0, b)["stato"] import pytest with pytest.raises(ValueError): U.bande_cuscino({"cuscino_preavviso_frac": 0.3, "cuscino_margine_frac": 0.2, "cuscino_riacquisto_frac": 0.4}) def test_isteresi_vendita_e_acquisto_convergono_allo_stesso_bersaglio(): """Dopo una vendita da SCOPERTO o un acquisto da ECCEDENTE lo slack e' margine x cuscino: OK, a distanza da entrambe le soglie. Un giro non ne innesca un altro.""" f, _ = U.frazione_cuscino(); b = U.bande_cuscino() E = 4450.0; cusc = E * f q = U.quota_ripristino(b["margine"]) usdc_dopo = E * (1 - q) g = W.giudica(usdc_dopo, E, cusc, b) assert g["stato"] == "OK" assert g["slack"] > g["preavviso_usd"] * 1.5 and g["slack"] < g["riacquisto_usd"] / 1.5 def test_un_bonifico_fa_scattare_il_riacquisto_da_solo(): """Conto a bersaglio (slack = 0,20 x cuscino), bonifico D in USDC: slack +0,7 D. Con D = 12% dell'equity si supera 0,40 x cuscino e si ricompra; con D = 3% no (resta in USDC).""" f, _ = U.frazione_cuscino(); b = U.bande_cuscino() E = 4450.0; cusc = E * f; usdc = cusc * (1 + b["margine"]) for D, atteso in ((0.03 * E, "OK"), (0.12 * E, "ECCEDENTE")): E2 = E + D; cusc2 = E2 * f assert W.giudica(usdc + D, E2, cusc2, b)["stato"] == atteso, D def test_non_vedo_e_blind_non_ok(): assert W.giudica(None, 4000.0, 1200.0, B)["stato"] == "BLIND" assert W.giudica(1500.0, None, None, B)["stato"] == "BLIND" def test_gli_stati_che_convertono_sono_due_e_hanno_un_verso(): assert W.AZIONE == {"SCOPERTO": "vendita", "ECCEDENTE": "acquisto"} # ------------------------------ 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}, "ECCEDENTE") 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.""" b = U.bande_cuscino() assert b["margine"] > b["preavviso"] f, _ = U.frazione_cuscino() E = 4450.0 C = _carica("usde_convert") q = U.quota_ripristino(b["margine"]) + C.TOLL # usde_convert si ferma entro TOLL (importato, P1) usdc_dopo = E * (1 - q) - 1.0 # e vende fino a 1 USDE in meno (passo del venue) assert W.giudica(usdc_dopo, E, E * f, b)["stato"] == "OK" def test_il_sink_telegram_fa_escape_di_minore_e_ampersand(monkeypatch): """Telegram in parse_mode HTML rifiuta un '<' nudo: il primo giro vero (13:53Z del 10/09) ha perso l'allerta PREAVVISO per «(< 10% del cuscino)». L'escape sta nel SINK (`notify`), non nei 45 chiamanti — qui si cattura il corpo della POST e si verifica che il '<' sia diventato <.""" from src.live import notifier as N catturato = {} def finto_urlopen(url, data, timeout=10): import urllib.parse catturato.update(urllib.parse.parse_qs(data.decode())) return None monkeypatch.setattr(N, "_cfg", lambda: ("tok", "chat")) monkeypatch.setattr(N.urllib.request, "urlopen", finto_urlopen) assert N.notify("⚠️ prova ", {"slack": "$+30 (< 10% del cuscino)", "a&b": "c"}) testo = catturato["text"][0] assert "(< 10%" in testo and "<x>" in testo and "a&b" in testo assert testo.startswith("") # il grassetto del titolo resta HTML vero def test_bande_fuori_ordine_danno_BLIND_con_motivo_non_un_crash(): import pytest with pytest.raises(ValueError): U.bande_cuscino({"cuscino_preavviso_frac": 0.5, "cuscino_margine_frac": 0.2, "cuscino_riacquisto_frac": 0.4}) monkey = {"cuscino_preavviso_frac": 0.5, "cuscino_margine_frac": 0.2, "cuscino_riacquisto_frac": 0.4} orig = U.cfg try: U.cfg = lambda: {**orig(), **monkey} g = W.giudica(1500.0, 4000.0, 1200.0) # bande=None -> config rotta finally: U.cfg = orig assert g["stato"] == "BLIND" and "fuori ordine" in g["motivo"] def test_da_eccedente_un_fallimento_si_ritenta_una_volta_al_giorno(): """USDC fermo non e' un rischio: un guasto persistente (tetto del venue, book illeggibile) non deve diventare 4 Telegram al giorno per sempre. Da SCOPERTO il ritmo resta 6h.""" now = 1_789_000_000_000 fallito = [{"ts": now - 8 * 3_600_000, "riconversione": {"tentata": True, "ok": False}}] ok, m = W.puo_riconvertire(fallito, now, True, 1.0, 0.99, stato="ECCEDENTE") assert not ok and "una volta al giorno" in m assert W.puo_riconvertire(fallito, now, True, 1.0, 0.99, stato="SCOPERTO")[0] riuscito = [{"ts": now - 8 * 3_600_000, "riconversione": {"tentata": True, "ok": True}}] assert W.puo_riconvertire(riuscito, now, True, 1.0, 0.99, stato="ECCEDENTE")[0] tardi = [{"ts": now - 25 * 3_600_000, "riconversione": {"tentata": True, "ok": False}}] assert W.puo_riconvertire(tardi, now, True, 1.0, 0.99, stato="ECCEDENTE")[0] def test_i_motivi_delle_guardie_non_hanno_minore_nudo(): """Difesa in profondita' oltre al sink: i motivi finiscono nel campo `azione` delle allerte.""" now = 1_789_000_000_000 casi = [W.puo_riconvertire([], now, False, 1.0, 0.99), W.puo_riconvertire([], now, True, None, 0.99), W.puo_riconvertire([], now, True, 0.98, 0.99), W.puo_riconvertire([{"ts": now - 1000, "riconversione": {"tentata": True, "ok": True}}], now, True, 1.0, 0.99)] assert all("<" not in m for _, m in casi) 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)