test(integration): IV-RV adaptive gate end-to-end con SQLite reale

Verifica integrazione tra Repository.iv_rv_history, compute_adaptive_threshold e dvol_lookback su un DB reale seedato con 30 giorni di market_snapshots bimodale. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-08 23:09:27 +00:00
parent 3190764f64
commit 98111814d2
1 changed files with 108 additions and 0 deletions
@@ -0,0 +1,108 @@
+"""End-to-end test del gate IV-RV adattivo + Vol-of-Vol guard via Repository."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+
+import pytest
+
+from cerbero_bite.state.db import connect, run_migrations
+from cerbero_bite.state.models import MarketSnapshotRecord
+from cerbero_bite.state.repository import Repository
+
+
+def _seed_history(
+    conn,
+    repo: Repository,
+    asset: str,
+    base: datetime,
+    n_ticks: int,
+    iv_rv_value: Decimal,
+    dvol_value: Decimal,
+) -> None:
+    for i in range(n_ticks):
+        repo.record_market_snapshot(
+            conn,
+            MarketSnapshotRecord(
+                timestamp=base + timedelta(minutes=15 * i),
+                asset=asset,
+                spot=Decimal("2000"),
+                dvol=dvol_value,
+                realized_vol_30d=Decimal("48"),
+                iv_minus_rv=iv_rv_value,
+                funding_perp_annualized=Decimal("0"),
+                funding_cross_annualized=Decimal("0"),
+                dealer_net_gamma=Decimal("0"),
+                gamma_flip_level=None,
+                oi_delta_pct_4h=None,
+                liquidation_long_risk="low",
+                liquidation_short_risk="low",
+                macro_days_to_event=None,
+                fetch_ok=True,
+                fetch_errors_json=None,
+            ),
+        )
+    conn.commit()
+
+
+@pytest.fixture
+def db_30d(tmp_path):
+    """30 giorni di storia con IV-RV bimodale: prima metà 1.0, seconda metà 5.0."""
+    db_path = tmp_path / "e2e.sqlite"
+    conn = connect(str(db_path))
+    run_migrations(conn)
+    repo = Repository()
+    base = datetime(2026, 4, 1, 0, 0, tzinfo=UTC)
+    _seed_history(conn, repo, "ETH", base, 1440, Decimal("1.0"), Decimal("50"))
+    _seed_history(
+        conn,
+        repo,
+        "ETH",
+        base + timedelta(days=15),
+        1440,
+        Decimal("5.0"),
+        Decimal("50"),
+    )
+    return conn, repo
+
+
+def test_iv_rv_history_p25_picks_up_recent_regime(db_30d) -> None:
+    """Sanity: con bimodale 1.0/5.0 e finestra 30g, P25 di tutta la
+    storia è 1.0 (il 25° centile è ancora nella metà bassa)."""
+    conn, repo = db_30d
+    history = repo.iv_rv_history(
+        conn,
+        asset="ETH",
+        max_days=60,
+        as_of=datetime(2026, 5, 1, 0, 0, tzinfo=UTC),
+    )
+    assert len(history) == 2880
+    from cerbero_bite.core.adaptive_threshold import compute_adaptive_threshold
+
+    threshold = compute_adaptive_threshold(
+        history=history,
+        percentile=Decimal("0.25"),
+        absolute_floor=Decimal("0"),
+        min_days=30,
+        target_days=60,
+    )
+    assert threshold == Decimal("1.0")
+
+
+def test_dvol_lookback_within_tolerance(db_30d) -> None:
+    conn, repo = db_30d
+    base = datetime(2026, 4, 1, 0, 0, tzinfo=UTC)
+    out = repo.dvol_lookback(conn, asset="ETH", reference=base + timedelta(hours=24))
+    assert out == Decimal("50")
+
+
+def test_dvol_lookback_returns_none_outside_tolerance(db_30d) -> None:
+    conn, repo = db_30d
+    out = repo.dvol_lookback(
+        conn,
+        asset="ETH",
+        reference=datetime(2025, 1, 1, tzinfo=UTC),
+        tolerance_minutes=15,
+    )
+    assert out is None