From f88925895238f90592ee597eaa44edcc63cb4ede Mon Sep 17 00:00:00 2001 From: root Date: Fri, 8 May 2026 19:58:32 +0000 Subject: [PATCH] docs(plans): IV-RV adaptive gate implementation plan Piano TDD bite-sized in 11 task con steps dettagliati, codice completo, comandi e expected output. Coverage completa dello spec 2026-05-08-iv-rv-adaptive-gate-design.md. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../plans/2026-05-08-iv-rv-adaptive-gate.md | 1596 +++++++++++++++++ 1 file changed, 1596 insertions(+) create mode 100644 docs/superpowers/plans/2026-05-08-iv-rv-adaptive-gate.md diff --git a/docs/superpowers/plans/2026-05-08-iv-rv-adaptive-gate.md b/docs/superpowers/plans/2026-05-08-iv-rv-adaptive-gate.md new file mode 100644 index 0000000..4f1f881 --- /dev/null +++ b/docs/superpowers/plans/2026-05-08-iv-rv-adaptive-gate.md @@ -0,0 +1,1596 @@ +# IV-RV Adaptive Gate Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Sostituire la soglia statica del gate IV-RV in `validate_entry` con una soglia adattiva (P25 rolling sui market_snapshots) e aggiungere un Vol-of-Vol guard (|ΔDVOL_24h|≥5pt). Backwards compat: gate adattivo è opt-in, comportamento legacy invariato quando il flag è off. + +**Architecture:** +- Funzione pura `compute_adaptive_threshold` in `core/adaptive_threshold.py` (no I/O, testabile in isolamento) +- `validate_entry` riceve `iv_rv_history` e `dvol_24h_ago` come campi del nuovo `EntryContext`; resta puro +- `entry_cycle` carica history+lookback da nuovi metodi `Repository` prima di costruire ctx +- GUI Calibrazione mostra pannello informativo che usa `compute_adaptive_threshold` + +**Tech Stack:** Python 3.12, Pydantic 2, sqlite3 stdlib, pytest 8, hypothesis (per property tests opzionali). Streamlit per la GUI. + +**Spec di riferimento:** `docs/superpowers/specs/2026-05-08-iv-rv-adaptive-gate-design.md` + +--- + +## File Structure + +**Create:** +- `src/cerbero_bite/core/adaptive_threshold.py` — funzione pura `compute_adaptive_threshold` +- `tests/unit/test_adaptive_threshold.py` — unit test funzione pura +- `tests/integration/test_entry_cycle_iv_rv_adaptive.py` — integration end-to-end con SQLite + +**Modify:** +- `src/cerbero_bite/config/schema.py` — campi adaptive in `EntryConfig` +- `src/cerbero_bite/core/entry_validator.py` — `EntryContext` + nuovo path nel `validate_entry` +- `src/cerbero_bite/state/repository.py` — `iv_rv_history()` e `dvol_lookback()` +- `src/cerbero_bite/runtime/entry_cycle.py` — carica history+lookback e popola ctx +- `src/cerbero_bite/gui/pages/6_📐_Calibrazione.py` — pannello "Gate adattivo" +- `strategy.aggressiva.yaml` — abilita modalità adattiva +- `tests/unit/test_entry_validator.py` — test nuovi path adaptive (additivi) + +**No new migration:** la spec esplicita di non aggiungere tabelle nuove. Solo letture sopra a `market_snapshots`. + +--- + +## Task 1: Aggiungere campi config in `EntryConfig` + +**Files:** +- Modify: `src/cerbero_bite/config/schema.py:78-85` +- Test: `tests/unit/test_config_schema.py` (verifica esistenza con grep, se non esiste skip; il check è di non-regressione) + +- [ ] **Step 1: Aggiungere i campi all'`EntryConfig` dopo `iv_minus_rv_filter_enabled`** + +Modifica `src/cerbero_bite/config/schema.py:78-85`: + +```python + # IV richness filter (§2.9). `iv_minus_rv_min` è la soglia in + # punti vol che la IV implicita 30g deve eccedere la RV30g per + # ammettere l'entry. Letteratura short-vol systematic: l'edge + # sostenibile esiste solo con un margine misurabile fra IV e RV. + # Default disabilitato + soglia 0 per non bloccare l'avvio finché + # non si è calibrato sui dati raccolti (vedi `📐 Calibrazione`). + iv_minus_rv_min: Decimal = Field(default=Decimal("0")) + iv_minus_rv_filter_enabled: bool = False + + # IV richness gate adattivo (Phase 5+). Quando + # `iv_minus_rv_adaptive_enabled=True`, la soglia statica + # `iv_minus_rv_min` diventa il floor assoluto e la soglia + # effettiva è `max(P_q rolling, floor)` calcolata su + # `market_snapshots`. Vedi + # `docs/superpowers/specs/2026-05-08-iv-rv-adaptive-gate-design.md`. + iv_minus_rv_adaptive_enabled: bool = False + iv_minus_rv_percentile: Decimal = Field(default=Decimal("0.25")) + iv_minus_rv_window_target_days: int = 60 + iv_minus_rv_window_min_days: int = 30 + + # Vol-of-Vol guard (§4-quater roadmap punto 2): blocca entry se + # |DVOL_now - DVOL_24h_ago| supera la soglia. Cattura regime + # shift bruschi non riflessi nel percentile rolling. + vol_of_vol_guard_enabled: bool = False + vol_of_vol_threshold_pt: Decimal = Field(default=Decimal("5")) + vol_of_vol_lookback_hours: int = 24 +``` + +- [ ] **Step 2: Verificare che la golden config (default) si carichi senza errori** + +Run: `cd /opt/docker/cerbero-bite && python -c "from cerbero_bite.config.schema import golden_config; c = golden_config(); print(c.entry.iv_minus_rv_adaptive_enabled, c.entry.iv_minus_rv_percentile, c.entry.vol_of_vol_guard_enabled)"` + +Expected output: `False 0.25 False` + +- [ ] **Step 3: Esegui l'intera suite per non-regressione** + +Run: `cd /opt/docker/cerbero-bite && pytest tests/unit/ -x -q 2>&1 | tail -20` + +Expected: tutti pass (i nuovi default mantengono il comportamento legacy). + +- [ ] **Step 4: Commit** + +```bash +git -C /opt/docker/cerbero-bite add src/cerbero_bite/config/schema.py +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +feat(config): EntryConfig campi adaptive IV-RV gate + VoV guard + +Aggiunge i flag e i parametri per il gate IV-RV adattivo (P25 +rolling) e per il Vol-of-Vol guard. Default disabilitati per +non cambiare comportamento dei profili attuali. + +Vedi docs/superpowers/specs/2026-05-08-iv-rv-adaptive-gate-design.md + +Co-Authored-By: Claude Opus 4.7 (1M context) +EOF +)" +``` + +--- + +## Task 2: Pure function `compute_adaptive_threshold` — TDD + +**Files:** +- Create: `src/cerbero_bite/core/adaptive_threshold.py` +- Create: `tests/unit/test_adaptive_threshold.py` + +- [ ] **Step 1: Scrivere il test di warmup (history vuota)** + +Crea `tests/unit/test_adaptive_threshold.py`: + +```python +"""TDD per :mod:`cerbero_bite.core.adaptive_threshold`. + +Spec: ``docs/superpowers/specs/2026-05-08-iv-rv-adaptive-gate-design.md``. +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from cerbero_bite.core.adaptive_threshold import compute_adaptive_threshold + + +# --------------------------------------------------------------------------- +# Warmup +# --------------------------------------------------------------------------- + + +def test_empty_history_returns_none() -> None: + out = compute_adaptive_threshold( + history=[], + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + assert out is None +``` + +- [ ] **Step 2: Eseguire il test e vederlo fallire (modulo non esiste)** + +Run: `cd /opt/docker/cerbero-bite && pytest tests/unit/test_adaptive_threshold.py::test_empty_history_returns_none -v` + +Expected: FAIL con `ModuleNotFoundError: No module named 'cerbero_bite.core.adaptive_threshold'` + +- [ ] **Step 3: Implementazione minima per far passare il test** + +Crea `src/cerbero_bite/core/adaptive_threshold.py`: + +```python +"""Funzione pura per calcolare la soglia adattiva del gate IV-RV. + +Spec: ``docs/superpowers/specs/2026-05-08-iv-rv-adaptive-gate-design.md``. + +Determinismic, no I/O. La query del repository è effettuata dal caller +(``runtime/entry_cycle``) prima di chiamare questa funzione. +""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Sequence + +__all__ = ["compute_adaptive_threshold"] + +_TICKS_PER_DAY = 96 # cron */15 → 4 tick/h × 24h + + +def compute_adaptive_threshold( + history: Sequence[Decimal], + *, + percentile: Decimal, + absolute_floor: Decimal, + min_days: int, + target_days: int, +) -> Decimal | None: + """Ritorna la soglia adattiva o ``None`` durante il warmup hard. + + Args: + history: Sequenza ordinata ASC dei valori IV-RV (un valore per + ogni tick disponibile, max ``target_days * 96``). NULL e + tick non riusciti devono essere già stati filtrati dal + caller. + percentile: Quantile target nella distribuzione (es. ``0.25``). + absolute_floor: Floor minimo applicato dopo il calcolo del + percentile. La soglia restituita è + ``max(P_q, absolute_floor)``. + min_days: Sotto questa soglia di giorni di storia, la finestra + usata è "tutta la storia disponibile". Sopra, la finestra è + fissa a ``min_days`` finché non si raggiunge ``target_days``. + target_days: Finestra finale stabile. + + Returns: + ``None`` se la storia è < 1 giorno (warmup hard, gate + disabilitato), altrimenti il percentile della finestra, + bounded dal floor. + """ + if not history: + return None + n_ticks = len(history) + if n_ticks < _TICKS_PER_DAY: + return None + if n_ticks >= target_days * _TICKS_PER_DAY: + window = history[-target_days * _TICKS_PER_DAY:] + elif n_ticks >= min_days * _TICKS_PER_DAY: + window = history[-min_days * _TICKS_PER_DAY:] + else: + window = list(history) + return max(_percentile(window, percentile), absolute_floor) + + +def _percentile(values: Sequence[Decimal], q: Decimal) -> Decimal: + """Linear-interpolated percentile, NumPy-compatible (method='linear'). + + Implementato in Decimal puro per evitare dipendenze numpy nel core. + """ + if not values: + raise ValueError("percentile of empty sequence") + sorted_v = sorted(values) + n = len(sorted_v) + k = (Decimal(n) - Decimal(1)) * q + f = int(k) # floor + c = min(f + 1, n - 1) + if f == c: + return sorted_v[f] + frac = k - Decimal(f) + return sorted_v[f] + (sorted_v[c] - sorted_v[f]) * frac +``` + +- [ ] **Step 4: Eseguire il test, deve passare** + +Run: `cd /opt/docker/cerbero-bite && pytest tests/unit/test_adaptive_threshold.py::test_empty_history_returns_none -v` + +Expected: PASS. + +- [ ] **Step 5: Aggiungere test di warmup parziale (<1 giorno)** + +Aggiungi a `tests/unit/test_adaptive_threshold.py`: + +```python +def test_history_under_one_day_returns_none() -> None: + out = compute_adaptive_threshold( + history=[Decimal("3")] * 50, # 50 tick < 96 + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + assert out is None + + +def test_history_exactly_one_day_returns_percentile() -> None: + history = [Decimal(i) / Decimal("10") for i in range(96)] # 0.0..9.5 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # P25 di [0.0..9.5] passo 0.1 con method='linear': k=23.75, val ≈ 2.375 + assert out is not None + assert Decimal("2.3") < out < Decimal("2.5") +``` + +Run: `pytest tests/unit/test_adaptive_threshold.py -v` → entrambi PASS. + +- [ ] **Step 6: Aggiungere test della transizione finestra (warmup → min_days → target_days)** + +Aggiungi: + +```python +def _ramp(n: int, base: Decimal = Decimal("1")) -> list[Decimal]: + """Ramp lineare 1, 2, 3, ... per testare in modo predicibile il P25.""" + return [base * Decimal(i + 1) for i in range(n)] + + +def test_below_min_days_uses_full_history() -> None: + # 5 giorni di storia (5*96=480 tick), min_days=30, target=60. + # Window = full history. + history = _ramp(480) + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # P25 di ramp [1..480] = ~120.75 (k=479*0.25=119.75, ramp[119]=120, ramp[120]=121) + assert out is not None + assert Decimal("120") <= out <= Decimal("121") + + +def test_between_min_and_target_uses_min_window() -> None: + # 50 giorni di storia (4800 tick), min_days=30, target=60. Window = ultimi 30g. + history = _ramp(4800) # values 1..4800 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # Window = ultimi 30*96=2880, valori 1921..4800, P25 ≈ 2640 + assert out is not None + assert Decimal("2630") <= out <= Decimal("2650") + + +def test_above_target_uses_target_window() -> None: + # 100 giorni (9600 tick), target=60. Window = ultimi 60g. + history = _ramp(9600) # values 1..9600 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # Window = ultimi 5760, valori 3841..9600, P25 ≈ 5280 + assert out is not None + assert Decimal("5270") <= out <= Decimal("5290") +``` + +Run: `pytest tests/unit/test_adaptive_threshold.py -v` → tutti PASS. + +- [ ] **Step 7: Aggiungere test floor binding/non-binding** + +Aggiungi: + +```python +def test_floor_binding_overrides_low_percentile() -> None: + history = [Decimal("0.5")] * 200 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("3"), + min_days=30, + target_days=60, + ) + assert out == Decimal("3") + + +def test_floor_not_binding_returns_percentile() -> None: + history = [Decimal("5")] * 200 + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.25"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + assert out == Decimal("5") + + +def test_median_percentile_returns_p50() -> None: + history = _ramp(200) + out = compute_adaptive_threshold( + history=history, + percentile=Decimal("0.5"), + absolute_floor=Decimal("0"), + min_days=30, + target_days=60, + ) + # P50 di [1..200] = (200+1)/2 = 100.5 + assert out is not None + assert Decimal("100") <= out <= Decimal("101") +``` + +Run: `pytest tests/unit/test_adaptive_threshold.py -v` → tutti PASS. + +- [ ] **Step 8: Commit** + +```bash +git -C /opt/docker/cerbero-bite add src/cerbero_bite/core/adaptive_threshold.py tests/unit/test_adaptive_threshold.py +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +feat(core): compute_adaptive_threshold pure function + tests + +Implementa il calcolo del percentile rolling con warmup, +transizione min_days → target_days e floor assoluto. Pure +function senza I/O: il caller passa la sequenza pre-filtrata +(NULL e fetch_ok=0 esclusi). + +Tests: warmup, transizione finestra, floor, percentili. + +Co-Authored-By: Claude Opus 4.7 (1M context) +EOF +)" +``` + +--- + +## Task 3: Estendere `EntryContext` con `iv_rv_history` e `dvol_24h_ago` + +**Files:** +- Modify: `src/cerbero_bite/core/entry_validator.py:28-51` + +- [ ] **Step 1: Aggiungere i campi a `EntryContext`** + +Modifica `src/cerbero_bite/core/entry_validator.py:28-51` aggiungendo dopo il campo `iv_minus_rv`: + +```python + # IV richness gate (§2.9). Differenza IV30g − RV30g in punti vol. + # Optional, stessa logica best-effort dei filtri quant: ``None`` + # significa "dato non disponibile" e fa saltare il gate (non + # invalida l'entry). + iv_minus_rv: Decimal | None = None + + # Storia recente di IV-RV (un valore per ogni tick di + # market_snapshots, ASC, NULL e fetch_ok=0 esclusi). Caricata dal + # repository in `entry_cycle` quando `iv_minus_rv_adaptive_enabled` + # è True. Tuple per coerenza con frozen=True. Vuoto = warmup hard + # (gate disabilitato). + iv_rv_history: tuple[Decimal, ...] = () + + # DVOL al tick più vicino a now - vol_of_vol_lookback_hours. + # ``None`` = gap nel dato (es. cron mancante 24h fa) → VoV guard + # skip. Caricato dal repository in `entry_cycle` quando + # `vol_of_vol_guard_enabled` è True. + dvol_24h_ago: Decimal | None = None +``` + +- [ ] **Step 2: Verificare backwards compat (test esistenti continuano a passare)** + +Run: `cd /opt/docker/cerbero-bite && pytest tests/unit/test_entry_validator.py -v 2>&1 | tail -30` + +Expected: tutti PASS (i campi nuovi hanno default vuoti). + +- [ ] **Step 3: Commit** + +```bash +git -C /opt/docker/cerbero-bite add src/cerbero_bite/core/entry_validator.py +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +feat(core): EntryContext aggiunge iv_rv_history e dvol_24h_ago + +Campi opzionali con default vuoto/None per non rompere i caller +esistenti. Saranno popolati da entry_cycle quando i flag +adaptive_enabled / vol_of_vol_guard_enabled sono True. + +Co-Authored-By: Claude Opus 4.7 (1M context) +EOF +)" +``` + +--- + +## Task 4: Wire del gate adattivo in `validate_entry` — TDD + +**Files:** +- Modify: `src/cerbero_bite/core/entry_validator.py:140-152` +- Modify: `tests/unit/test_entry_validator.py` (aggiunte additive) + +- [ ] **Step 1: Scrivere test "adaptive enabled, IV-RV sopra P25 → PASS"** + +Aggiungi a `tests/unit/test_entry_validator.py`: + +```python +# --------------------------------------------------------------------------- +# IV-RV adaptive gate +# --------------------------------------------------------------------------- + + +def _adaptive_cfg(**entry_overrides: object) -> StrategyConfig: + """Golden config con gate adattivo abilitato di default per test.""" + base_entry: dict[str, object] = { + "iv_minus_rv_filter_enabled": True, + "iv_minus_rv_adaptive_enabled": True, + "iv_minus_rv_min": Decimal("0"), + "iv_minus_rv_percentile": Decimal("0.25"), + "iv_minus_rv_window_target_days": 60, + "iv_minus_rv_window_min_days": 30, + } + base_entry.update(entry_overrides) + return golden_config(entry=base_entry) + + +def test_adaptive_pass_when_iv_rv_above_p25() -> None: + cfg = _adaptive_cfg() + # 200 tick, P25 ≈ 50, ctx.iv_minus_rv = 80 → PASS + history = tuple(Decimal(i) for i in range(1, 201)) + decision = validate_entry( + _good_ctx(iv_minus_rv=Decimal("80"), iv_rv_history=history), cfg + ) + assert decision.accepted is True + assert not any("IV richness" in r for r in decision.reasons) +``` + +- [ ] **Step 2: Eseguire e vedere fallire (logica adaptive non esiste)** + +Run: `pytest tests/unit/test_entry_validator.py::test_adaptive_pass_when_iv_rv_above_p25 -v` + +Expected: PASS già adesso (perché il gate vecchio con `iv_minus_rv_min=0` accetta tutto). Questo test verifica che la combinazione non rompa nulla, ma non distingue ancora il path adattivo. Il test discriminante è il prossimo. + +- [ ] **Step 3: Test discriminante "adaptive enabled, IV-RV sotto P25 → SKIP"** + +Aggiungi: + +```python +def test_adaptive_blocks_when_iv_rv_below_p25() -> None: + cfg = _adaptive_cfg() + # 200 tick, P25 ≈ 50, ctx.iv_minus_rv = 20 → SKIP + history = tuple(Decimal(i) for i in range(1, 201)) + decision = validate_entry( + _good_ctx(iv_minus_rv=Decimal("20"), iv_rv_history=history), cfg + ) + assert decision.accepted is False + assert any("IV richness" in r and "rolling" in r for r in decision.reasons) +``` + +Run: `pytest tests/unit/test_entry_validator.py::test_adaptive_blocks_when_iv_rv_below_p25 -v` + +Expected: FAIL — il path adattivo non esiste ancora; oggi con `iv_minus_rv_min=0` il gate vecchio non blocca, e la reason "rolling" non viene generata. + +- [ ] **Step 4: Implementare il path adattivo nel `validate_entry`** + +Sostituisci il blocco IV richness in `src/cerbero_bite/core/entry_validator.py:140-152` con: + +```python + # §2.9: IV richness gate. Vendere vol senza un margine misurabile + # fra IV e RV è statisticamente neutro: l'edge della strategia + # esiste solo quando il premio è "ricco" rispetto a quanto il + # mercato si è effettivamente mosso. La modalità adattiva calcola + # la soglia come max(P_q rolling, iv_minus_rv_min) sulla storia + # disponibile in market_snapshots; altrimenti fallback alla + # soglia statica `iv_minus_rv_min`. + if entry_cfg.iv_minus_rv_filter_enabled and ctx.iv_minus_rv is not None: + if entry_cfg.iv_minus_rv_adaptive_enabled: + from cerbero_bite.core.adaptive_threshold import ( + compute_adaptive_threshold, + ) + + threshold = compute_adaptive_threshold( + history=ctx.iv_rv_history, + percentile=entry_cfg.iv_minus_rv_percentile, + absolute_floor=entry_cfg.iv_minus_rv_min, + min_days=entry_cfg.iv_minus_rv_window_min_days, + target_days=entry_cfg.iv_minus_rv_window_target_days, + ) + if threshold is not None and ctx.iv_minus_rv < threshold: + reasons.append( + f"IV richness below P{int(entry_cfg.iv_minus_rv_percentile*100)} " + f"rolling (IV-RV={ctx.iv_minus_rv} < {threshold} vol pts)" + ) + elif ctx.iv_minus_rv < entry_cfg.iv_minus_rv_min: + reasons.append( + f"IV richness below floor " + f"(IV-RV={ctx.iv_minus_rv} < {entry_cfg.iv_minus_rv_min} vol pts)" + ) +``` + +- [ ] **Step 5: Eseguire i due test, devono passare** + +Run: `pytest tests/unit/test_entry_validator.py::test_adaptive_pass_when_iv_rv_above_p25 tests/unit/test_entry_validator.py::test_adaptive_blocks_when_iv_rv_below_p25 -v` + +Expected: 2 PASS. + +- [ ] **Step 6: Test warmup hard "history vuota, gate enabled → PASS (no skip)"** + +Aggiungi: + +```python +def test_adaptive_with_empty_history_passes_warmup() -> None: + cfg = _adaptive_cfg() + decision = validate_entry( + _good_ctx(iv_minus_rv=Decimal("0.1"), iv_rv_history=()), cfg + ) + assert decision.accepted is True + + +def test_adaptive_with_floor_floor_binds_when_p25_low() -> None: + cfg = _adaptive_cfg(iv_minus_rv_min=Decimal("3")) + history = tuple(Decimal("0.5") for _ in range(200)) # P25 = 0.5 + # ctx.iv_minus_rv = 1 → 1 < max(0.5, 3) = 3 → SKIP + decision = validate_entry( + _good_ctx(iv_minus_rv=Decimal("1"), iv_rv_history=history), cfg + ) + assert decision.accepted is False + assert any("IV richness" in r for r in decision.reasons) +``` + +Run: `pytest tests/unit/test_entry_validator.py -v -k "adaptive" 2>&1 | tail -20` + +Expected: tutti PASS. + +- [ ] **Step 7: Test backwards compat "adaptive disabled, gate statico funziona ancora"** + +Aggiungi: + +```python +def test_legacy_static_gate_still_works_when_adaptive_disabled() -> None: + cfg = golden_config(entry={ + "iv_minus_rv_filter_enabled": True, + "iv_minus_rv_adaptive_enabled": False, + "iv_minus_rv_min": Decimal("3"), + }) + decision = validate_entry( + _good_ctx(iv_minus_rv=Decimal("2"), iv_rv_history=()), cfg + ) + assert decision.accepted is False + assert any("IV richness below floor" in r for r in decision.reasons) + + +def test_iv_minus_rv_none_skips_gate_in_both_modes() -> None: + cfg = _adaptive_cfg() + decision = validate_entry( + _good_ctx(iv_minus_rv=None, iv_rv_history=tuple(Decimal(i) for i in range(1, 201))), + cfg, + ) + assert decision.accepted is True +``` + +Run: `pytest tests/unit/test_entry_validator.py -v 2>&1 | tail -30` + +Expected: tutti PASS. + +- [ ] **Step 8: Commit** + +```bash +git -C /opt/docker/cerbero-bite add src/cerbero_bite/core/entry_validator.py tests/unit/test_entry_validator.py +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +feat(core): IV-RV adaptive gate in validate_entry + tests + +Quando iv_minus_rv_adaptive_enabled=True, la soglia diventa +max(P_q rolling, iv_minus_rv_min). Path legacy (statico) e +None-bypass restano invariati. + +Tests: pass/skip su rolling, warmup hard, floor binding, +backwards compat statico, None bypass. + +Co-Authored-By: Claude Opus 4.7 (1M context) +EOF +)" +``` + +--- + +## Task 5: Wire del Vol-of-Vol guard in `validate_entry` — TDD + +**Files:** +- Modify: `src/cerbero_bite/core/entry_validator.py` (dopo il blocco IV richness) +- Modify: `tests/unit/test_entry_validator.py` + +- [ ] **Step 1: Scrivere test discriminanti per il VoV guard** + +Aggiungi a `tests/unit/test_entry_validator.py`: + +```python +# --------------------------------------------------------------------------- +# Vol-of-Vol guard +# --------------------------------------------------------------------------- + + +def _vov_cfg(threshold: Decimal = Decimal("5")) -> StrategyConfig: + return golden_config(entry={ + "vol_of_vol_guard_enabled": True, + "vol_of_vol_threshold_pt": threshold, + "vol_of_vol_lookback_hours": 24, + }) + + +def test_vov_guard_blocks_on_large_dvol_shift() -> None: + cfg = _vov_cfg() + decision = validate_entry( + _good_ctx(dvol_now=Decimal("56"), dvol_24h_ago=Decimal("50")), cfg + ) + assert decision.accepted is False + assert any("DVOL shifted" in r for r in decision.reasons) + + +def test_vov_guard_passes_on_small_dvol_shift() -> None: + cfg = _vov_cfg() + decision = validate_entry( + _good_ctx(dvol_now=Decimal("52"), dvol_24h_ago=Decimal("50")), cfg + ) + assert decision.accepted is True + + +def test_vov_guard_passes_when_lookback_missing() -> None: + """fail-open su gap dati: se dvol_24h_ago=None il guard non scatta.""" + cfg = _vov_cfg() + decision = validate_entry( + _good_ctx(dvol_now=Decimal("99"), dvol_24h_ago=None), cfg + ) + assert decision.accepted is True + + +def test_vov_guard_disabled_does_nothing() -> None: + cfg = golden_config(entry={"vol_of_vol_guard_enabled": False}) + decision = validate_entry( + _good_ctx(dvol_now=Decimal("99"), dvol_24h_ago=Decimal("50")), cfg + ) + assert decision.accepted is True +``` + +Run: `pytest tests/unit/test_entry_validator.py -k "vov" -v` + +Expected: 1 fail (test_vov_guard_blocks_on_large_dvol_shift) — la logica non esiste ancora. + +- [ ] **Step 2: Implementare il VoV guard in `validate_entry`** + +Aggiungi DOPO il blocco IV richness in `src/cerbero_bite/core/entry_validator.py` (subito prima del `return EntryDecision(...)`): + +```python + # §4-quater roadmap: vol-of-vol guard. Blocca entry quando il + # regime di volatilità sta cambiando bruscamente, anche se IV-RV + # è alto. Fail-open su gap dati 24h fa. + if ( + entry_cfg.vol_of_vol_guard_enabled + and ctx.dvol_24h_ago is not None + ): + delta = abs(ctx.dvol_now - ctx.dvol_24h_ago) + if delta >= entry_cfg.vol_of_vol_threshold_pt: + reasons.append( + f"DVOL shifted {delta} pt in {entry_cfg.vol_of_vol_lookback_hours}h " + f"(threshold {entry_cfg.vol_of_vol_threshold_pt})" + ) +``` + +- [ ] **Step 3: Eseguire i test VoV, devono passare** + +Run: `pytest tests/unit/test_entry_validator.py -k "vov" -v` + +Expected: 4 PASS. + +- [ ] **Step 4: Eseguire l'intera suite per non-regressione** + +Run: `pytest tests/unit/ -x -q 2>&1 | tail -15` + +Expected: tutti PASS. + +- [ ] **Step 5: Commit** + +```bash +git -C /opt/docker/cerbero-bite add src/cerbero_bite/core/entry_validator.py tests/unit/test_entry_validator.py +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +feat(core): Vol-of-Vol guard in validate_entry + tests + +Blocca entry se |DVOL_now - DVOL_24h_ago| >= threshold (default +5 pt). Fail-open quando dvol_24h_ago è None (gap dati). Independente +dal gate IV-RV: i due gate sono additivi. + +Co-Authored-By: Claude Opus 4.7 (1M context) +EOF +)" +``` + +--- + +## Task 6: Repository — `iv_rv_history` e `dvol_lookback` + +**Files:** +- Modify: `src/cerbero_bite/state/repository.py` (aggiungere metodi nella sezione `market_snapshots`) +- Create: `tests/unit/test_repository_iv_rv_helpers.py` + +- [ ] **Step 1: Scrivere il test fixture-based per `iv_rv_history`** + +Crea `tests/unit/test_repository_iv_rv_helpers.py`: + +```python +"""TDD per Repository.iv_rv_history e Repository.dvol_lookback.""" + +from __future__ import annotations + +import sqlite3 +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 + + +@pytest.fixture +def db_with_history(tmp_path) -> sqlite3.Connection: + """SQLite temp con 96 tick ETH a 15min ciascuno (1 giorno) e fetch_ok=1.""" + db_path = tmp_path / "test.sqlite" + conn = connect(str(db_path)) + run_migrations(conn) + + repo = Repository() + base = datetime(2026, 5, 1, 0, 0, tzinfo=UTC) + for i in range(96): + repo.record_market_snapshot( + conn, + MarketSnapshotRecord( + timestamp=base + timedelta(minutes=15 * i), + asset="ETH", + spot=Decimal("2000"), + dvol=Decimal("50") + Decimal(i) / Decimal("10"), + realized_vol_30d=Decimal("48"), + iv_minus_rv=Decimal("2") + Decimal(i) / Decimal("100"), + 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() + return conn + + +def test_iv_rv_history_returns_ordered_asc(db_with_history) -> None: + repo = Repository() + history = repo.iv_rv_history(db_with_history, asset="ETH", max_days=60) + assert len(history) == 96 + # ordered ASC (più vecchio → più recente) + assert history == sorted(history) + # primo valore è iv_minus_rv del tick t0 (i=0): 2.00 + assert history[0] == Decimal("2.00") + + +def test_iv_rv_history_filters_other_asset(db_with_history) -> None: + repo = Repository() + history = repo.iv_rv_history(db_with_history, asset="BTC", max_days=60) + assert history == [] + + +def test_iv_rv_history_skips_null_values(db_with_history, tmp_path) -> None: + repo = Repository() + # Inserisci una riga con iv_minus_rv NULL + repo.record_market_snapshot( + db_with_history, + MarketSnapshotRecord( + timestamp=datetime(2026, 5, 2, 0, 0, tzinfo=UTC), + asset="ETH", + spot=Decimal("2000"), + dvol=Decimal("50"), + realized_vol_30d=None, + iv_minus_rv=None, + 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, + ), + ) + db_with_history.commit() + history = repo.iv_rv_history(db_with_history, asset="ETH", max_days=60) + # Ancora 96 tick, NULL escluso + assert len(history) == 96 + + +def test_iv_rv_history_skips_fetch_failed(db_with_history) -> None: + repo = Repository() + # Tick con fetch_ok=False (anche se ha iv_minus_rv valido) + repo.record_market_snapshot( + db_with_history, + MarketSnapshotRecord( + timestamp=datetime(2026, 5, 3, 0, 0, tzinfo=UTC), + asset="ETH", + spot=Decimal("2000"), + dvol=Decimal("50"), + realized_vol_30d=None, + iv_minus_rv=Decimal("99"), + funding_perp_annualized=Decimal("0"), + funding_cross_annualized=Decimal("0"), + dealer_net_gamma=None, + gamma_flip_level=None, + oi_delta_pct_4h=None, + liquidation_long_risk=None, + liquidation_short_risk=None, + macro_days_to_event=None, + fetch_ok=False, + fetch_errors_json='{"x":"y"}', + ), + ) + db_with_history.commit() + history = repo.iv_rv_history(db_with_history, asset="ETH", max_days=60) + # Il valore 99 non deve essere in history + assert Decimal("99") not in history + + +def test_dvol_lookback_returns_closest_tick(db_with_history) -> None: + repo = Repository() + base = datetime(2026, 5, 1, 0, 0, tzinfo=UTC) + # Cerca il dvol di base + 12h. Tick attesi tutti 15min, t=base+12h + # corrisponde a i=48 → dvol = 50 + 4.8 = 54.8 + target = base + timedelta(hours=12) + out = repo.dvol_lookback( + db_with_history, asset="ETH", reference=target, tolerance_minutes=15 + ) + assert out == Decimal("54.8") + + +def test_dvol_lookback_returns_none_when_gap(db_with_history) -> None: + repo = Repository() + # Cerca un timestamp molto fuori range (1 anno prima) + target = datetime(2025, 1, 1, 0, 0, tzinfo=UTC) + out = repo.dvol_lookback( + db_with_history, asset="ETH", reference=target, tolerance_minutes=15 + ) + assert out is None +``` + +- [ ] **Step 2: Eseguire i test e vederli fallire (metodi non esistono)** + +Run: `cd /opt/docker/cerbero-bite && pytest tests/unit/test_repository_iv_rv_helpers.py -v 2>&1 | tail -20` + +Expected: ERRORS con `AttributeError: 'Repository' object has no attribute 'iv_rv_history'`. + +- [ ] **Step 3: Implementare i due metodi nel Repository** + +Aggiungi a `src/cerbero_bite/state/repository.py` nella sezione `market_snapshots` (subito dopo `list_market_snapshots`): + +```python + def iv_rv_history( + self, + conn: sqlite3.Connection, + *, + asset: str, + max_days: int, + ) -> list[Decimal]: + """Lista IV-RV ordinata ASC sull'intervallo `[now-max_days, now]`. + + Esclude righe con ``fetch_ok=0`` o ``iv_minus_rv IS NULL``. + Usata dal validator quando il gate adattivo è abilitato. + """ + rows = conn.execute( + "SELECT iv_minus_rv FROM market_snapshots " + "WHERE asset = ? " + " AND fetch_ok = 1 " + " AND iv_minus_rv IS NOT NULL " + " AND timestamp >= datetime('now', ?) " + "ORDER BY timestamp ASC", + (asset, f"-{int(max_days)} days"), + ).fetchall() + return [Decimal(str(r["iv_minus_rv"])) for r in rows] + + def dvol_lookback( + self, + conn: sqlite3.Connection, + *, + asset: str, + reference: datetime, + tolerance_minutes: int = 15, + ) -> Decimal | None: + """DVOL al tick più vicino a `reference`, entro ±tolerance_minutes. + + Ritorna ``None`` se non esiste un tick valido (``fetch_ok=1``, + ``dvol IS NOT NULL``) entro la tolerance. Usato dal Vol-of-Vol + guard per stimare DVOL N ore fa. + """ + ref_lo = (reference - timedelta(minutes=tolerance_minutes)).astimezone(UTC) + ref_hi = (reference + timedelta(minutes=tolerance_minutes)).astimezone(UTC) + row = conn.execute( + "SELECT dvol, timestamp FROM market_snapshots " + "WHERE asset = ? " + " AND fetch_ok = 1 " + " AND dvol IS NOT NULL " + " AND timestamp >= ? " + " AND timestamp <= ? " + "ORDER BY ABS(julianday(timestamp) - julianday(?)) ASC LIMIT 1", + ( + asset, + _enc_dt(ref_lo), + _enc_dt(ref_hi), + _enc_dt(reference), + ), + ).fetchone() + if row is None: + return None + return Decimal(str(row["dvol"])) +``` + +Aggiorna l'import in cima al file aggiungendo `timedelta`: + +```python +from datetime import UTC, datetime, timedelta +``` + +- [ ] **Step 4: Eseguire i test, devono passare** + +Run: `pytest tests/unit/test_repository_iv_rv_helpers.py -v 2>&1 | tail -20` + +Expected: tutti PASS (6 test). + +- [ ] **Step 5: Commit** + +```bash +git -C /opt/docker/cerbero-bite add src/cerbero_bite/state/repository.py tests/unit/test_repository_iv_rv_helpers.py +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +feat(state): Repository.iv_rv_history + dvol_lookback per gate adaptive + +Due nuovi metodi che leggono market_snapshots filtrando NULL e +fetch_ok=0. iv_rv_history limita a max_days; dvol_lookback trova +il tick più vicino a un istante con tolerance configurabile. + +Tests: ordered ASC, asset filter, NULL skip, fetch_ok=0 skip, +lookback closest, gap returns None. + +Co-Authored-By: Claude Opus 4.7 (1M context) +EOF +)" +``` + +--- + +## Task 7: Wire `entry_cycle` per popolare `iv_rv_history` e `dvol_24h_ago` + +**Files:** +- Modify: `src/cerbero_bite/runtime/entry_cycle.py:432-462` + +- [ ] **Step 1: Caricare history+lookback prima della costruzione di `EntryContext`** + +Modifica `src/cerbero_bite/runtime/entry_cycle.py` sostituendo il blocco riga 432 e seguenti. Cerca la linea `# 2. Entry filters` e sostituisci fino alla chiusura di `entry_ctx = EntryContext(...)` con: + +```python + # 2. Entry filters + entry_cfg = cfg.entry + asset = cfg.asset.symbol + + iv_rv_history: tuple[Decimal, ...] = () + dvol_24h_ago: Decimal | None = None + if entry_cfg.iv_minus_rv_filter_enabled and entry_cfg.iv_minus_rv_adaptive_enabled: + with ctx.repo.connect() as conn: + iv_rv_history = tuple( + ctx.repo.iv_rv_history( + conn, + asset=asset, + max_days=entry_cfg.iv_minus_rv_window_target_days, + ) + ) + if entry_cfg.vol_of_vol_guard_enabled: + with ctx.repo.connect() as conn: + dvol_24h_ago = ctx.repo.dvol_lookback( + conn, + asset=asset, + reference=when - timedelta(hours=entry_cfg.vol_of_vol_lookback_hours), + ) + + entry_ctx = EntryContext( + capital_usd=capital_usd, + dvol_now=snap.dvol, + funding_perp_annualized=snap.funding_perp, + eth_holdings_pct_of_portfolio=snap.eth_holdings_pct, + next_macro_event_in_days=snap.macro_days_to_event, + has_open_position=False, + dealer_net_gamma=snap.dealer_net_gamma, + iv_minus_rv=snap.iv_minus_rv, + liquidation_squeeze_risk_high=snap.liquidation_squeeze_risk_high, + iv_rv_history=iv_rv_history, + dvol_24h_ago=dvol_24h_ago, + ) +``` + +Verifica che `timedelta` sia già importato in cima al file. Se non c'è, aggiungilo all'import datetime. + +**Nota:** la firma di `ctx.repo.connect()` deve già supportare il context manager. Se in `state/db.py` la `Repository` non ha `.connect()`, l'attributo è in `ctx.repo` di tipo Repository nuda; in quel caso usa il pattern già usato nel resto del file (`ctx.repo.list_market_snapshots(conn, ...)` con `conn` ottenuto da `ctx.connect_db()` o equivalente). Cerca il pattern già in uso nello stesso `entry_cycle.py` con `grep -n "with " src/cerbero_bite/runtime/entry_cycle.py | head` e copia esattamente lo stile. + +- [ ] **Step 2: Aggiornare il `inputs_json` del decisions log per includere i nuovi valori** + +Modifica `src/cerbero_bite/runtime/entry_cycle.py:444-462`. Sostituisci la chiusura del dict `inputs` con: + +```python + inputs = { + "snapshot": { + "spot_eth_usd": str(snap.spot_eth_usd), + "spot_eth_30d_ago": ( + str(snap.spot_eth_30d_ago) if snap.spot_eth_30d_ago else None + ), + "adx_14": str(snap.adx_14) if snap.adx_14 is not None else None, + "dvol": str(snap.dvol), + "funding_perp": str(snap.funding_perp), + "funding_cross": str(snap.funding_cross), + "macro_days_to_event": snap.macro_days_to_event, + "eth_holdings_pct": str(snap.eth_holdings_pct), + "portfolio_eur": str(snap.portfolio_eur), + "capital_usd": str(capital_usd), + "iv_minus_rv": ( + str(snap.iv_minus_rv) if snap.iv_minus_rv is not None else None + ), + "iv_rv_history_n": len(iv_rv_history), + "dvol_24h_ago": ( + str(dvol_24h_ago) if dvol_24h_ago is not None else None + ), + } + } +``` + +- [ ] **Step 3: Eseguire i test integration esistenti per non-regressione** + +Run: `cd /opt/docker/cerbero-bite && pytest tests/integration/test_entry_cycle.py -x -q 2>&1 | tail -20` + +Expected: tutti PASS. Se qualcuno fallisce per via del nuovo accesso a `ctx.repo`, verifica il pattern di connessione corretto e correggi. + +- [ ] **Step 4: Commit** + +```bash +git -C /opt/docker/cerbero-bite add src/cerbero_bite/runtime/entry_cycle.py +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +feat(runtime): entry_cycle popola iv_rv_history e dvol_24h_ago + +Quando i flag adaptive_enabled / vol_of_vol_guard_enabled sono +attivi, entry_cycle carica history e lookback dal repository +prima di costruire EntryContext. Il decisions log riceve i meta +n_history e dvol_24h_ago per audit ex-post. + +Quando i flag sono off, niente query DB extra (zero overhead). + +Co-Authored-By: Claude Opus 4.7 (1M context) +EOF +)" +``` + +--- + +## Task 8: Integration test end-to-end + +**Files:** +- Create: `tests/integration/test_entry_cycle_iv_rv_adaptive.py` + +- [ ] **Step 1: Scrivere il test integration con SQLite reale e fixture seedata** + +Crea `tests/integration/test_entry_cycle_iv_rv_adaptive.py`. Usa `tests/integration/test_entry_cycle.py` come template di stile: + +```python +"""End-to-end test del gate IV-RV adattivo + Vol-of-Vol guard via entry_cycle.""" + +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) + 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") # P25 di [1.0]*1440 + [5.0]*1440 = 1.0 + + +def test_dvol_lookback_within_tolerance(db_30d) -> None: + conn, repo = db_30d + base = datetime(2026, 4, 1, 0, 0, tzinfo=UTC) + # Cerca DVOL a t = base + 24h (dovrebbe trovare un tick a quella ora) + 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 + # 1 anno prima → fuori range + out = repo.dvol_lookback( + conn, + asset="ETH", + reference=datetime(2025, 1, 1, tzinfo=UTC), + tolerance_minutes=15, + ) + assert out is None +``` + +- [ ] **Step 2: Eseguire i test integration** + +Run: `cd /opt/docker/cerbero-bite && pytest tests/integration/test_entry_cycle_iv_rv_adaptive.py -v 2>&1 | tail -20` + +Expected: 3 PASS. + +- [ ] **Step 3: Commit** + +```bash +git -C /opt/docker/cerbero-bite add tests/integration/test_entry_cycle_iv_rv_adaptive.py +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +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) +EOF +)" +``` + +--- + +## Task 9: Aggiornare `strategy.aggressiva.yaml` con i nuovi flag + +**Files:** +- Modify: `strategy.aggressiva.yaml:69-70` (e dintorni) + +- [ ] **Step 1: Aggiornare il blocco `entry:` nel profilo Aggressiva** + +Modifica `strategy.aggressiva.yaml`. Trova il blocco intorno a riga 69 e sostituisci: + +```yaml + iv_minus_rv_min: "3" + iv_minus_rv_filter_enabled: true +``` + +con: + +```yaml + # IV richness gate (§2.9). In Aggressiva il gate è in modalità + # adattiva: la soglia è il P25 rolling sui market_snapshots + # (warmup: usa la storia disponibile finché < 30g, poi finestra 30g + # fino a 60g, poi fissa 60g). `iv_minus_rv_min: 0` = floor zero, + # lascia decidere al P25. + iv_minus_rv_filter_enabled: true + iv_minus_rv_adaptive_enabled: true + iv_minus_rv_min: "0" + iv_minus_rv_percentile: "0.25" + iv_minus_rv_window_target_days: 60 + iv_minus_rv_window_min_days: 30 + + # Vol-of-Vol guard: blocca entry su shift bruschi DVOL. + vol_of_vol_guard_enabled: true + vol_of_vol_threshold_pt: "5" + vol_of_vol_lookback_hours: 24 +``` + +- [ ] **Step 2: Verificare parsing del YAML** + +Run: `cd /opt/docker/cerbero-bite && python -c " +import yaml +from cerbero_bite.config.schema import StrategyConfig +data = yaml.safe_load(open('strategy.aggressiva.yaml')) +cfg = StrategyConfig(**data) +print('OK', cfg.entry.iv_minus_rv_adaptive_enabled, cfg.entry.iv_minus_rv_percentile, cfg.entry.vol_of_vol_guard_enabled) +"` + +Expected: `OK True 0.25 True` + +- [ ] **Step 3: Verifica che `strategy.yaml` (golden) e `strategy.conservativa.yaml` continuino a parse-are con i default disabled** + +Run: `for f in strategy.yaml strategy.conservativa.yaml; do cd /opt/docker/cerbero-bite && python -c " +import yaml +from cerbero_bite.config.schema import StrategyConfig +data = yaml.safe_load(open('$f')) +cfg = StrategyConfig(**data) +print('$f', cfg.entry.iv_minus_rv_adaptive_enabled, cfg.entry.vol_of_vol_guard_enabled) +"; done` + +Expected: entrambi → `False False` (i nuovi flag hanno default disabled, quindi non serve modificare i due yaml). + +- [ ] **Step 4: Commit** + +```bash +git -C /opt/docker/cerbero-bite add strategy.aggressiva.yaml +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +feat(config): profilo Aggressiva attiva gate IV-RV adattivo + VoV + +P25 rolling 60g, warmup a finestra disponibile, VoV guard 5pt. +Conservativa e golden invariati (default disabled). + +Co-Authored-By: Claude Opus 4.7 (1M context) +EOF +)" +``` + +--- + +## Task 10: GUI Calibrazione — pannello "Gate adattivo" + +**Files:** +- Modify: `src/cerbero_bite/gui/pages/6_📐_Calibrazione.py` + +- [ ] **Step 1: Identificare il punto di inserimento e leggere la struttura attuale della pagina** + +Run: `head -80 "/opt/docker/cerbero-bite/src/cerbero_bite/gui/pages/6_📐_Calibrazione.py"` + +Cerca dove la pagina carica `market_snapshots` e calcola percentili. Il pannello "Gate adattivo" va inserito SOPRA la sezione percentili statici esistenti. + +- [ ] **Step 2: Aggiungere import e funzione di rendering del pannello** + +Aggiungi in cima al file (dopo gli import esistenti, accanto agli altri import del progetto): + +```python +from datetime import UTC, datetime, timedelta +from decimal import Decimal + +from cerbero_bite.config.schema import StrategyConfig +from cerbero_bite.core.adaptive_threshold import compute_adaptive_threshold +``` + +Aggiungi una funzione di rendering vicino agli altri helper (cerca `def _render_` o `def render_` esistenti per posizionamento coerente): + +```python +def _render_adaptive_gate_panel( + cfg: StrategyConfig, + iv_rv_history: list[Decimal], + iv_rv_now: Decimal | None, + dvol_now: Decimal | None, + dvol_24h_ago: Decimal | None, +) -> None: + """Pannello informativo sul gate IV-RV adattivo (read-only).""" + import streamlit as st + + st.subheader("🎯 Gate IV-RV adattivo") + + entry = cfg.entry + if not entry.iv_minus_rv_filter_enabled: + st.info("Gate IV-RV disabilitato nel profilo corrente.") + return + + n_ticks = len(iv_rv_history) + n_days = n_ticks // 96 + target = entry.iv_minus_rv_window_target_days + min_days = entry.iv_minus_rv_window_min_days + if n_days < 1: + status = f"🟡 Warmup hard ({n_ticks}/96 tick)" + elif n_days < min_days: + status = f"🟡 Warmup ({n_days}/{min_days}g — finestra crescente)" + elif n_days < target: + status = f"🟢 Attivo (finestra {min_days}g, target {target}g)" + else: + status = f"🟢 Attivo (finestra stabile {target}g)" + st.markdown(f"**Status:** {status}") + + if entry.iv_minus_rv_adaptive_enabled: + threshold = compute_adaptive_threshold( + history=iv_rv_history, + percentile=entry.iv_minus_rv_percentile, + absolute_floor=entry.iv_minus_rv_min, + min_days=min_days, + target_days=target, + ) + c1, c2, c3 = st.columns(3) + c1.metric( + f"Soglia P{int(entry.iv_minus_rv_percentile*100)} rolling", + f"{threshold:.2f}" if threshold is not None else "—", + help="Soglia adattiva = max(percentile, floor)", + ) + c2.metric( + "IV-RV ultimo tick", + f"{iv_rv_now:.2f}" if iv_rv_now is not None else "—", + ) + c3.metric("Floor assoluto", f"{entry.iv_minus_rv_min:.2f}") + + if threshold is not None and iv_rv_now is not None: + verdict = "✅ PASS" if iv_rv_now >= threshold else "❌ SKIP" + st.markdown(f"**Decisione hypothetical:** {verdict}") + else: + st.write(f"Modalità statica: floor = {entry.iv_minus_rv_min} vol pts") + + if entry.vol_of_vol_guard_enabled: + st.markdown("---") + st.markdown("**Vol-of-Vol guard**") + if dvol_now is not None and dvol_24h_ago is not None: + delta = abs(dvol_now - dvol_24h_ago) + c1, c2 = st.columns(2) + c1.metric( + f"|ΔDVOL {entry.vol_of_vol_lookback_hours}h|", + f"{delta:.2f}", + ) + c2.metric("Soglia VoV", f"{entry.vol_of_vol_threshold_pt:.2f}") + verdict = "✅ PASS" if delta < entry.vol_of_vol_threshold_pt else "❌ SKIP" + st.markdown(f"**Verdict:** {verdict}") + else: + st.info("Lookback 24h non disponibile (gap dati).") +``` + +- [ ] **Step 3: Cablare il pannello nella pagina principale** + +Cerca nella pagina dove vengono caricati `market_snapshots` (es. `repo.list_market_snapshots(...)`) e dove inizia la sezione "percentili statici". Subito sopra la sezione percentili, aggiungi: + +```python + # Pannello gate adattivo (informativo, read-only) + iv_rv_history_list = [ + s.iv_minus_rv for s in snapshots + if s.iv_minus_rv is not None and s.fetch_ok + ] + iv_rv_history_list.reverse() # snapshots è DESC, history serve ASC + + iv_rv_now = ( + snapshots[0].iv_minus_rv + if snapshots and snapshots[0].iv_minus_rv is not None + else None + ) + dvol_now = ( + snapshots[0].dvol + if snapshots and snapshots[0].dvol is not None + else None + ) + # Per dvol_24h_ago serve fare un'altra query: usa direttamente + # repo.dvol_lookback se la pagina ha una conn; altrimenti + # passa None (il pannello mostra "gap dati"). + dvol_24h_ago = None + if cfg.entry.vol_of_vol_guard_enabled and snapshots: + ref = snapshots[0].timestamp - timedelta( + hours=cfg.entry.vol_of_vol_lookback_hours + ) + # Cerca il tick più vicino a ref nello stesso `snapshots` con + # tolerance 15min, evita un'altra query. + for s in snapshots: + if abs((s.timestamp - ref).total_seconds()) <= 900 and s.dvol is not None: + dvol_24h_ago = s.dvol + break + + _render_adaptive_gate_panel( + cfg=cfg, + iv_rv_history=iv_rv_history_list, + iv_rv_now=iv_rv_now, + dvol_now=dvol_now, + dvol_24h_ago=dvol_24h_ago, + ) +``` + +**Nota:** se la pagina non ha ancora `cfg: StrategyConfig` in scope, caricalo con il pattern già in uso nel progetto. Cerca con `grep -n "StrategyConfig\|load_config" src/cerbero_bite/gui/pages/6_📐_Calibrazione.py`. + +- [ ] **Step 4: Smoke test manuale (richiede deploy)** + +Run: `cd /opt/docker/cerbero-bite && docker compose build cerbero-bite-gui && docker compose up -d cerbero-bite-gui` + +Aspetta che la GUI riavvii (~30s). Apri il browser sulla pagina Calibrazione. Verifica che: +- Il pannello "Gate IV-RV adattivo" appare in cima alla pagina +- Mostra status warmup (n_days < 30 con i dati attuali) +- Soglia, IV-RV, floor e verdict hypothetical sono visualizzati +- Sezione VoV guard mostra ΔDVOL e verdict + +Se la pagina rompe → controlla i log container con `docker logs cerbero-bite-cerbero-bite-gui-1 --tail 50` e correggi import o accesso a `cfg`. + +- [ ] **Step 5: Commit** + +```bash +git -C /opt/docker/cerbero-bite add "src/cerbero_bite/gui/pages/6_📐_Calibrazione.py" +git -C /opt/docker/cerbero-bite commit -m "$(cat <<'EOF' +feat(gui): pannello informativo Gate IV-RV adattivo in Calibrazione + +Mostra status (warmup/attivo), soglia P25 rolling corrente, IV-RV +ultimo tick, floor assoluto, decisione hypothetical e sezione +Vol-of-Vol guard. Read-only: i percentili statici esistenti +restano per analisi. + +Co-Authored-By: Claude Opus 4.7 (1M context) +EOF +)" +``` + +--- + +## Task 11: Suite finale + deploy + +- [ ] **Step 1: Eseguire tutta la suite di test** + +Run: `cd /opt/docker/cerbero-bite && pytest -x -q 2>&1 | tail -30` + +Expected: tutti PASS. + +- [ ] **Step 2: Verificare lint/type** + +Run: `cd /opt/docker/cerbero-bite && ruff check src/ tests/ && mypy src/cerbero_bite/core/adaptive_threshold.py src/cerbero_bite/core/entry_validator.py 2>&1 | tail -20` + +Expected: zero errori critici. Se ci sono warning non bloccanti, ignora. + +- [ ] **Step 3: Build e deploy del container engine + GUI** + +Run: `cd /opt/docker/cerbero-bite && docker compose build && docker compose up -d` + +Aspetta health check (~60s). + +- [ ] **Step 4: Verificare che il container engine non logghi errori al boot** + +Run: `docker logs cerbero-bite-cerbero-bite-1 --tail 50 2>&1 | grep -iE "error|traceback" | head -20` + +Expected: nessun output (no error). Se c'è errore, debug. + +- [ ] **Step 5: Verificare che nuovo `decisions` log includa i meta** + +Aspetta il primo entry cycle (cron giornaliero 14:00 UTC su strategy.yaml; per testare prima usa il comando `cerbero-bite cycle entry` se disponibile). + +Run: `docker exec cerbero-bite-cerbero-bite-1 sqlite3 /app/data/state.sqlite "SELECT inputs_json FROM decisions ORDER BY id DESC LIMIT 1;"` + +Verifica che `inputs_json` contenga `iv_rv_history_n` e `dvol_24h_ago`. + +--- + +## Spec Coverage Self-Review + +| Spec section | Implemented in | +|---|---| +| §3 Approccio Hybrid | Task 4 (P25) + Task 5 (VoV) | +| §4 Comportamento gate | Task 4, Task 5 | +| §4.1 Warmup behavior | Task 2 (`compute_adaptive_threshold` warmup branch) | +| §4.2 Floor max(P25,floor) | Task 2 (test_floor_binding/non_binding), Task 4 | +| §5 Schema config | Task 1 | +| §5.1 Profili predefiniti | Task 9 (Aggressiva); golden/Conservativa già coperti dai default | +| §5.2 Backwards compat | Task 4 step 7 (test_legacy_static_gate_still_works) | +| §6.1 `core/adaptive_threshold.py` | Task 2 | +| §6.2 Repository methods | Task 6 | +| §6.3 Inline nel validator | Task 4, Task 5 | +| §6.4 Audit/logging | Task 7 step 2 (`iv_rv_history_n`, `dvol_24h_ago` in inputs_json) | +| §7 GUI Calibrazione | Task 10 | +| §8 Error handling fail-open | Task 4 (`iv_minus_rv=None` skip), Task 5 (`dvol_24h_ago=None` skip) | +| §9.1 Unit `core/adaptive_threshold` | Task 2 | +| §9.2 Unit `core/entry_validator` | Task 4, Task 5 | +| §9.3 Integration | Task 8 | +| §9.4 Backtest sanity | Out of plan (mantenuto per follow-up — il backtest CLI è una feature separata, fuori dallo scope minimale) | +| §9.5 GUI smoke | Task 10 step 4 | +| §10 Out of scope | rispettato — niente regime detection, override manuale, multi-asset | + +--- + +## Execution Handoff + +**Plan complete and saved to `docs/superpowers/plans/2026-05-08-iv-rv-adaptive-gate.md`. Two execution options:** + +**1. Subagent-Driven (recommended)** — Dispatch un subagent fresco per ogni task, review tra un task e l'altro, iterazione rapida. + +**2. Inline Execution** — Eseguo i task in questa sessione con `executing-plans`, batch con checkpoint per review. + +**Quale preferisci?**