Adriano/Cerbero-Bite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
080acf829d87767bd17f4f4da1654be38449c1ca
Cerbero-Bite/docs/superpowers/plans/2026-05-08-iv-rv-adaptive-gate.md
T
root f889258952 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) <noreply@anthropic.com>
2026-05-08 19:58:32 +00:00

1597 lines
56 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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?**
Reference in New Issue View Git Blame Copy Permalink