332 lines
15 KiB
Python
332 lines
15 KiB
Python
"""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 _finto_lettore_quota(valori):
|
|
"""Ritorna i valori dati in ordine, uno per chiamata (per monkeypatch di _leggi_quota_usde)."""
|
|
it = iter(valori)
|
|
return lambda: next(it)
|
|
|
|
|
|
def test_riconverti_lancia_lo_script_vero_con_esegui(monkeypatch):
|
|
# rc==0: "ok" a prescindere dalla quota (lo script vero ha gia' le sue guardie, P15) — la
|
|
# quota qui non si muove, per mostrare che il caso "ok" non dipende dalla rilettura.
|
|
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)
|
|
monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.67, 0.67]))
|
|
e = W.riconverti(0.67)
|
|
assert e["ok"] and e["esito"] == "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")
|
|
assert e["quota_prima"] == 0.67 and e["quota_dopo"] == 0.67
|
|
|
|
|
|
def test_riconverti_rc_diverso_da_zero_con_quota_mossa_e_parziale(monkeypatch):
|
|
# #35 B10: rc!=0 (es. ordine parzialmente fillato poi rifiutato) ma la quota si e' mossa
|
|
# per davvero -> "parziale", non "fallita": la vendita/acquisto e' avvenuto, solo non finito.
|
|
class R:
|
|
returncode, stdout, stderr = 1, "", "book illeggibile a meta' piano\n"
|
|
|
|
monkeypatch.setattr(W.subprocess, "run", lambda cmd, **kw: R())
|
|
monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.75, 0.70]))
|
|
e = W.riconverti(0.64)
|
|
assert e["esito"] == "parziale" and e["ok"] is False
|
|
assert e["quota_prima"] == 0.75 and e["quota_dopo"] == 0.70
|
|
|
|
|
|
def test_riconverti_rc_diverso_da_zero_con_quota_ferma_e_fallita(monkeypatch):
|
|
class R:
|
|
returncode, stdout, stderr = 1, "", "prezzo fuori banda: niente ordini\n"
|
|
|
|
monkeypatch.setattr(W.subprocess, "run", lambda cmd, **kw: R())
|
|
monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.75, 0.751]))
|
|
e = W.riconverti(0.64)
|
|
assert e["esito"] == "fallita" and e["ok"] is False
|
|
|
|
|
|
def test_riconverti_timeout_con_quota_mossa_e_parziale(monkeypatch):
|
|
# #35 B10: il subprocess viene ucciso a TIMEOUT_CONVERT_S ma alcuni ordini erano gia' fillati
|
|
# prima del kill -> oggi (ok=rc==0) diventava "FALLITA", corretto e' "parziale".
|
|
def finto_run(cmd, **kw):
|
|
raise W.subprocess.TimeoutExpired(cmd, kw.get("timeout"))
|
|
|
|
monkeypatch.setattr(W.subprocess, "run", finto_run)
|
|
monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.80, 0.72]))
|
|
e = W.riconverti(0.64)
|
|
assert e["rc"] is None and e["esito"] == "parziale" and e["ok"] is False
|
|
assert e["coda"][0].startswith("timeout dopo")
|
|
|
|
|
|
def test_riconverti_rilettura_dopo_fallita_e_fallita_esplicita(monkeypatch):
|
|
# se non si riesce a rileggere la quota dopo, non si puo' dire "parziale": si dichiara
|
|
# l'incertezza invece di dedurre un movimento zero che non si e' verificato.
|
|
class R:
|
|
returncode, stdout, stderr = 1, "", "errore\n"
|
|
|
|
monkeypatch.setattr(W.subprocess, "run", lambda cmd, **kw: R())
|
|
monkeypatch.setattr(W, "_leggi_quota_usde", _finto_lettore_quota([0.75, None]))
|
|
e = W.riconverti(0.64)
|
|
assert e["esito"] == "fallita (quota dopo non leggibile)" and e["ok"] is False
|
|
assert e["quota_dopo"] is None
|
|
|
|
|
|
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 <x>", {"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("<b>") # 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)
|