"""TDD per :mod:`cerbero_bite.core.backtest`.""" from __future__ import annotations from datetime import UTC, datetime, timedelta from decimal import Decimal import pytest from cerbero_bite.config import StrategyConfig, golden_config from cerbero_bite.core.backtest import ( bs_put_delta, bs_put_price, daily_picks, estimate_credit_eth, find_strike_for_delta, normal_cdf, run_backtest, simulate_entry_filters, ) from cerbero_bite.state.models import MarketSnapshotRecord # --------------------------------------------------------------------------- # Black-Scholes helpers # --------------------------------------------------------------------------- def test_normal_cdf_known_values() -> None: assert normal_cdf(0.0) == pytest.approx(0.5, abs=1e-6) assert normal_cdf(1.0) == pytest.approx(0.8413, abs=1e-3) assert normal_cdf(-1.0) == pytest.approx(0.1587, abs=1e-3) assert normal_cdf(2.0) == pytest.approx(0.9772, abs=1e-3) def test_bs_put_price_atm_positive_and_less_than_strike() -> None: p = bs_put_price(spot=3000, strike=3000, t_years=18 / 365, sigma=0.50) assert p > 0 assert p < 3000 # cap def test_bs_put_price_far_otm_close_to_zero() -> None: p = bs_put_price(spot=3000, strike=1500, t_years=18 / 365, sigma=0.50) assert 0 <= p < 5 # essentially zero def test_bs_put_delta_atm_around_minus_half() -> None: d = bs_put_delta(spot=3000, strike=3000, t_years=18 / 365, sigma=0.50) assert d == pytest.approx(-0.475, abs=0.05) def test_bs_put_delta_far_otm_close_to_zero() -> None: d = bs_put_delta(spot=3000, strike=1500, t_years=18 / 365, sigma=0.50) assert -0.05 < d <= 0 def test_find_strike_for_delta_monotone() -> None: spot = 3000.0 dvol = 50.0 dte = 18 s_010 = find_strike_for_delta( spot=spot, dvol_pct=dvol, dte_days=dte, target_delta_abs=0.10, ) s_020 = find_strike_for_delta( spot=spot, dvol_pct=dvol, dte_days=dte, target_delta_abs=0.20, ) # |Δ|=0.20 (più ITM) ⇒ strike più alto di |Δ|=0.10 (più OTM). assert s_020 > s_010 # Verifica che il delta corrisponda a target ± tolleranza. achieved = abs( bs_put_delta( spot=spot, strike=s_020, t_years=dte / 365, sigma=dvol / 100, ) ) assert achieved == pytest.approx(0.20, abs=0.02) def test_estimate_credit_returns_positive_credit_in_normal_regime() -> None: credit_eth, short_k, long_k = estimate_credit_eth( spot=3000, dvol_pct=50, dte_days=18, width_pct=0.04, delta_target_abs=0.12, ) # Sanity: credit > 0, short_k < spot, long_k = short_k - 4%×spot assert credit_eth > 0 assert short_k < 3000 assert long_k < short_k assert short_k - long_k == pytest.approx(0.04 * 3000, abs=1.0) # --------------------------------------------------------------------------- # Daily picks + entry filter simulation # --------------------------------------------------------------------------- def _snap( *, ts: datetime, spot: float = 3000, dvol: float = 50, funding: float = 0.0, macro_d: int | None = None, asset: str = "ETH", ) -> MarketSnapshotRecord: return MarketSnapshotRecord( timestamp=ts, asset=asset, spot=Decimal(str(spot)), dvol=Decimal(str(dvol)), funding_perp_annualized=Decimal(str(funding)), funding_cross_annualized=Decimal("0"), dealer_net_gamma=Decimal("100"), liquidation_long_risk="low", liquidation_short_risk="low", macro_days_to_event=macro_d, fetch_ok=True, ) def test_daily_picks_extracts_one_per_calendar_day() -> None: day1 = datetime(2026, 5, 4, 14, 0, tzinfo=UTC) day2 = datetime(2026, 5, 5, 14, 0, tzinfo=UTC) # Tuesday: PICKED ora (crypto 24/7) snapshots = [ _snap(ts=day1), _snap(ts=day1 + timedelta(minutes=15)), # stesso giorno, deduplicato _snap(ts=day2), ] picks = daily_picks(snapshots) assert len(picks) == 2 assert picks[0].timestamp == day1 assert picks[1].timestamp == day2 def test_daily_picks_skips_other_hours() -> None: snapshots = [ _snap(ts=datetime(2026, 5, 4, 13, 0, tzinfo=UTC)), # 13:00 → skipped _snap(ts=datetime(2026, 5, 5, 15, 30, tzinfo=UTC)), # 15:30 → skipped ] assert daily_picks(snapshots) == [] def test_daily_picks_filters_by_asset() -> None: day = datetime(2026, 5, 4, 14, 0, tzinfo=UTC) snapshots = [ _snap(ts=day, asset="BTC"), _snap(ts=day, asset="ETH"), ] picks = daily_picks(snapshots, asset="ETH") assert len(picks) == 1 assert picks[0].snapshot.asset == "ETH" def test_simulate_entry_filters_accepts_clean_snapshot( ) -> None: cfg: StrategyConfig = golden_config() monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC) snap = _snap(ts=monday, dvol=50, funding=0.10) picks = [ type("MP", (), {"timestamp": monday, "snapshot": snap})() # type: ignore[arg-type] ] # Hack: build via real dataclass from cerbero_bite.core.backtest import DailyPick picks = [DailyPick(timestamp=monday, snapshot=snap)] results = simulate_entry_filters(picks, cfg, capital_usd=Decimal("1500")) assert len(results) == 1 assert results[0].accepted is True def test_simulate_entry_filters_rejects_dvol_out_of_band() -> None: cfg = golden_config() monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC) snap = _snap(ts=monday, dvol=20, funding=0.10) # below 35 from cerbero_bite.core.backtest import DailyPick picks = [DailyPick(timestamp=monday, snapshot=snap)] results = simulate_entry_filters(picks, cfg, capital_usd=Decimal("1500")) assert results[0].accepted is False assert any("dvol" in r.lower() for r in results[0].reasons) def test_simulate_entry_filters_skips_incomplete_snapshot() -> None: cfg = golden_config() monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC) incomplete = MarketSnapshotRecord( timestamp=monday, asset="ETH", spot=Decimal("3000"), # dvol=None ⇒ skipped fetch_ok=False, ) from cerbero_bite.core.backtest import DailyPick picks = [DailyPick(timestamp=monday, snapshot=incomplete)] results = simulate_entry_filters(picks, cfg, capital_usd=Decimal("1500")) assert results[0].accepted is False assert results[0].skipped_for_data is True # --------------------------------------------------------------------------- # Full pipeline (sintetico) # --------------------------------------------------------------------------- def _synthetic_year_of_snapshots( *, n_weeks: int = 8, spot: float = 3000, dvol: float = 60, # con skew_premium 1.5 ⇒ credit/width ≈ 35% (sopra soglia 30%) funding: float = 0.10, ) -> list[MarketSnapshotRecord]: """Genera N settimane di snapshot sintetici ETH a 4 tick/settimana.""" rows: list[MarketSnapshotRecord] = [] monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC) for week in range(n_weeks): base = monday + timedelta(weeks=week) # Lunedì 14:00 è il pick rows.append(_snap(ts=base, spot=spot, dvol=dvol, funding=funding)) # Tick intermedi che NON cadono alle 14:00: # offset +1h (=15:00) così vengono ignorati da `daily_picks`. for d in (2, 8, 14, 19): rows.append( _snap( ts=base + timedelta(days=d, hours=1), spot=spot * (1 + 0.005 * d), # +0.5% al giorno dvol=dvol - 1.5 * d, # vol che scende lentamente funding=funding, ) ) return rows def test_run_backtest_produces_report_with_trades() -> None: # Per il test scaliamo il credit/width gate al 15%: il modello BS # senza skew completo sottostima i premi OTM rispetto al reale. # Vedi `estimate_credit_eth.skew_premium` docstring per dettagli. from cerbero_bite.config.schema import StructureConfig cfg = golden_config() cfg = cfg.model_copy( update={ "structure": StructureConfig( **{ **cfg.structure.model_dump(), "credit_to_width_ratio_min": Decimal("0.15"), } ) } ) snapshots = _synthetic_year_of_snapshots(n_weeks=4) report = run_backtest(snapshots, cfg, capital_usd=Decimal("1500")) # Sanity: 4 picks, almeno 1 trade chiuso assert report.n_picks == 4 assert report.n_completed >= 1 assert report.cumulative_pnl_usd != Decimal("0") # Bull-put + ETH al rialzo + DVOL che scende ⇒ atteso win assert report.n_winners >= 1 def test_run_backtest_handles_no_picks_gracefully() -> None: cfg = golden_config() # Solo tick infrasettimanali, niente Monday 14:00. monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC) snapshots = [_snap(ts=monday + timedelta(hours=1))] report = run_backtest(snapshots, cfg, capital_usd=Decimal("1500")) assert report.n_picks == 0 assert report.n_completed == 0 assert report.cumulative_pnl_usd == Decimal("0")