ops(live): cabla il forward-monitor DVOLSPREAD nel cron + gate pre-registrato

Chiude la lezione (c) dell'ondata 26/07: "un lead in forward-monitor senza
monitor e senza scadenza e' un lead perso" — DVOLSPREAD era in limbo da 35
giorni. Ora ha config congelata, monitor nel cron e gate con data.

Config congelata = la cella scelta IN-SAMPLE (zwin=180 k=2.0 lw=0.6 zw=1.1
tgt=0.17 svw=60), NON quella pubblicata dall'agente: quella era 83a/729
sull'hold-out ma 471a/729 in-sample e fallisce il deflated-Sharpe (0.947),
la congelata lo passa (0.953). Riusa make_book di r0726_dvolspread_gate,
nessuna reimplementazione.

Strumentazione specifica: il book va flat quando manca il DVOL, quindi un feed
rotto produrrebbe zeri che contati come evidenza direbbero "nessuna perdita"
invece di "nessuna misura". Contabilita' a 3 stati (attive / flat-da-segnale /
flat-senza-dato), finestra misurata in barre ATTIVE, e guardia che esce 1 se
FROZEN diverge dallo stato salvato.

Gate pre-registrato: kill 2026-10-24 se Sharpe < -0.50; decisione 2027-01-24
solo se Sharpe>0 E marginale ADDS E deflated-Sharpe>=0.95 E weights_tilt_null;
veto d'integrita' se barre attive <80% (si estende, non si decide su dati
mancanti). La soglia su Sharpe e' debole di proposito: con ~180 barre
SE(Sharpe)~1.4, una soglia alta sarebbe finta precisione.

Inception 2026-07-25, apertura +0.184 = $111/gamba (cap $300).
Book/pesi INVARIATI. Suite 255 verdi.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tests/test_paper_dvolspread.py

