From 52a905d497b47dfc5c8b84ddcc27281a15ec9445 Mon Sep 17 00:00:00 2001 From: "Adriano (goldrake)" Date: Sun, 27 Sep 2026 15:17:01 +0000 Subject: [PATCH] book: leva_tetto unica per cuscino USDE e giornale (#28) La leva lorda massima del libro era ridichiarata con tre formule diverse: usde.frazione_cuscino (N_ASSET fisso, senza scala), la regola `leva` del giornale (senza scala) e journal.movimenti_capitale (con scala). A scala 1,25 il cuscino USDC risultava 0,30 dell'equity invece di 0,375. Nuova src/live/book.leva_tetto(cfg) = len(ASSETS) x min(WEIGHT, frac) x scala, al piu' LEVA_LORDA_MAX; su ScalaNonAutorizzata non solleva, ritorna il tetto di codice (cuscino_watch non deve morire sulla config sbagliata). usde.frazione_cuscino e la regola `leva` del giornale ora la derivano; movimenti_capitale idem, nel suo try/except esistente. Co-Authored-By: Claude Sonnet 5 --- src/live/book.py | 36 +++++++++++ src/live/journal.py | 26 ++++---- src/live/usde.py | 12 ++-- tests/test_leva_tetto.py | 128 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 182 insertions(+), 20 deletions(-) create mode 100644 tests/test_leva_tetto.py diff --git a/src/live/book.py b/src/live/book.py index a0a02a5..05d8fa4 100644 --- a/src/live/book.py +++ b/src/live/book.py @@ -136,6 +136,42 @@ def _scala(real_equity: float | None = None, eq_fallback: str | None = None, return s +def leva_tetto(cfg: dict | None = None) -> tuple[float, str]: + """Leva lorda massima che il libro PUO' prendere. UNA formula, non tre (#28): prima era + ridichiarata in `usde.frazione_cuscino` (N_ASSET fisso, senza scala) e nella regola `leva` del + giornale (senza scala) -> a scala 1,25 il cuscino USDC risultava 0,30 dell'equity invece di + 0,375, e la regola `leva` mostrava un tetto diverso da quello che il libro rispetta davvero. + + leva = len(ASSETS) x min(WEIGHT, frac) x scala, al piu' LEVA_LORDA_MAX. -> (leva, come). + - frac = cfg["max_notional_per_asset_frac"] se presente, altrimenti WEIGHT. Il min(WEIGHT, frac) + viene dal fatto che `book_net_target` non supera mai weight*equity*(W_TP01+W_SKH) = + weight*equity (i due pesi sommano a 1): |raw| <= weight*equity, quindi |net_target| <= + scala*min(cap, weight*equity) -- un frac oltre WEIGHT alza il cap nominale ma non la leva + REALE che il libro puo' esprimere. + - scala = la scala DICHIARATA (`_scala(real_equity=1.0, cfg=cfg)`, gia' validata): un tetto + descrive quanto la config AUTORIZZA, non quanto un giro con equity illeggibile ha preso. + - Su `ScalaNonAutorizzata` (config fuori scaletta o fuori tetto) NON si solleva: un + sorvegliante (`cuscino_watch`) non deve morire proprio quando la config e' sbagliata + (verifica fable B5) -> si ritorna il tetto di CODICE, dichiarando l'anomalia in `come`. + """ + if cfg is None: + cfg = json.loads(CONFIG.read_text()) if CONFIG.exists() else {} + try: + scala = _scala(real_equity=1.0, cfg=cfg) + except ScalaNonAutorizzata as e: + breve = str(e).splitlines()[0] + breve = breve if len(breve) <= 80 else breve[:77] + "..." + return LEVA_LORDA_MAX, (f"config fuori scaletta/tetto ({breve}): " + f"tetto di codice LEVA_LORDA_MAX") + raw_frac = cfg.get("max_notional_per_asset_frac") + frac = float(raw_frac) if raw_frac is not None else WEIGHT + frac_eff = min(WEIGHT, frac) + leva = len(ASSETS) * frac_eff * scala + base = f"{len(ASSETS)} asset x frac {frac_eff} x scala {scala} = {leva:.2f}x" + if leva > LEVA_LORDA_MAX + 1e-12: + return LEVA_LORDA_MAX, f"{base}, troncato al tetto di codice {LEVA_LORDA_MAX}x" + return leva, base + def _read_watermark() -> float | None: """Ultima equity REALE osservata sul conto. None se mai vista o file illeggibile.""" diff --git a/src/live/journal.py b/src/live/journal.py index d500cd4..7221f1d 100644 --- a/src/live/journal.py +++ b/src/live/journal.py @@ -192,12 +192,10 @@ def movimenti_capitale(con, fino_ts: str, rows=None) -> dict: return dict(eventi=[], certi=0.0, ambigui=0.0, soglia=soglia, avvisi=avvisi) try: cfg = json.loads((PROJECT_ROOT / "config" / "live.json").read_text()) - # la leva lorda massima che il libro puo' tenere: frac x n_asset x scala, mai oltre il - # tetto di CODICE (P1: derivata da `book`, non ridichiarata — la chiave di scala del - # 01/09 entra qui da sola) - scala = float(cfg.get("book_scale_k") or 1.0) - leva_tetto = min(float(cfg["max_notional_per_asset_frac"]) * len(ASSETS) * scala, - float(_book.LEVA_LORDA_MAX)) + # la leva lorda massima che il libro puo' tenere: UNA sola formula, in `book.leva_tetto` + # (P1: derivata da `book`, non ridichiarata — #28 ne aveva tre diverse fra qui, + # usde.frazione_cuscino e la regola `leva` sotto, e non tornavano) + leva_tetto = _book.leva_tetto(cfg)[0] except Exception: leva_tetto = float(_book.LEVA_LORDA_MAX) # config illeggibile: il tetto di codice px = {a: _px_1h(a) for a in ASSETS} if candidati else {} @@ -567,17 +565,17 @@ def lettura(con, voce: dict) -> list[dict]: # --- quanto capitale e' esposto ---------------------------------------------------------- lv = lb.get("leva_lorda") if lv is not None: - # Il tetto si LEGGE da config/live.json. Ridichiararlo qui e' lo schema che questo - # progetto ha gia' pagato quattro volte (fee_watch, test di guardia sul cap, taratura - # venue_watch, indice del namespace): un sorvegliante che ridichiara il proprio - # bersaglio continua a passare il giorno che il bersaglio cambia. + # Il tetto si LEGGE da `book.leva_tetto`, non ridichiarato qui. Ridichiararlo e' lo schema + # che questo progetto ha gia' pagato quattro volte (fee_watch, test di guardia sul cap, + # taratura venue_watch, indice del namespace) — e una quinta, con una formula diversa da + # quella di `movimenti_capitale` e di `usde.frazione_cuscino`, e' quella corretta da #28. + from src.live import book as _book # lazy: prende il tetto vivo, anche monkeypatchato try: cfg = json.loads((PROJECT_ROOT / "config" / "live.json").read_text()) - cap_frac = float(cfg["max_notional_per_asset_frac"]) - tetto = cap_frac * len(lb["asset"]) + tetto, come = _book.leva_tetto(cfg) dire("leva", f"Leva lorda **{lv:.2f}x** su un tetto di {tetto:.2f}x " - f"(`max_notional_per_asset_frac` {cap_frac:.2f} x {len(lb['asset'])} " - f"asset, letto da `config/live.json`): margine {tetto - lv:.2f}x.") + f"({come}, da `config/live.json` e `book.leva_tetto`): " + f"margine {tetto - lv:.2f}x.") except Exception as e: dire("leva", f"Leva lorda **{lv:.2f}x**; tetto non verificabile " f"({type(e).__name__} su config/live.json).", "attenzione") diff --git a/src/live/usde.py b/src/live/usde.py index c1fdd8d..6cdf66c 100644 --- a/src/live/usde.py +++ b/src/live/usde.py @@ -104,12 +104,12 @@ def valuta(eq_usde: float, px: float | None) -> tuple[float, str]: # il saldo USDC va negativo alla prima perdita del libro e Deribit lo finanzia allo 0,05%/GIORNO # (KB "Cross collateral specifications": 18,25%/anno, 4,3x la resa USDE). Il cuscino richiesto e' # il disaster-SL sulla MASSIMA esposizione lorda che il libro puo' prendere: -# n_asset x frac x disaster_sl_pct x equity_totale +# book.leva_tetto() x disaster_sl_pct x equity_totale # tutto letto da config/live.json e src/live/book.py, MAI ridichiarato (P1). Prima del 10/09 la # formula viveva in scripts/live/usde_convert.py; il sorvegliante cuscino_watch la deriva da qui, -# e usde_convert pure — due lettori, una formula. - -N_ASSET = 2 +# e usde_convert pure — due lettori, una formula. ⚠️ Fino al #28 la leva lorda era RIDICHIARATA +# qui (N_ASSET fisso x WEIGHT, senza scala): a scala 1,25 dava 0,30 invece di 0,375. Ora la stessa +# `book.leva_tetto` che usa anche il giornale (src/live/journal.py). def frazione_cuscino() -> tuple[float, str]: @@ -117,8 +117,8 @@ def frazione_cuscino() -> tuple[float, str]: cfg = json.loads(CFG_PATH.read_text()) sl = float(cfg["disaster_sl_pct"]) from src.live import book as B # lazy: book importa shadow, che importa questo modulo - frac = float(B.WEIGHT) - return N_ASSET * frac * sl, f"{N_ASSET} asset x frac {frac} x SL {sl:.0%}" + leva, come_leva = B.leva_tetto(cfg) + return leva * sl, f"{come_leva} x SL {sl:.0%}" def cuscino_richiesto_usd(equity_tot: float) -> tuple[float, str]: diff --git a/tests/test_leva_tetto.py b/tests/test_leva_tetto.py new file mode 100644 index 0000000..ecaaebe --- /dev/null +++ b/tests/test_leva_tetto.py @@ -0,0 +1,128 @@ +"""book.leva_tetto — la leva lorda massima del libro, in UNA sola formula (#28). + +Prima della correzione la leva lorda tornava con tre formule diverse: `usde.frazione_cuscino` +(N_ASSET fisso, senza scala), la regola `leva` del giornale (senza scala) e +`journal.movimenti_capitale` (con scala, ma ridichiarata). A scala 1,25 il cuscino USDC +risultava 0,30 dell'equity invece di 0,375 — il caso che T2 riproduce. +""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from src.live import book as B # noqa: E402 +from src.live import journal as J # noqa: E402 +from src.live import tradesdb as T # noqa: E402 +from src.live import usde as U # noqa: E402 + + +def _cfg(**kw) -> dict: + base = dict(max_notional_per_asset_frac=0.5, disaster_sl_pct=0.3) + base.update(kw) + return base + + +# ---------------------------------------------------------------- T1: scala assente + +def test_T1_scala_assente_e_frac_uguale_al_weight(): + leva, come = B.leva_tetto(_cfg()) + assert leva == pytest.approx(len(B.ASSETS) * B.WEIGHT) + assert "scala 1.0" in come + + +# ---------------------------------------------------------------- T2: il caso dell'issue + +def test_T2_scala_1_25_alza_la_leva_e_il_cuscino_usde(tmp_path, monkeypatch): + """A scala 1,25 il cuscino USDC deve valere 0,375 dell'equity, non 0,30 (#28).""" + cfg = _cfg(book_scale_k=1.25) + leva, _ = B.leva_tetto(cfg) + assert leva == pytest.approx(1.25) + + # frazione_cuscino legge da CFG_PATH: si monkeypatcha il file, come fanno gli altri test + # su usde/_scala (grep "CFG_PATH\|CONFIG" tests/). + cpath = tmp_path / "live.json" + cpath.write_text(json.dumps(cfg)) + monkeypatch.setattr(U, "CFG_PATH", cpath) + f, come = U.frazione_cuscino() + assert f == pytest.approx(0.375) + assert "SL 30%" in come + + +# ---------------------------------------------------------------- T3: fuori scaletta non solleva + +def test_T3_scala_fuori_scaletta_non_solleva_e_torna_il_tetto_di_codice(): + """cuscino_watch non deve morire proprio quando la config e' sbagliata (verifica fable B5).""" + leva, come = B.leva_tetto(_cfg(book_scale_k=1.1)) + assert leva == pytest.approx(B.LEVA_LORDA_MAX) + assert "tetto di codice" in come + + +# ---------------------------------------------------------------- T4: frac assente + +def test_T4_frac_assente_usa_weight(): + cfg = dict(disaster_sl_pct=0.3) # niente max_notional_per_asset_frac + leva, come = B.leva_tetto(cfg) + assert leva == pytest.approx(len(B.ASSETS) * B.WEIGHT) + assert f"frac {B.WEIGHT}" in come + + +# ---------------------------------------------------------------- T5: il tronco al tetto di codice + +def test_T5_leva_oltre_il_tetto_di_codice_viene_troncata(monkeypatch): + """Verifica diretta del troncamento. Con WEIGHT alzato il prodotto supera LEVA_LORDA_MAX pur + restando `_scala` silenziosa (frac ASSENTE dalla config -> il suo controllo sul prodotto non + scatta, perche' guarda solo `max_notional_per_asset_frac` quando e' dichiarato).""" + monkeypatch.setattr(B, "WEIGHT", 0.9) + leva, come = B.leva_tetto(dict(book_scale_k=1.25)) # frac assente -> usa WEIGHT = 0.9 + assert leva == pytest.approx(B.LEVA_LORDA_MAX) + assert "troncato" in come + + +# ---------------------------------------------------------------- T6: il giornale usa la stessa leva_tetto + +@pytest.fixture() +def db(tmp_path): + con = T.connect(tmp_path / "t.db") + yield con + con.close() + + +def _voce(mercato=None, libro=None, pnl=None, salute=None, giorno="2026-08-21"): + return dict(giorno=giorno, ts_scritto=T.ora(), mercato=mercato or {}, + libro=libro or {"errore": "nessun giro", "giri": 0}, pnl=pnl or {}, + salute=salute or dict(giri_book=24, feed_skh_min=0, atteso_giri=24, + giorno_in_corso=False, giri_mancanti=0)) + + +def _libro(**asset): + a = {k: dict(dict(tp_frac=0.0, skh_sign=0, skh_entry=None, target=0.0, posizione=0.0, + azione="HOLD (a target)"), **v) for k, v in asset.items()} + lordo = sum(abs(v["posizione"]) for v in a.values()) + return dict(giri=24, ultimo_giro="2026-08-21T23:07:01+00:00", equity=600.0, barra="2026-08-21", + feed_skh_min=0, nozionale_lordo=lordo, leva_lorda=lordo / 600.0, asset=a) + + +def _pnl(**kw): + base = dict(equity_inizio=600.0, equity_fine=600.0, delta_equity=0.0, realizzato_lordo=0.0, + realizzato_netto=0.0, roundtrip_chiusi=0, fill=0, fee=0.0, + cumulato_da_arming=0.0, letture_equity=24) + return {**base, **kw} + + +def test_T6_regola_leva_margine_positivo_a_scala_1_25(db, monkeypatch, tmp_path): + """lv=1.2 su un tetto di 1.25x (frac 0.5, scala 1.25): margine positivo, non negativo.""" + (tmp_path / "config").mkdir() + (tmp_path / "config" / "live.json").write_text( + json.dumps(dict(max_notional_per_asset_frac=0.5, book_scale_k=1.25))) + monkeypatch.setattr(J, "PROJECT_ROOT", tmp_path) + lb = _libro(BTC=dict(posizione=360.0), ETH=dict(posizione=360.0)) # 720/600 = 1.2x + assert lb["leva_lorda"] == pytest.approx(1.2) + r = J.lettura(db, _voce(libro=lb, pnl=_pnl())) + riga = next(x for x in r if x["id"] == "leva") + assert "1.25x" in riga["testo"] and "margine 0.05x" in riga["testo"]