"""TDD for :mod:`cerbero_bite.core.exit_decision`. Spec: ``docs/01-strategy-rules.md §7`` and ``docs/03-algorithms.md §6``. The eight mandatory cases listed in §6 are exercised here, with the ``mark=30% credit`` case re-interpreted to satisfy the time-stop exception consistently with rule 1 (which always fires earlier when mark ≤ 50% credit). """ from __future__ import annotations from datetime import UTC, datetime, timedelta from decimal import Decimal from uuid import uuid4 import pytest from cerbero_bite.config import SpreadType, StrategyConfig, golden_config from cerbero_bite.core.exit_decision import PositionSnapshot, evaluate from cerbero_bite.core.types import OptionLeg @pytest.fixture def cfg() -> StrategyConfig: return golden_config() def _legs(spread_type: SpreadType = "bull_put", size: int = 1) -> list[OptionLeg]: expiry = datetime(2026, 5, 15, 8, 0, tzinfo=UTC) if spread_type == "bull_put": return [ OptionLeg( instrument="ETH-X-2475-P", side="SELL", strike=Decimal("2475"), expiry=expiry, type="P", size=size, mid_price_eth=Decimal("0.020"), delta=Decimal("-0.12"), gamma=Decimal("0.001"), theta=Decimal("-0.0005"), vega=Decimal("0.10"), ), OptionLeg( instrument="ETH-X-2350-P", side="BUY", strike=Decimal("2350"), expiry=expiry, type="P", size=size, mid_price_eth=Decimal("0.005"), delta=Decimal("-0.08"), gamma=Decimal("0.001"), theta=Decimal("-0.0003"), vega=Decimal("0.07"), ), ] return [ OptionLeg( instrument="ETH-X-3525-C", side="SELL", strike=Decimal("3525"), expiry=expiry, type="C", size=size, mid_price_eth=Decimal("0.020"), delta=Decimal("0.12"), gamma=Decimal("0.001"), theta=Decimal("-0.0005"), vega=Decimal("0.10"), ), OptionLeg( instrument="ETH-X-3645-C", side="BUY", strike=Decimal("3645"), expiry=expiry, type="C", size=size, mid_price_eth=Decimal("0.005"), delta=Decimal("0.08"), gamma=Decimal("0.001"), theta=Decimal("-0.0003"), vega=Decimal("0.07"), ), ] def _snapshot( *, spread_type: SpreadType = "bull_put", credit_received_eth: str = "0.030", mark_combo_now_eth: str = "0.020", dvol_at_entry: str = "50", dvol_now: str = "50", delta_short_now: str | None = None, return_4h_now: str = "0", days_to_expiry: int = 14, spot_at_entry: str = "3000", spot_now: str = "3000", eth_price_usd_now: str = "3000", ) -> PositionSnapshot: now = datetime(2026, 4, 27, 14, 0, tzinfo=UTC) expiry = now + timedelta(days=days_to_expiry) if delta_short_now is None: delta_short_now = "-0.12" if spread_type == "bull_put" else "0.12" legs = _legs(spread_type=spread_type) return PositionSnapshot( proposal_id=uuid4(), spread_type=spread_type, legs=legs, credit_received_eth=Decimal(credit_received_eth), credit_received_usd=Decimal(credit_received_eth) * Decimal(eth_price_usd_now), spot_at_entry=Decimal(spot_at_entry), dvol_at_entry=Decimal(dvol_at_entry), expiry=expiry, opened_at=now - timedelta(days=4), eth_price_usd_now=Decimal(eth_price_usd_now), spot_now=Decimal(spot_now), dvol_now=Decimal(dvol_now), mark_combo_now_eth=Decimal(mark_combo_now_eth), delta_short_now=Decimal(delta_short_now), return_4h_now=Decimal(return_4h_now), now=now, ) # --------------------------------------------------------------------------- # Mandatory cases from docs/03-algorithms.md §6 # --------------------------------------------------------------------------- def test_mark_at_50pct_credit_triggers_close_profit(cfg: StrategyConfig) -> None: snap = _snapshot(credit_received_eth="0.030", mark_combo_now_eth="0.015") res = evaluate(snap, cfg) assert res.action == "CLOSE_PROFIT" def test_mark_at_250pct_credit_triggers_close_stop(cfg: StrategyConfig) -> None: snap = _snapshot(credit_received_eth="0.030", mark_combo_now_eth="0.075") res = evaluate(snap, cfg) assert res.action == "CLOSE_STOP" def test_dvol_increase_above_band_triggers_close_vol(cfg: StrategyConfig) -> None: snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.020", # 67% credit, no profit/stop dvol_at_entry="50", dvol_now="62", # +12, above +10 trigger ) res = evaluate(snap, cfg) assert res.action == "CLOSE_VOL" def test_six_days_left_mark_above_skip_threshold_triggers_close_time( cfg: StrategyConfig, ) -> None: # mark = 80% credit > 70% credit (skip threshold) → CLOSE_TIME snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.024", days_to_expiry=6, ) res = evaluate(snap, cfg) assert res.action == "CLOSE_TIME" def test_six_days_left_mark_in_skip_zone_holds(cfg: StrategyConfig) -> None: # mark = 60% credit ∈ (50%, 70%] → close to profit, time stop is skipped. snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.018", days_to_expiry=6, ) res = evaluate(snap, cfg) assert res.action == "HOLD" def test_short_delta_breach_triggers_close_delta(cfg: StrategyConfig) -> None: snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.020", delta_short_now="-0.32", ) res = evaluate(snap, cfg) assert res.action == "CLOSE_DELTA" def test_4h_adverse_move_bull_put_triggers_close_averse(cfg: StrategyConfig) -> None: snap = _snapshot( spread_type="bull_put", credit_received_eth="0.030", mark_combo_now_eth="0.020", return_4h_now="-0.07", ) res = evaluate(snap, cfg) assert res.action == "CLOSE_AVERSE" def test_4h_adverse_move_bear_call_triggers_close_averse(cfg: StrategyConfig) -> None: snap = _snapshot( spread_type="bear_call", credit_received_eth="0.030", mark_combo_now_eth="0.020", return_4h_now="0.07", ) res = evaluate(snap, cfg) assert res.action == "CLOSE_AVERSE" def test_neutral_state_returns_hold(cfg: StrategyConfig) -> None: snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.020", ) res = evaluate(snap, cfg) assert res.action == "HOLD" # --------------------------------------------------------------------------- # Trigger ordering — first match wins # --------------------------------------------------------------------------- def test_profit_wins_over_vol_stop(cfg: StrategyConfig) -> None: # mark = 30% credit AND DVOL +12 → profit takes precedence snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.009", dvol_at_entry="50", dvol_now="62", ) res = evaluate(snap, cfg) assert res.action == "CLOSE_PROFIT" def test_stop_wins_over_delta_breach(cfg: StrategyConfig) -> None: snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.075", # 250% → stop delta_short_now="-0.40", # would also be CLOSE_DELTA ) res = evaluate(snap, cfg) assert res.action == "CLOSE_STOP" # --------------------------------------------------------------------------- # PnL reporting # --------------------------------------------------------------------------- def test_pnl_in_eth_and_usd_reflect_credit_minus_debit(cfg: StrategyConfig) -> None: snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.018", eth_price_usd_now="3100", ) res = evaluate(snap, cfg) # pnl_eth = 0.030 - 0.018 = 0.012; pnl_usd = 0.012 × 3100 = 37.2 assert res.pnl_estimate_eth == Decimal("0.012") assert res.pnl_estimate_usd == Decimal("37.2") def test_iron_condor_adverse_move_either_direction(cfg: StrategyConfig) -> None: snap = _snapshot( spread_type="iron_condor", credit_received_eth="0.030", mark_combo_now_eth="0.020", return_4h_now="-0.06", ) res = evaluate(snap, cfg) assert res.action == "CLOSE_AVERSE" # --------------------------------------------------------------------------- # §7-bis (D): vol-collapse harvest # --------------------------------------------------------------------------- def _harvest_cfg( cfg: StrategyConfig, *, threshold: str = "15" ) -> StrategyConfig: """Clona la golden config con la soglia di vol-harvest abilitata.""" from cerbero_bite.config import ExitConfig return cfg.model_copy( update={ "exit": ExitConfig( **{ **cfg.exit.model_dump(), "vol_harvest_dvol_decrease": Decimal(threshold), } ) } ) def test_vol_harvest_disabled_by_default_does_not_fire(cfg: StrategyConfig) -> None: # Default: vol_harvest_dvol_decrease = 0 ⇒ filtro disabilitato. snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.022", # in profit (debit < credit) dvol_at_entry="60", dvol_now="40", # crollato di 20 punti ) res = evaluate(snap, cfg) assert res.action == "HOLD" def test_vol_harvest_fires_when_dvol_collapsed_in_profit( cfg: StrategyConfig, ) -> None: harvest = _harvest_cfg(cfg, threshold="15") snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.022", # in profit ma sopra profit_take 50% dvol_at_entry="60", dvol_now="42", # −18, supera la soglia 15 ) res = evaluate(snap, harvest) assert res.action == "CLOSE_VOL_HARVEST" assert "harvest" in res.reason def test_vol_harvest_does_not_fire_when_in_loss(cfg: StrategyConfig) -> None: # Anche se DVOL crolla, se siamo in perdita non vogliamo harvest: # è una funzione di "esci con il profitto in mano", non un panico. harvest = _harvest_cfg(cfg, threshold="15") snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.040", # debit > credit ⇒ in perdita dvol_at_entry="60", dvol_now="42", ) res = evaluate(snap, harvest) assert res.action != "CLOSE_VOL_HARVEST" def test_vol_harvest_does_not_fire_below_threshold(cfg: StrategyConfig) -> None: harvest = _harvest_cfg(cfg, threshold="15") snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.022", dvol_at_entry="60", dvol_now="50", # −10, sotto la soglia 15 ) res = evaluate(snap, harvest) assert res.action == "HOLD" def test_profit_take_wins_over_vol_harvest(cfg: StrategyConfig) -> None: # Quando il profit-take è già colpito, non passiamo per vol-harvest. harvest = _harvest_cfg(cfg, threshold="15") snap = _snapshot( credit_received_eth="0.030", mark_combo_now_eth="0.014", # ≤ 50% credit ⇒ profit-take dvol_at_entry="60", dvol_now="42", ) res = evaluate(snap, harvest) assert res.action == "CLOSE_PROFIT"