@@ -0,0 +1,168 @@
+"""Test del forward-monitor DVOLSPREAD (scripts/live/paper_dvolspread.py).
+
+Blindano le proprieta' che rendono la finestra forward INTERPRETABILE. Il rischio specifico di
+questo lead non e' che il segnale sbagli — e' che il monitor **conti come evidenza dei giorni in
+cui non c'era il dato**: il book va flat quando il DVOL manca, e una serie di zeri sembra
+"nessuna perdita" quando invece e' "nessuna misura". Da qui la contabilita' a tre stati.
+
+I test scrivono in una directory temporanea: NON toccano lo stato reale del monitor.
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+for p in (ROOT, ROOT / "scripts" / "research", ROOT / "scripts" / "research" / "alt",
+          ROOT / "scripts" / "research" / "ortho", ROOT / "scripts" / "live"):
+    sys.path.insert(0, str(p))
+
+pytest.importorskip("pandas")
+import pandas as pd  # noqa: E402
+
+import paper_dvolspread as pm  # noqa: E402
+
+
+@pytest.fixture()
+def sandbox(tmp_path, monkeypatch):
+    """Reindirizza lo stato del monitor in tmp: i test non devono mai scrivere in data/."""
+    monkeypatch.setattr(pm, "STATE_DIR", tmp_path)
+    monkeypatch.setattr(pm, "STATE_FILE", tmp_path / "state.json")
+    monkeypatch.setattr(pm, "RETURNS_FILE", tmp_path / "returns.jsonl")
+    return tmp_path
+
+
+def _fake_panel(n=10, has_dvol=None, wb=None):
+    ts = np.arange(n, dtype=np.int64) * 86_400_000 + 1_700_000_000_000
+    return dict(ts=ts, dt=pd.to_datetime(ts, unit="ms", utc=True).values,
+                wb=np.full(n, 0.2) if wb is None else np.asarray(wb, float),
+                we=np.full(n, -0.2) if wb is None else -np.asarray(wb, float),
+                rb=np.full(n, 0.001), re=np.zeros(n),
+                has_dvol=np.ones(n, bool) if has_dvol is None else np.asarray(has_dvol, bool))
+
+
+# ------------------------------------------------------------------ config congelata
+
+def test_la_config_congelata_e_la_cella_IN_SAMPLE_non_quella_pubblicata():
+    """La cella pubblicata dall'agente e' selection-on-holdout e FALLISCE il deflated-Sharpe
+    (0.947). Il monitor deve girare quella scelta in-sample (DSR 0.953). Se qualcuno rimettesse
+    i parametri pubblicati, il monitor misurerebbe l'oggetto sbagliato."""
+    assert pm.FROZEN == dict(zwin=180, tanh_k=2.0, lw=0.6, zw=1.1, tgt=0.17, svw=60)
+    pubblicata = dict(zwin=150, tanh_k=1.8, lw=0.45, zw=1.0, tgt=0.13, svw=50)
+    assert pm.FROZEN != pubblicata
+
+
+def test_il_gate_e_pre_registrato_e_ordinato():
+    kill = pd.Timestamp(pm.KILL_DATE, tz="UTC")
+    gate = pd.Timestamp(pm.GATE_DATE, tz="UTC")
+    assert kill < gate, "il kill anticipato deve precedere la decisione piena"
+    assert pm.KILL_SHARPE < 0.0
+    assert 0.5 <= pm.MIN_ACTIVE_FRAC <= 1.0
+
+
+# ------------------------------------------------------------------ market-neutralita'
+
+def test_il_book_e_market_neutral_per_costruzione():
+    """w_eth == -w_btc: e' l'unica ragione per cui il lead ha corr ~0.11 a TP01."""
+    import ortholib as ol
+    btc, eth = ol.aligned()
+    wb, we = pm.make_book(**pm.FROZEN)(btc, eth)
+    assert np.allclose(np.asarray(wb, float), -np.asarray(we, float), atol=1e-12)
+
+
+def test_la_posizione_sta_dentro_il_cap_di_nozionale():
+    """A $600 il guardrail live e' $300/asset -> |w| <= 0.5 sempre."""
+    import ortholib as ol
+    btc, eth = ol.aligned()
+    wb, _ = pm.make_book(**pm.FROZEN)(btc, eth)
+    w = np.nan_to_num(np.asarray(wb, float))
+    assert np.abs(w).max() <= pm.CAP_NOTIONAL / pm.REAL_CAPITAL + 1e-9
+
+
+# ------------------------------------------------------------------ contabilita' dell'attivita'
+
+def test_i_giorni_senza_DVOL_non_sono_contati_come_attivi(sandbox):
+    """IL test centrale. Se il feed DVOL si rompe il book va flat: quei giorni devono finire in
+    'flat-senza-dato', non passare per giorni buoni senza perdite."""
+    P = _fake_panel(n=10, has_dvol=[True] * 4 + [False] * 6)
+    st = pm.init_state(P)
+    st["last_ts"] = int(P["ts"][0])          # forza il processing delle barre 1..9
+    st = pm.advance(st, P)
+    assert st["n_flat_nodata"] == 6, st
+    assert st["n_active"] == 3, st           # barre 1,2,3 con dvol e posizione != 0
+    assert st["n_active"] + st["n_flat_signal"] + st["n_flat_nodata"] == st["n_bars"]
+
+
+def test_flat_da_segnale_e_flat_senza_dato_sono_stati_distinti(sandbox):
+    """Posizione 0 CON dvol = scelta del segnale (informazione). Posizione 0 SENZA dvol =
+    assenza di misura. Confonderli renderebbe il veto d'integrita' inutile."""
+    P = _fake_panel(n=6, has_dvol=[True] * 6, wb=[0.2, 0.0, 0.0, 0.2, 0.0, 0.2])
+    st = pm.init_state(P)
+    st["last_ts"] = int(P["ts"][0])
+    st = pm.advance(st, P)
+    assert st["n_flat_nodata"] == 0
+    assert st["n_flat_signal"] == 3, st
+    assert st["n_active"] == 2, st
+
+
+# ------------------------------------------------------------------ macchina a stati
+
+def test_advance_e_idempotente(sandbox):
+    """Rilanciare il cron due volte lo stesso giorno non deve raddoppiare la finestra."""
+    P = _fake_panel(n=8)
+    st = pm.init_state(P)
+    st["last_ts"] = int(P["ts"][0])
+    st = pm.advance(st, P)
+    snap = dict(st)
+    st2 = pm.advance(st, P)
+    for k in ("n_bars", "n_active", "cap_modeled", "cap_real", "last_ts"):
+        assert st2[k] == snap[k], k
+
+
+def test_inception_e_forward_only(sandbox):
+    """All'inizializzazione la finestra parte dall'ultima barra chiusa: lo storico NON entra."""
+    P = _fake_panel(n=50)
+    st = pm.init_state(P)
+    assert st["n_bars"] == 0
+    assert st["start_ts"] == int(P["ts"][-1])
+    assert st["cap_modeled"] == pm.MODELED_CAPITAL
+    assert pm.advance(st, P)["n_bars"] == 0      # nessuna barra nuova -> nessun avanzamento
+
+
+def test_il_libro_reale_salta_i_ribilanci_sotto_il_min_order(sandbox):
+    """A $600 un Delta-peso da meno di $5 di nozionale NON si esegue.
+
+    ⚠️ La soglia si misura sul gap CUMULATO contro la posizione TENUTA, non sul passo di una
+    singola barra: un drift da $3/barra esegue comunque ogni due barre. Il test copre entrambi
+    i rami — drift che resta sotto soglia per tutta la finestra, e salto singolo sopra soglia."""
+    step = 0.5 / pm.REAL_CAPITAL          # $0.50/barra -> dopo 5 barre $2.50, sempre < $5
+    P = _fake_panel(n=6, wb=[0.2 + k * step for k in range(6)])
+    st = pm.init_state(P)
+    st["last_ts"] = int(P["ts"][0])
+    st["wb_real"] = st["wb_modeled"] = 0.2
+    st = pm.advance(st, P)
+    assert abs(st["wb_real"] - 0.2) < 1e-12, "sotto soglia il libro reale non deve ribilanciare"
+    assert abs(st["wb_modeled"] - (0.2 + 5 * step)) < 1e-9, "il modellato ribilancia sempre"
+
+    # ramo opposto: un salto ampio DEVE eseguire anche sul libro reale
+    P2 = _fake_panel(n=3, wb=[0.2, 0.45, 0.45])
+    st2 = pm.init_state(P2)
+    st2["last_ts"] = int(P2["ts"][0])
+    st2["wb_real"] = st2["wb_modeled"] = 0.2
+    st2 = pm.advance(st2, P2)
+    assert abs(st2["wb_real"] - 0.45) < 1e-9, "sopra soglia il libro reale deve ribilanciare"
+
+
+def test_le_barre_registrate_portano_il_flag_del_dvol(sandbox):
+    """Il log deve permettere di ricostruire a posteriori quali barre erano valide."""
+    P = _fake_panel(n=5, has_dvol=[True, True, False, True, False])
+    st = pm.init_state(P)
+    st["last_ts"] = int(P["ts"][0])
+    pm.advance(st, P)
+    rows = [json.loads(x) for x in pm.RETURNS_FILE.read_text().splitlines()]
+    assert len(rows) == 4
+    assert [r["has_dvol"] for r in rows] == [True, False, True, False]