Adriano Dal Pastro
e69cc5bd81
I due confronti `abs(net_target - formula) < 1e-6` erano insoddisfacibili per costruzione: `book_report` PUBBLICA valori arrotondati (`net_target` a 2 decimali, `tp_frac` a 4) mentre l'ordine viene costruito sul valore non arrotondato (`build_book_order(inst, net, ...)`), quindi un test che ricalcola la formula dal `tp_frac` pubblicato eredita DUE arrotondamenti. Nessun ordine e' mai stato sbagliato: lo scarto e' 0.5-1.5 centesimi su target da $114 e $954, con min_order a $5. ⚠️ Il difetto vero non e' la tolleranza, e' la POTENZA: con `tp_frac=0` e `skh_sign=0` il target e' esattamente `0.0`, i due arrotondamenti sono esatti e l'invariante passa senza essere mai esercitata. Dal 2026-06-23 al 2026-08-18 il book e' stato flat quasi ininterrottamente -> per due mesi questi test non hanno verificato nulla su una formula di produzione, e il difetto e' emerso solo quando TP01 e SKH01 sono andati long insieme. Stessa lezione del 26/07 (test_skh_partial_entry): un self-check su eventi rari si campiona sugli EVENTI, non sulla popolazione. - `_budget_arrotondamento(equity)`: tolleranza DERIVATA (0.005 del round del net + WEIGHT*equity*W_TP01*0.00005 del round del tp_frac propagato), non tarata sul risultato. - `test_la_formula_del_report_ha_potenza_anche_a_libro_flat`: segnale FORZATO su 4 frazioni scomode (long+SKH long, long+SKH short, flat+SKH short, cap) con contatore che verifica che tutti i casi diano target != 0 -> la copertura non dipende dal mercato. - `test_il_budget_di_arrotondamento_non_copre_un_errore_di_formula`: controllo POSITIVO, i modi reali di rompere la formula (pesi scambiati, cap non applicato, segno invertito) devono stare oltre 100x il budget. Una tolleranza che assolve tutto non e' una tolleranza. Verificato per MUTAZIONE, non a occhio: `round(net, 0)` in book.py -> falliscono tutti e tre (incluso il nuovo, che e' il punto); `W_TP01 <-> W_SKH` -> falliscono test_net_target_sizing e il controllo positivo. src/live/book.py ripristinato bit-identico, produzione NON toccata. 38/38 in tests/test_book_live.py; suite 619 passati, 1 fallito (il noto test_gtaa_band_gate, altro asse, invariato). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
575 lines
29 KiB
Python
575 lines
29 KiB
Python
"""Test del BOOK DERIBIT-ONLY live (TP01+SKH01 nettati in software, un solo conto).
|
|
|
|
Coprono: la formula di netting (sizing 75/25 + cap, long/short/flat/flip), la PARITA' coi pesi del
|
|
backtest (deribit_book_sleeves), la sicurezza del gate (disarmato -> nessun ordine), e il reconcile
|
|
CON SEGNO (close+open sui flip, reduce reduce_only) — senza toccare la rete (trader fittizio).
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
# Data FRESCA per i report finti: dal 2026-07-25 book_execute ha uno staleness-gate bloccante
|
|
# (feed piu' vecchio di 2 giorni -> non esegue). Questi test riguardano skh_error / pos_error /
|
|
# eq_fallback, NON la staleness: con una data fissa marcirebbero appena supera la soglia.
|
|
# Lo staleness-gate ha i suoi test dedicati in tests/test_book_staleness_gate.py.
|
|
def _fresh_bar() -> str:
|
|
import pandas as _pd
|
|
return str(_pd.Timestamp.now(tz="UTC").normalize().date())
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from src.live.book import W_SKH, W_TP01, book_net_target, build_book_order
|
|
from src.live.execution import Fill
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# BUDGET DI ARROTONDAMENTO del report (2026-08-21).
|
|
# ---------------------------------------------------------------------------
|
|
def _budget_arrotondamento(equity: float) -> float:
|
|
"""Tolleranza per confrontare il `net_target` PUBBLICATO con la formula pura.
|
|
|
|
`book_report` pubblica valori arrotondati — `net_target` a 2 decimali e `tp_frac` a 4 —
|
|
mentre l'ordine viene costruito sul valore NON arrotondato (`build_book_order(inst, net, ...)`).
|
|
Un test che ricalcola la formula dal `tp_frac` pubblicato eredita quindi DUE arrotondamenti e
|
|
non puo' confrontare a 1e-6:
|
|
|
|
- `round(net, 2)` -> fino a 0.005 di scarto;
|
|
- `round(tp_frac, 4)` -> fino a 0.00005 di scarto sulla frazione, che la formula
|
|
amplifica per WEIGHT * equity * W_TP01.
|
|
|
|
⚠️ Questo NON e' un allentamento del test: i modi in cui la formula puo' rompersi davvero
|
|
(peso sbagliato, cap non applicato, segno invertito, W_TP01/W_SKH scambiati) valgono decine
|
|
di dollari su target da $114, tre ordini di grandezza sopra questo budget.
|
|
"""
|
|
from src.live.book import WEIGHT
|
|
return 0.005 + WEIGHT * equity * W_TP01 * 0.00005
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Formula di netting: 75/25, cap, combinazioni long/short/flat.
|
|
# ---------------------------------------------------------------------------
|
|
def test_net_target_sizing():
|
|
E, cap = 600.0, 300.0
|
|
assert book_net_target(0.0, 0, E, cap) == 0.0 # tutto flat
|
|
assert book_net_target(0.0, 1, E, cap) == 75.0 # solo SKH long = 0.25*0.5*600
|
|
assert book_net_target(0.0, -1, E, cap) == -75.0 # solo SKH short
|
|
assert book_net_target(1.0, 0, E, cap) == 225.0 # solo TP01 pieno = 0.75*0.5*600
|
|
assert book_net_target(1.0, -1, E, cap) == 150.0 # TP long + SKH short (hedge parziale)
|
|
assert book_net_target(1.0, 1, E, cap) == 300.0 # capped
|
|
assert book_net_target(2.0, 1, E, cap) == 300.0 # cap superiore
|
|
assert book_net_target(2.0, -1, E, cap) == 300.0 # 0.5*600*(1.5-0.25)=375 -> cap 300
|
|
|
|
|
|
def test_net_target_clamps_negative():
|
|
# TP flat e SKH short forte non sfora il cap negativo
|
|
assert book_net_target(0.0, -1, 4000.0, 300.0) == -300.0
|
|
# tp_frac negativo trattato come 0 (TP01 e' long-flat)
|
|
assert book_net_target(-5.0, 1, 600.0, 300.0) == 75.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CAP DINAMICO (frontiera 2026-07-03): con max_notional_per_asset_frac in config il
|
|
# cap = equity*frac (cresce col capitale). GUARDRAIL: dinamico SOLO con equity reale
|
|
# fidata; su fallback/offline/no-frac -> cap FISSO (protezione downside).
|
|
# ---------------------------------------------------------------------------
|
|
import json as _json
|
|
|
|
|
|
def _write_cfg(monkeypatch, tmp_path, watermark: float | None = None, **keys):
|
|
"""⚠️ Isola SIA la config SIA il watermark dell'equity. Senza la seconda riga di
|
|
monkeypatch questi test leggono `data/live/equity_seen.json` REALE e il loro esito dipende
|
|
da quanto c'e' sul conto — difetto trovato il 2026-07-26, quando l'aggiunta del watermark
|
|
fece fallire un test che non lo menzionava nemmeno."""
|
|
import src.live.book as book
|
|
p = tmp_path / "live.json"
|
|
p.write_text(_json.dumps(keys))
|
|
monkeypatch.setattr(book, "CONFIG", p)
|
|
wm = tmp_path / "equity_seen.json"
|
|
monkeypatch.setattr(book, "EQUITY_WATERMARK", wm)
|
|
if watermark is not None:
|
|
wm.write_text(_json.dumps({"real_equity": float(watermark)}))
|
|
return book
|
|
|
|
|
|
def test_cap_dynamic_scales_with_trusted_equity(monkeypatch, tmp_path):
|
|
book = _write_cfg(monkeypatch, tmp_path, max_notional_per_asset_usd=300, max_notional_per_asset_frac=0.5)
|
|
# equity reale fidata -> cap = equity/2 (cresce col capitale)
|
|
assert book._cap(equity=5000.0, real_equity=5000.0, eq_fallback=None) == 2500.0
|
|
assert book._cap(equity=598.0, real_equity=598.0, eq_fallback=None) == 299.0 # ~inerte a $600
|
|
|
|
|
|
def test_cap_falls_back_to_fixed_on_eq_fallback(monkeypatch, tmp_path):
|
|
"""Senza watermark (conto mai visto) il fallback e' la taglia sicura, MAI equity/2 del paper
|
|
($1.000 su un conto ignoto = pericoloso). Comportamento storico, invariato."""
|
|
book = _write_cfg(monkeypatch, tmp_path, max_notional_per_asset_usd=300, max_notional_per_asset_frac=0.5)
|
|
assert book._cap(equity=2000.0, real_equity=None, eq_fallback="paper_cap") == 300.0
|
|
assert book._cap(equity=2000.0, real_equity=None, eq_fallback=None) == 300.0 # real None -> fisso
|
|
assert book._cap(equity=0.0, real_equity=0.0, eq_fallback=None) == 300.0 # equity 0 -> fisso
|
|
|
|
|
|
def test_cap_di_fallback_non_supera_il_conto_osservato(monkeypatch, tmp_path):
|
|
"""2026-07-26: col cap di config alzato a $3.000 in previsione di un deposito, il fallback
|
|
deve restare legato all'ULTIMA EQUITY VISTA, altrimenti su un conto da $597 permetterebbe
|
|
$2.000 di nozionale lordo = 3.35x di leva proprio quando l'equity non e' leggibile."""
|
|
book = _write_cfg(monkeypatch, tmp_path, watermark=596.92,
|
|
max_notional_per_asset_usd=3000, max_notional_per_asset_frac=0.5)
|
|
cap = book._cap(equity=2000.0, real_equity=None, eq_fallback="paper_cap")
|
|
assert cap == 596.92 * 0.5
|
|
assert 2 * cap <= 596.92, "leva > 1x in fallback"
|
|
|
|
|
|
def test_cap_di_fallback_cresce_col_conto_dopo_un_deposito(monkeypatch, tmp_path):
|
|
book = _write_cfg(monkeypatch, tmp_path, watermark=6047.0,
|
|
max_notional_per_asset_usd=3000, max_notional_per_asset_frac=0.5)
|
|
assert book._cap(equity=2000.0, real_equity=None, eq_fallback="paper_cap") == 3000.0
|
|
|
|
|
|
def test_cap_fixed_when_no_frac_in_config(monkeypatch, tmp_path):
|
|
book = _write_cfg(monkeypatch, tmp_path, max_notional_per_asset_usd=300) # niente frac
|
|
assert book._cap(equity=5000.0, real_equity=5000.0, eq_fallback=None) == 300.0
|
|
|
|
|
|
def test_book_report_dynamic_cap_online(monkeypatch, tmp_path):
|
|
"""Integrazione: online con equity reale -> cap_per_asset = equity/2 e la formula lo rispetta."""
|
|
book = _write_cfg(monkeypatch, tmp_path, max_notional_per_asset_usd=300, max_notional_per_asset_frac=0.5)
|
|
base = book.shadow_report(offline=True, equity_override=5000.0)
|
|
monkeypatch.setattr(book, "shadow_report",
|
|
lambda **k: {**base, "real_equity": 5000.0, "equity": 5000.0, "eq_fallback": None})
|
|
r = book.book_report()
|
|
assert r["cap_per_asset"] == 2500.0
|
|
for a in r["assets"]:
|
|
expect = book_net_target(a["tp_frac"], a["skh_sign"], 5000.0, 2500.0)
|
|
assert abs(a["net_target"] - expect) < _budget_arrotondamento(5000.0)
|
|
|
|
|
|
def test_weights_match_backtest_sleeves():
|
|
"""I pesi del book live DEVONO coincidere con quelli del backtest (deribit_book_sleeves)."""
|
|
from src.portfolio.sleeves import deribit_book_sleeves
|
|
w = {s.name.split("_")[0]: s.weight for s in deribit_book_sleeves()}
|
|
assert abs(w["TP01"] - W_TP01) < 1e-12 and abs(w["SKH01"] - W_SKH) < 1e-12
|
|
assert abs((W_TP01 + W_SKH) - 1.0) < 1e-12
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Costruzione ordine: side, reduce_only, flip, close, soglia minima.
|
|
# ---------------------------------------------------------------------------
|
|
def test_build_order_open_long():
|
|
o = build_book_order("BTC_USDC-PERPETUAL", 75.0, 0.0, 60000.0, min_usd=5.0)
|
|
assert o["side"] == "buy" and not o["reduce_only"] and not o["needs_flip"] and not o["is_close"]
|
|
|
|
|
|
def test_build_order_reduce_same_sign():
|
|
o = build_book_order("BTC_USDC-PERPETUAL", 75.0, 150.0, 60000.0)
|
|
assert o["side"] == "sell" and o["reduce_only"] and not o["needs_flip"]
|
|
|
|
|
|
def test_build_order_flip():
|
|
o = build_book_order("BTC_USDC-PERPETUAL", -75.0, 75.0, 60000.0)
|
|
assert o["needs_flip"] and o["side"] == "sell"
|
|
|
|
|
|
def test_build_order_close_to_flat():
|
|
o = build_book_order("BTC_USDC-PERPETUAL", 0.0, 75.0, 60000.0)
|
|
assert o["is_close"] and o["reduce_only"] and o["side"] == "sell"
|
|
|
|
|
|
def test_build_order_below_min_is_none():
|
|
assert build_book_order("BTC_USDC-PERPETUAL", 75.0, 73.0, 60000.0, min_usd=5.0) is None
|
|
assert build_book_order("BTC_USDC-PERPETUAL", 0.0, 0.0, 60000.0) is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sicurezza del GATE: disarmato (execution_enabled=false) -> do_execute False.
|
|
# ---------------------------------------------------------------------------
|
|
def test_gate_requires_both_switches():
|
|
def do_execute(enabled, want_execute):
|
|
return bool(want_execute) and bool(enabled)
|
|
assert not do_execute(False, False)
|
|
assert not do_execute(True, False) # armato ma senza --execute
|
|
assert not do_execute(False, True) # --execute ma disarmato
|
|
assert do_execute(True, True) # solo con entrambi
|
|
|
|
|
|
def test_config_default_disarmed(tmp_path, monkeypatch):
|
|
"""load_config di book_execute mette execution_enabled=False di default (fail-safe)."""
|
|
import importlib
|
|
be = importlib.import_module("scripts.live.book_execute") if False else None
|
|
# carica il modulo via path (scripts/ non e' un package importabile per nome)
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("book_execute", PROJECT_ROOT / "scripts/live/book_execute.py")
|
|
mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)
|
|
monkeypatch.setattr(mod, "CONFIG", tmp_path / "nope.json") # config assente
|
|
assert mod.load_config()["execution_enabled"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Reconcile CON SEGNO (long/short/flip) senza rete: trader fittizio.
|
|
# ---------------------------------------------------------------------------
|
|
class FakeTrader:
|
|
"""Replica la logica di DeribitTrader.rebalance_signed registrando le chiamate, senza rete."""
|
|
def __init__(self, pos):
|
|
self.pos = float(pos)
|
|
self.calls = []
|
|
|
|
def position_usd(self, instrument):
|
|
return self.pos
|
|
|
|
def _mk(self, side, amount, reduce_only, label):
|
|
self.calls.append((label, side, round(amount, 6), reduce_only))
|
|
return Fill(instrument="X", side=side, amount=amount, filled=amount, price=60000.0,
|
|
fee_usdc=0.0, order_id="1", state="filled", verified=True)
|
|
|
|
def close(self, instrument, label="x"):
|
|
if abs(self.pos) < 1.0:
|
|
return None
|
|
f = self._mk("sell" if self.pos > 0 else "buy", abs(self.pos) / 60000.0, True, label)
|
|
self.pos = 0.0
|
|
return f
|
|
|
|
def open(self, instrument, side, amount, label="x"):
|
|
f = self._mk(side, amount, False, label)
|
|
self.pos += (amount * 60000.0) * (1 if side == "buy" else -1)
|
|
return f
|
|
|
|
def _submit(self, instrument, side, amount, *, reduce_only, label, **k):
|
|
f = self._mk(side, amount, reduce_only, label)
|
|
self.pos += (amount * 60000.0) * (1 if side == "buy" else -1)
|
|
return f
|
|
|
|
# importa il metodo reale da DeribitTrader (testiamo proprio quella logica)
|
|
from src.live.execution import DeribitTrader as _DT
|
|
rebalance_signed = _DT.rebalance_signed
|
|
|
|
|
|
def test_reconcile_open_from_flat():
|
|
t = FakeTrader(0.0)
|
|
t.rebalance_signed("BTC_USDC-PERPETUAL", 75.0, 60000.0, min_usd=5.0)
|
|
labels = [c[0] for c in t.calls]
|
|
assert labels == ["book-open"] and t.calls[0][1] == "buy"
|
|
|
|
|
|
def test_reconcile_flip_closes_then_opens():
|
|
t = FakeTrader(75.0) # long, target short -> flip
|
|
t.rebalance_signed("BTC_USDC-PERPETUAL", -75.0, 60000.0, min_usd=5.0)
|
|
labels = [c[0] for c in t.calls]
|
|
assert labels == ["book-flip-close", "book-open"]
|
|
assert t.calls[1][1] == "sell" # apre short dopo il close
|
|
|
|
|
|
def test_reconcile_reduce_same_sign_is_reduce_only():
|
|
t = FakeTrader(150.0)
|
|
t.rebalance_signed("BTC_USDC-PERPETUAL", 75.0, 60000.0, min_usd=5.0)
|
|
assert [c[0] for c in t.calls] == ["book-reduce"]
|
|
assert t.calls[0][3] is True # reduce_only
|
|
|
|
|
|
def test_reconcile_target_flat_closes():
|
|
t = FakeTrader(75.0)
|
|
t.rebalance_signed("BTC_USDC-PERPETUAL", 0.0, 60000.0, min_usd=5.0)
|
|
assert [c[0] for c in t.calls] == ["book-exit"]
|
|
|
|
|
|
def test_reconcile_below_min_noop():
|
|
t = FakeTrader(73.0)
|
|
t.rebalance_signed("BTC_USDC-PERPETUAL", 75.0, 60000.0, min_usd=5.0)
|
|
assert t.calls == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PARITA' d'integrazione: il report reale (offline, niente rete) applica ESATTAMENTE
|
|
# la formula pura per ogni asset -> wiring corretto e riproducibile.
|
|
# ---------------------------------------------------------------------------
|
|
# ---------------------------------------------------------------------------
|
|
# Feed live effimero (src/live/livefeed): merge/dedup, fallback, loader onorato.
|
|
# Nessuna rete (la coda fresca è iniettata / il fetch è stubbato a errore).
|
|
# ---------------------------------------------------------------------------
|
|
def test_merge_tail_dedups_and_tail_wins():
|
|
import pandas as pd
|
|
from src.live.livefeed import merge_tail
|
|
base = pd.DataFrame({"timestamp": [0, 300000, 600000], "open": [1, 2, 3], "high": [1, 2, 3],
|
|
"low": [1, 2, 3], "close": [1, 2, 3], "volume": [1, 1, 1],
|
|
"datetime": pd.to_datetime([0, 300000, 600000], unit="ms", utc=True)})
|
|
tail = pd.DataFrame({"timestamp": [600000, 900000], "open": [9, 4], "high": [9, 4],
|
|
"low": [9, 4], "close": [9, 4], "volume": [2, 2]})
|
|
m = merge_tail(base, tail)
|
|
assert list(m["timestamp"]) == [0, 300000, 600000, 900000] # esteso + ordinato
|
|
assert m.loc[m["timestamp"] == 600000, "close"].iloc[0] == 9 # la coda VINCE sul duplicato
|
|
assert "datetime" in m.columns and m["datetime"].is_monotonic_increasing
|
|
|
|
|
|
def test_merge_tail_empty_returns_base():
|
|
import pandas as pd
|
|
from src.live.livefeed import merge_tail
|
|
base = pd.DataFrame({"timestamp": [0], "open": [1], "high": [1], "low": [1], "close": [1], "volume": [1]})
|
|
assert merge_tail(base, pd.DataFrame()).equals(base)
|
|
|
|
|
|
def test_fresh_5m_falls_back_to_certified_on_error(monkeypatch):
|
|
import src.live.livefeed as lf
|
|
from src.data.downloader import load_data
|
|
monkeypatch.setattr(lf, "_fetch_recent_5m", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("net")))
|
|
got = lf.fresh_5m("BTC")
|
|
base = load_data("BTC", "5m")
|
|
assert len(got) == len(base) and got["timestamp"].iloc[-1] == base["timestamp"].iloc[-1]
|
|
|
|
|
|
def test_skyhook_positions_honors_custom_loader():
|
|
from src.portfolio import sleeves
|
|
calls = []
|
|
|
|
def spy(a):
|
|
calls.append(a)
|
|
return sleeves.load_data(a, "5m")
|
|
pos = sleeves._skyhook_positions(load5m=spy)
|
|
assert set(pos.keys()) == {"BTC", "ETH"} and calls == ["BTC", "ETH"]
|
|
|
|
|
|
def test_book_report_uses_pure_formula_offline():
|
|
from src.live.book import book_report
|
|
r = book_report(offline=True, equity_override=600.0)
|
|
assert r["equity"] == 600.0 and r["cap_per_asset"] > 0
|
|
for a in r["assets"]:
|
|
expect = book_net_target(a["tp_frac"], a["skh_sign"], 600.0, r["cap_per_asset"])
|
|
assert abs(a["net_target"] - expect) < _budget_arrotondamento(600.0), \
|
|
f"{a['asset']}: net incoerente con la formula"
|
|
assert a["skh_sign"] in (-1, 0, 1)
|
|
# offline -> conto assunto flat -> nessuna posizione reale, report deterministico
|
|
assert all(a["position_usd"] == 0.0 for a in r["assets"])
|
|
|
|
|
|
def test_la_formula_del_report_ha_potenza_anche_a_libro_flat(monkeypatch):
|
|
"""⚠️ I due test qui sopra leggono il segnale REALE del giorno: quando il libro e' flat
|
|
(`tp_frac=0`, `skh_sign=0`) il target e' esattamente `0.0`, i due arrotondamenti del report
|
|
sono esatti e l'invariante passa **senza essere mai esercitata**.
|
|
|
|
E' successo davvero: dal 2026-06-23 al 2026-08-18 il book e' stato flat quasi ininterrottamente
|
|
e questi test hanno avuto potenza ZERO su una formula di produzione; il difetto di tolleranza
|
|
e' emerso solo il 2026-08-21, quando TP01 e SKH01 sono andati long insieme.
|
|
E' la lezione gia' codificata il 26/07 (`test_skh_partial_entry`): **un self-check su eventi
|
|
rari si campiona sugli EVENTI, non sulla popolazione.**
|
|
|
|
Qui il segnale e' FORZATO su frazioni scomode, cosi' la copertura non dipende dal mercato.
|
|
"""
|
|
import src.live.book as book
|
|
|
|
# frazioni scelte per NON essere clean a 4 decimali e per coprire i rami della formula:
|
|
# long parziale + SKH long, long parziale + SKH short (hedge), flat + SKH short, cap.
|
|
casi = [
|
|
(0.17578123, 1),
|
|
(0.29149997, -1),
|
|
(0.0, -1),
|
|
(1.0, 1), # -> clampato al cap
|
|
]
|
|
equity, cap = 5000.0, 2500.0
|
|
# ⚠️ `base` va preso PRIMA del loop: dalla seconda iterazione `book.shadow_report` e' gia'
|
|
# sostituito dal monkeypatch e si leggerebbe il finto della iterazione precedente.
|
|
base = book.shadow_report(offline=True, equity_override=equity)
|
|
|
|
non_banali = 0
|
|
for tp_vero, sign in casi:
|
|
finto = {**base, "real_equity": equity, "equity": equity, "eq_fallback": None,
|
|
"assets": [{**a, "target": tp_vero, "position_usd": 0.0} for a in base["assets"]]}
|
|
monkeypatch.setattr(book, "shadow_report", lambda *a, _f=finto, **k: _f)
|
|
monkeypatch.setattr(book, "_skyhook_positions",
|
|
lambda load5m=None, _s=sign: {a: ("flat" if _s == 0 else
|
|
{"dir": "LONG" if _s > 0 else "SHORT"})
|
|
for a in book.ASSETS})
|
|
r = book.book_report()
|
|
atteso_vero = book_net_target(tp_vero, sign, equity, cap)
|
|
non_banali += (atteso_vero != 0.0)
|
|
for a in r["assets"]:
|
|
assert a["skh_sign"] == sign
|
|
# (a) il valore pubblicato e' la formula sul tp_frac VERO, a meno del solo round(.,2)
|
|
assert abs(a["net_target"] - atteso_vero) <= 0.005, \
|
|
f"tp={tp_vero} sign={sign}: net pubblicato incoerente con la formula"
|
|
# (b) e ricalcolandola dal tp_frac PUBBLICATO resta dentro il budget dichiarato
|
|
da_pubblicato = book_net_target(a["tp_frac"], a["skh_sign"], equity, cap)
|
|
assert abs(a["net_target"] - da_pubblicato) < _budget_arrotondamento(equity)
|
|
|
|
# controllo di POTENZA: ogni caso deve produrre un target != 0, altrimenti il test
|
|
# passerebbe per lo stesso motivo per cui passava a libro flat (0.0 arrotondato e' 0.0).
|
|
assert non_banali == len(casi), f"solo {non_banali}/{len(casi)} casi esercitano l'invariante"
|
|
|
|
|
|
def test_il_budget_di_arrotondamento_non_copre_un_errore_di_formula():
|
|
"""Controllo POSITIVO del budget: deve restare molto piu' stretto dei modi in cui la
|
|
formula puo' rompersi davvero, o sarebbe una tolleranza che assolve tutto."""
|
|
E, cap = 5000.0, 2500.0
|
|
giusto = book_net_target(0.5, 1, E, cap)
|
|
budget = _budget_arrotondamento(E)
|
|
pesi_scambiati = 0.5 * E * (W_SKH * 0.5 + W_TP01 * 1) # W_TP01/W_SKH invertiti
|
|
senza_cap = 0.5 * E * (W_TP01 * 5.0 + W_SKH * 1) # cap non applicato
|
|
for sbagliato, nome in ((pesi_scambiati, "pesi scambiati"), (senza_cap, "cap non applicato"),
|
|
(-giusto, "segno invertito")):
|
|
assert abs(giusto - sbagliato) > 100 * budget, f"{nome}: il budget lo assolverebbe"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SKH feed fallito: book_report NON crasha (forza flat + flagga skh_error) e
|
|
# book_execute DEVE farlo emergere (log + alert), non ingoiarlo silenziosamente.
|
|
# ---------------------------------------------------------------------------
|
|
def test_book_report_flags_skh_feed_error(monkeypatch):
|
|
import src.live.book as book
|
|
monkeypatch.setattr(book, "_skyhook_positions",
|
|
lambda *a, **k: (_ for _ in ()).throw(RuntimeError("feed 5m giu")))
|
|
r = book.book_report(offline=True, equity_override=600.0)
|
|
assert "skh_error" in r and "feed 5m giu" in r["skh_error"] # errore catturato + esposto
|
|
assert all(a["skh_sign"] == 0 for a in r["assets"]) # SKH forzato flat (fail-safe)
|
|
|
|
|
|
def test_book_execute_surfaces_skh_error(monkeypatch, capsys):
|
|
"""Se il report porta skh_error, _run() lo stampa E chiama notify (niente flat silenzioso)."""
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("book_execute", PROJECT_ROOT / "scripts/live/book_execute.py")
|
|
mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)
|
|
|
|
canned = dict(
|
|
last_data=_fresh_bar(), online=True, real_equity=600.0, equity=600.0, eq_basis="test",
|
|
cap_per_asset=300.0, skh_error="RuntimeError: feed 5m giu",
|
|
assets=[dict(asset="BTC", instrument="BTC_USDC-PERPETUAL", tp_frac=0.0, skh_sign=0,
|
|
skh_state="flat", net_target=0.0, position_usd=0.0, mark=60000.0, order=None)],
|
|
orders=[],
|
|
)
|
|
alerts = []
|
|
monkeypatch.setattr(mod, "book_report", lambda **k: canned)
|
|
monkeypatch.setattr(mod, "notify", lambda title, det=None: alerts.append((title, det)))
|
|
monkeypatch.setattr(mod, "load_config",
|
|
lambda: dict(execution_enabled=False, min_order_usd=5.0, disaster_sl_pct=0.30))
|
|
monkeypatch.setattr(sys, "argv", ["book_execute.py"]) # niente --execute -> dry-run
|
|
mod._run()
|
|
|
|
out = capsys.readouterr().out
|
|
assert "SKH FEED ERRORE" in out # stampato nel log
|
|
assert any("SKH feed fallito" in title for title, _ in alerts) # alert inviato
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GATE FAIL-SAFE posizione: se la read della posizione reale fallisce (ONLINE ma
|
|
# posizione IGNOTA -> assunta flat), l'esecutore NON deve operare a cieco.
|
|
# ---------------------------------------------------------------------------
|
|
def test_positions_flags_read_failure():
|
|
from src.live.shadow import ASSETS, _positions
|
|
|
|
class BadClient:
|
|
def position_usd(self, inst):
|
|
raise RuntimeError("api 500")
|
|
|
|
pos, note, err = _positions(BadClient())
|
|
assert all(pos[a] == 0.0 for a in ASSETS) # assunta flat (fail-safe)
|
|
assert err is not None and "non leggibile" in err # ma SEGNALATA (non silenziosa)
|
|
|
|
|
|
def test_positions_ok_and_offline_have_no_error():
|
|
from src.live.shadow import ASSETS, _positions
|
|
|
|
class GoodClient:
|
|
def position_usd(self, inst):
|
|
return 123.0
|
|
|
|
_, _, err_ok = _positions(GoodClient())
|
|
_, _, err_off = _positions(None) # offline -> gestito dal gate 'online'
|
|
assert err_ok is None and err_off is None
|
|
|
|
|
|
def test_book_report_propagates_pos_error(monkeypatch):
|
|
import src.live.book as book
|
|
base = book.shadow_report(offline=True, equity_override=600.0)
|
|
monkeypatch.setattr(book, "shadow_report", lambda **k: {**base, "pos_error": "IGNOTA"})
|
|
r = book.book_report(offline=True, equity_override=600.0)
|
|
assert r.get("pos_error") == "IGNOTA" # propagato fino al book
|
|
|
|
|
|
def test_book_execute_halts_on_unreadable_position(monkeypatch, capsys):
|
|
"""ARMATO + --execute + ordine presente: il gate DEVE fermarsi PRIMA di costruire il trader."""
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("book_execute", PROJECT_ROOT / "scripts/live/book_execute.py")
|
|
mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)
|
|
|
|
canned = dict(
|
|
last_data=_fresh_bar(), online=True, real_equity=598.0, equity=598.0, eq_basis="mainnet USDC",
|
|
cap_per_asset=300.0, skh_error=None,
|
|
pos_error="posizione non leggibile, assunta FLAT: BTC (RuntimeError: api 500)",
|
|
assets=[dict(asset="BTC", instrument="BTC_USDC-PERPETUAL", tp_frac=1.0, skh_sign=1,
|
|
skh_state="flat", net_target=225.0, position_usd=0.0, mark=60000.0,
|
|
order=dict(side="buy"))], # ordine presente: senza gate proverebbe a eseguire
|
|
orders=[dict(side="buy")],
|
|
)
|
|
alerts = []
|
|
monkeypatch.setattr(mod, "book_report", lambda **k: canned)
|
|
monkeypatch.setattr(mod, "notify", lambda title, det=None: alerts.append((title, det)))
|
|
monkeypatch.setattr(mod, "load_config",
|
|
lambda: dict(execution_enabled=True, min_order_usd=5.0, disaster_sl_pct=0.30))
|
|
monkeypatch.setattr(sys, "argv", ["book_execute.py", "--execute"]) # ARMATO + execute
|
|
|
|
def boom(*a, **k):
|
|
raise AssertionError("DeribitTrader NON deve essere costruito: il gate deve fermarsi prima")
|
|
monkeypatch.setattr(mod, "DeribitTrader", boom)
|
|
|
|
mod._run()
|
|
out = capsys.readouterr().out
|
|
assert "POSIZIONE NON LEGGIBILE" in out # stampato
|
|
assert any("posizione non leggibile" in t for t, _ in alerts) # alert inviato
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DIAGNOSTICA equity (Opzione B): ONLINE ma equity reale non leggibile -> sizing su
|
|
# paper_cap. NON blocca (l'hard-cap $/asset protegge), ma DEVE essere segnalato.
|
|
# ---------------------------------------------------------------------------
|
|
def test_shadow_flags_eq_fallback_when_online_but_equity_unreadable(monkeypatch):
|
|
import src.live.shadow as sh
|
|
monkeypatch.setattr(sh, "_safe_client", lambda: object()) # online
|
|
monkeypatch.setattr(sh, "_marks", lambda c, d: ({a: 60000.0 for a in sh.ASSETS},
|
|
{a: "mainnet" for a in sh.ASSETS}))
|
|
monkeypatch.setattr(sh, "_positions", lambda c: ({a: 0.0 for a in sh.ASSETS}, "mainnet", None))
|
|
monkeypatch.setattr(sh, "_equity", lambda c, m: (None, "conto flat / non finanziato")) # equity IGNOTA
|
|
r = sh.shadow_report() # online, no override
|
|
assert r["online"] is True and r["real_equity"] is None
|
|
assert r.get("eq_fallback") and "paper_cap" in r["eq_fallback"] # fallback SEGNALATO
|
|
|
|
|
|
def test_shadow_no_eq_fallback_offline_or_override():
|
|
from src.live.shadow import shadow_report
|
|
assert shadow_report(offline=True, equity_override=600.0).get("eq_fallback") is None # override
|
|
assert shadow_report(offline=True).get("eq_fallback") is None # offline != fallback-online
|
|
|
|
|
|
def test_book_report_propagates_eq_fallback(monkeypatch):
|
|
import src.live.book as book
|
|
base = book.shadow_report(offline=True, equity_override=600.0)
|
|
monkeypatch.setattr(book, "shadow_report", lambda **k: {**base, "eq_fallback": "PAPER_CAP"})
|
|
assert book.book_report(offline=True, equity_override=600.0).get("eq_fallback") == "PAPER_CAP"
|
|
|
|
|
|
def test_book_execute_eq_fallback_warns_but_proceeds(monkeypatch, capsys):
|
|
"""eq_fallback: avvisa + notify MA PROSEGUE (non e' un halt, a differenza di pos_error)."""
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("book_execute", PROJECT_ROOT / "scripts/live/book_execute.py")
|
|
mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)
|
|
|
|
canned = dict(
|
|
last_data=_fresh_bar(), online=True, real_equity=None, equity=2000.0,
|
|
eq_basis="paper capital (ipotetico)", cap_per_asset=300.0, skh_error=None, pos_error=None,
|
|
eq_fallback="equity reale non leggibile (conto flat) -> sizing su paper_cap $2,000",
|
|
assets=[dict(asset="BTC", instrument="BTC_USDC-PERPETUAL", tp_frac=0.0, skh_sign=0,
|
|
skh_state="flat", net_target=0.0, position_usd=0.0, mark=60000.0, order=None)],
|
|
orders=[],
|
|
)
|
|
alerts = []
|
|
|
|
class FakeDT:
|
|
def ensure_disaster_sl(self, inst, sl_pct): return {"state": "flat"}
|
|
def position_usd(self, inst): return 0.0
|
|
|
|
monkeypatch.setattr(mod, "book_report", lambda **k: canned)
|
|
monkeypatch.setattr(mod, "notify", lambda title, det=None: alerts.append((title, det)))
|
|
monkeypatch.setattr(mod, "DeribitTrader", lambda: FakeDT())
|
|
monkeypatch.setattr(mod, "load_config",
|
|
lambda: dict(execution_enabled=True, min_order_usd=5.0, disaster_sl_pct=0.30))
|
|
monkeypatch.setattr(sys, "argv", ["book_execute.py", "--execute"])
|
|
|
|
mod._run()
|
|
out = capsys.readouterr().out
|
|
assert "EQUITY FALLBACK" in out # avvisato
|
|
assert any("equity fallback" in t for t, _ in alerts) # alert inviato
|
|
assert "Nessuna azione" in out # HA PROSEGUITO (non halt)
|