# 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?**