From 3a5cf2554bbf2525bcdd8713f02c1656bd7c5af8 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 8 May 2026 22:29:48 +0000 Subject: [PATCH] feat(core): IV-RV adaptive gate in validate_entry + tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. Aggiunge anche due model_validator a StrategyConfig per fail-fast su config invalida (window_min_days < target_days, percentile in (0,1)) — risponde alla code review T1. Tests: pass/skip su rolling, warmup hard, floor binding, backwards compat statico, None bypass. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/cerbero_bite/config/schema.py | 10 +++ src/cerbero_bite/core/entry_validator.py | 38 +++++++++--- tests/unit/test_entry_validator.py | 79 ++++++++++++++++++++++++ 3 files changed, 117 insertions(+), 10 deletions(-) diff --git a/src/cerbero_bite/config/schema.py b/src/cerbero_bite/config/schema.py index 0b066e7..015e303 100644 --- a/src/cerbero_bite/config/schema.py +++ b/src/cerbero_bite/config/schema.py @@ -407,6 +407,16 @@ class StrategyConfig(BaseModel): if self.entry.dvol_min >= self.entry.dvol_max: raise ValueError("dvol_min must be < dvol_max") + e = self.entry + if e.iv_minus_rv_window_min_days >= e.iv_minus_rv_window_target_days: + raise ValueError( + "iv_minus_rv_window_min_days must be < iv_minus_rv_window_target_days" + ) + if not (Decimal("0") < e.iv_minus_rv_percentile < Decimal("1")): + raise ValueError( + "iv_minus_rv_percentile must be in (0, 1)" + ) + return self diff --git a/src/cerbero_bite/core/entry_validator.py b/src/cerbero_bite/core/entry_validator.py index 4ab5e5c..70e8866 100644 --- a/src/cerbero_bite/core/entry_validator.py +++ b/src/cerbero_bite/core/entry_validator.py @@ -153,16 +153,34 @@ def validate_entry(ctx: EntryContext, cfg: StrategyConfig) -> EntryDecision: # §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. - if ( - entry_cfg.iv_minus_rv_filter_enabled - and ctx.iv_minus_rv is not None - and 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)" - ) + # 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: + pct = int(entry_cfg.iv_minus_rv_percentile * 100) + reasons.append( + f"IV richness below P{pct} rolling " + f"(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)" + ) return EntryDecision(accepted=not reasons, reasons=reasons) diff --git a/tests/unit/test_entry_validator.py b/tests/unit/test_entry_validator.py index 106c1f3..939e2e4 100644 --- a/tests/unit/test_entry_validator.py +++ b/tests/unit/test_entry_validator.py @@ -351,3 +351,82 @@ def test_bias_zero_division_safe(cfg: StrategyConfig) -> None: # eth_30d_ago == 0 must not crash; treat as no-bias (neutral) ctx = _trend(eth_now="3000", eth_30d_ago="0", funding_cross="0", dvol="60", adx="15") assert compute_bias(ctx, cfg) is None + + +# --------------------------------------------------------------------------- +# 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() + 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) + + +def test_adaptive_blocks_when_iv_rv_below_p25() -> None: + cfg = _adaptive_cfg() + 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) + + +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)) + 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) + + +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