"""Lock del forward-monitor XSR01 (candidato trovato 2026-07-25). Pin degli invarianti che rendono la finestra forward interpretabile. Se uno di questi salta, i numeri accumulati non valgono piu' e la finestra va azzerata consapevolmente (--reset), non lasciata scorrere con una config diversa. """ from __future__ import annotations import sys from pathlib import Path import numpy as np import pytest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "scripts" / "live")) sys.path.insert(0, str(ROOT / "scripts" / "research")) pytest.importorskip("pandas") import pandas as pd # noqa: E402 paper_xsr = pytest.importorskip("paper_xsr") def _panel(): try: return paper_xsr.build_panel() except (FileNotFoundError, ValueError): pytest.skip("parquet Hyperliquid assenti") def test_config_congelata(): """La config e' il contratto della finestra forward: cambiarla la invalida.""" from r0725_statarb_multi import CAP, SGN, TARGET_VOL, VOL_WIN, W assert (W, SGN, TARGET_VOL, VOL_WIN, CAP) == (45, +1, 0.20, 30, 2.0) assert paper_xsr.FEE_LEG == 0.0005 assert paper_xsr.MIN_ORDER == 5.0 assert paper_xsr.MODELED_CAPITAL == 2000.0 def test_dollar_neutral_per_costruzione(): """Il demeaning DEVE dare somma dei pesi ~0: e' cio' che annulla la gamba BTC comune.""" _, _, W, _, _, _ = _panel() net = np.abs(W.sum(axis=1)) assert np.nanmax(net) < 1e-12, f"non dollar-neutral: max |somma pesi| = {np.nanmax(net)}" def test_timestamp_in_millisecondi_non_1970(): """Trappola pandas gia' pagata dal progetto: astype/view int64 su indice tz-aware. I ts devono essere ms epoch plausibili, non nanosecondi ne' secondi.""" ts, idx, _, _, _, _ = _panel() assert ts.dtype == np.int64 assert pd.Timestamp(int(ts[-1]), unit="ms", tz="UTC").year >= 2024 assert pd.Timestamp(int(ts[0]), unit="ms", tz="UTC").year >= 2023 # coerenza con l'indice da cui derivano assert pd.Timestamp(int(ts[-1]), unit="ms", tz="UTC").date() == idx[-1].date() def test_universo_50_alt_senza_btc(): _, _, _, _, _, syms = _panel() assert len(syms) >= 45, f"attesi ~50 alt certificati, trovati {len(syms)}" assert "BTC" not in syms, "BTC e' il riferimento, non una gamba" def test_step_causale_la_posizione_nuova_non_guadagna_la_barra_corrente(): """Il rendimento della barra si calcola sui pesi TENUTI (decisi prima), mai sul target.""" k = 3 bk = paper_xsr._book(1000.0, k) bk["w"] = [0.10, -0.10, 0.0] w_target = np.array([-0.50, 0.50, 0.0]) # ribaltamento totale r_alt = np.array([0.10, -0.10, 0.0]) # premia i pesi VECCHI net = paper_xsr._step(bk, w_target, r_alt, 0.0, None) assert net > 0, "il netto deve riflettere i pesi tenuti, non il nuovo target" assert bk["w"] == pytest.approx(list(w_target)) # e dopo si e' ribilanciato def test_min_order_blocca_le_gambe_piccole(): """A capitale piccolo le gambe sub-min-order NON si muovono: e' il punto dei libri REAL.""" k = 2 bk = paper_xsr._book(600.0, k) w_target = np.array([0.001, -0.001]) # 600*0.001 = $0.60 << $5 paper_xsr._step(bk, w_target, np.zeros(k), 0.0, paper_xsr.MIN_ORDER) assert bk["w"] == [0.0, 0.0], "gamba sub-min-order eseguita: il libro REAL sarebbe fittizio" assert bk["n_skip"] == 2 big = paper_xsr._book(600.0, k) paper_xsr._step(big, np.array([0.20, -0.20]), np.zeros(k), 0.0, paper_xsr.MIN_ORDER) assert big["w"] == pytest.approx([0.20, -0.20]) # $120 per gamba: si esegue assert big["n_fill"] == 2 def test_fee_addebitata_su_alt_piu_gamba_btc(): """La fee copre le gambe alt E la gamba BTC netta: a ribilanciamento nullo, netto = 0.""" k = 2 bk = paper_xsr._book(1000.0, k) net = paper_xsr._step(bk, np.zeros(k), np.zeros(k), 0.0, None) assert net == pytest.approx(0.0) bk2 = paper_xsr._book(1000.0, k) net2 = paper_xsr._step(bk2, np.array([0.5, -0.5]), np.zeros(k), 0.0, None) assert net2 == pytest.approx(-paper_xsr.FEE_LEG * 1.0) # |0.5|+|−0.5| alt, BTC netto 0