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 <noreply@anthropic.com>
This commit is contained in:
@@ -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."""
|
||||
|
||||
+12
-14
@@ -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")
|
||||
|
||||
+6
-6
@@ -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]:
|
||||
|
||||
@@ -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"]
|
||||
Reference in New Issue
Block a user