Files

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

56 KiB

Raw Blame History

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:

    # 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

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:

"""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:

"""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:

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:

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:

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

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:

    # 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

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:

# ---------------------------------------------------------------------------
# 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:

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:

    # §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:

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:

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

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:

# ---------------------------------------------------------------------------
# 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(...)):

    # §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

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:

"""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):

    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:

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

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:

    # 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:

    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

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:

"""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

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:

  iv_minus_rv_min: "3"
  iv_minus_rv_filter_enabled: true

con:

  # 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

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):

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):

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:

    # 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

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?

56 KiB Raw Blame History Unescape Escape

IV-RV Adaptive Gate Implementation Plan

File Structure

Task 1: Aggiungere campi config in EntryConfig

Task 2: Pure function compute_adaptive_threshold — TDD

Task 3: Estendere EntryContext con iv_rv_history e dvol_24h_ago

Task 4: Wire del gate adattivo in validate_entry — TDD

Task 5: Wire del Vol-of-Vol guard in validate_entry — TDD

Task 6: Repository — iv_rv_history e dvol_lookback

Task 7: Wire entry_cycle per popolare iv_rv_history e dvol_24h_ago

Task 8: Integration test end-to-end

Task 9: Aggiornare strategy.aggressiva.yaml con i nuovi flag

Task 10: GUI Calibrazione — pannello "Gate adattivo"

Task 11: Suite finale + deploy

Spec Coverage Self-Review

Execution Handoff

56 KiB

Raw Blame History

Task 1: Aggiungere campi config in `EntryConfig`

Task 2: Pure function `compute_adaptive_threshold` — TDD

Task 3: Estendere `EntryContext` con `iv_rv_history` e `dvol_24h_ago`

Task 4: Wire del gate adattivo in `validate_entry` — TDD

Task 5: Wire del Vol-of-Vol guard in `validate_entry` — TDD

Task 6: Repository — `iv_rv_history` e `dvol_lookback`

Task 7: Wire `entry_cycle` per popolare `iv_rv_history` e `dvol_24h_ago`

Task 9: Aggiornare `strategy.aggressiva.yaml` con i nuovi flag