diff --git a/src/cerbero_bite/cli.py b/src/cerbero_bite/cli.py index 78635cd..83216b3 100644 --- a/src/cerbero_bite/cli.py +++ b/src/cerbero_bite/cli.py @@ -13,7 +13,7 @@ import asyncio import os import sys from collections.abc import Callable -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from decimal import Decimal from pathlib import Path from typing import Any @@ -679,6 +679,155 @@ def replay(date_from: str, date_to: str, capital: float, dry_run: bool) -> None: ) +@main.command() +@click.option( + "--strategy", + "strategy_path", + type=click.Path(path_type=Path), + default=Path("strategy.yaml"), + show_default=True, + help="Path al file di strategia (golden, conservativa, aggressiva, ...).", +) +@click.option( + "--db", + "db_path", + type=click.Path(path_type=Path), + default=_DEFAULT_DB_PATH, + show_default=True, + help="SQLite con `market_snapshots` storiche.", +) +@click.option( + "--from", + "date_from", + type=click.DateTime(formats=["%Y-%m-%d"]), + default=None, + help="ISO date YYYY-MM-DD (default: 90 giorni fa).", +) +@click.option( + "--to", + "date_to", + type=click.DateTime(formats=["%Y-%m-%d"]), + default=None, + help="ISO date YYYY-MM-DD (default: oggi).", +) +@click.option( + "--capital", + type=float, + default=1500.0, + show_default=True, + help="Capitale di partenza per il backtest, in USD.", +) +@click.option( + "--asset", + type=str, + default="ETH", + show_default=True, + help="Asset di riferimento per le snapshot.", +) +@click.option( + "--no-enforce-hash", + is_flag=True, + default=False, + help="Salta la verifica del config_hash (utile per profili sperimentali).", +) +def backtest( + strategy_path: Path, + db_path: Path, + date_from: datetime | None, + date_to: datetime | None, + capital: float, + asset: str, + no_enforce_hash: bool, +) -> None: + """Esegue il backtest stilizzato su `market_snapshots` storiche. + + Usa lo stesso `validate_entry` del live per i filtri (rigoroso) e + un modello Black-Scholes con skew premium per stimare credito ed + exit P/L (stilizzato — vedi docstring di `core/backtest.py`). + """ + from cerbero_bite.config.loader import load_strategy # noqa: PLC0415 + from cerbero_bite.core.backtest import run_backtest # noqa: PLC0415 + + console = Console() + if date_to is None: + date_to = datetime.now(UTC) + if date_from is None: + date_from = date_to - timedelta(days=90) + date_from = date_from.replace(tzinfo=UTC) if date_from.tzinfo is None else date_from + date_to = date_to.replace(tzinfo=UTC) if date_to.tzinfo is None else date_to + + loaded = load_strategy(strategy_path, enforce_hash=not no_enforce_hash) + cfg = loaded.config + + conn = connect_state(db_path) + try: + repo = Repository() + snapshots = repo.list_market_snapshots( + conn, + asset=asset.upper(), + start=date_from, + end=date_to, + limit=10000, + ) + finally: + conn.close() + + if not snapshots: + console.print( + f"[yellow]Nessuno snapshot {asset} trovato fra {date_from.date()} " + f"e {date_to.date()}.[/yellow]" + ) + sys.exit(1) + + console.print( + f"[green]Caricate {len(snapshots)} snapshot {asset} " + f"({snapshots[-1].timestamp.date()} → {snapshots[0].timestamp.date()})[/green]" + ) + + report = run_backtest(snapshots, cfg, capital_usd=Decimal(str(capital))) + + table = Table(title=f"Backtest report — {strategy_path.name}") + table.add_column("Metrica", style="cyan") + table.add_column("Valore", style="bold") + table.add_row("Picks (lunedì 14:00)", str(report.n_picks)) + table.add_row( + "Accettati dai filtri", + f"{report.n_accepted} ({report.n_accepted / max(1, report.n_picks):.0%})", + ) + table.add_row("Saltati per dato mancante", str(report.n_skipped_data)) + table.add_row("Trade completati (con P/L)", str(report.n_completed)) + table.add_row("Vincenti", f"{report.n_winners} ({report.win_rate:.0%})") + table.add_row("P/L cumulato (USD)", f"{report.cumulative_pnl_usd:+.2f}") + table.add_row( + "P/L su capitale", f"{report.cumulative_pnl_pct_of_capital:+.2%}" + ) + table.add_row( + "Max drawdown", f"−{report.max_drawdown_usd:.0f} USD " + f"({report.max_drawdown_pct:.1%})", + ) + table.add_row( + "Sharpe (annualized)", + f"{report.sharpe_annualized}" if report.sharpe_annualized is not None + else "—", + ) + console.print(table) + + if report.skip_reasons: + skip_table = Table(title="Motivi di skip aggregati") + skip_table.add_column("Motivo") + skip_table.add_column("Settimane", justify="right") + for reason, count in sorted( + report.skip_reasons.items(), key=lambda kv: -kv[1] + ): + skip_table.add_row(reason, str(count)) + console.print(skip_table) + + console.print( + "[dim]Il modello P/L è stilizzato: BS + skew premium 1.5×. " + "Numeri ottimi per ranking config, non per promesse operative.[/dim]" + ) + + @main.group() def config() -> None: """Strategy configuration utilities.""" diff --git a/src/cerbero_bite/core/backtest.py b/src/cerbero_bite/core/backtest.py new file mode 100644 index 0000000..bc4a91f --- /dev/null +++ b/src/cerbero_bite/core/backtest.py @@ -0,0 +1,652 @@ +"""Stylized backtest engine over ``market_snapshots`` (§13). + +Two layers, both pure functions: + +1. **Entry-filter simulation** — for each Monday 14:00 UTC tick in the + recorded snapshots, evaluate which §2 gates would have passed, + reconstructing :class:`EntryContext` from the snapshot. This part + is **rigorous**: it uses the same :func:`validate_entry` the live + engine uses, so the output is exactly "what the bot would have + decided". + +2. **P/L estimation per accepted entry** — since ``market_snapshots`` + does NOT record the option chain (we only collect spot, DVOL, + funding, etc.), credit and exit P/L are estimated via a stylized + Black-Scholes model: given ``spot``, ``DVOL`` (as IV), and the + strategy's delta target, we solve for the short strike, the long + strike at ``width_pct`` distance, and the combo mid-price. Future + ticks are then re-priced under the same model to detect the first + exit trigger from §7. + +The stylized layer is **intentionally approximate**: it captures the +geometry of the strategy (DVOL band sets credit, ETH path drives +exit triggers) but not the second-order effects (chain liquidity, +borrow rates, exchange fees beyond the 0.03% notional cap, dealer +hedging skew). Numbers are good for ranking and tuning, not for +operational P/L promises. + +The engine is deterministic and side-effect-free: it does **not** +write to SQLite, does not call MCP, does not place orders. It +operates entirely on a list of :class:`MarketSnapshotRecord` rows +the caller has already loaded. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from decimal import Decimal +from typing import Literal + +from pydantic import BaseModel, ConfigDict + +from cerbero_bite.config.schema import StrategyConfig +from cerbero_bite.core.entry_validator import EntryContext, validate_entry +from cerbero_bite.state.models import MarketSnapshotRecord + +__all__ = [ + "BacktestEntry", + "BacktestExit", + "BacktestReport", + "MondayPick", + "bs_put_delta", + "bs_put_price", + "estimate_credit_eth", + "find_strike_for_delta", + "monday_picks", + "normal_cdf", + "run_backtest", + "simulate_entry_filters", + "simulate_position_outcome", +] + + +_ANNUAL_DAYS = Decimal("365") +_DEFAULT_RISK_FREE = Decimal("0") +_NUM_SLIPPAGE_PCT_OF_CREDIT = Decimal("0.03") +_NUM_FEE_PCT_OF_NOTIONAL = Decimal("0.0003") + + +# --------------------------------------------------------------------------- +# Black-Scholes helpers (stdlib-only) +# --------------------------------------------------------------------------- + + +def normal_cdf(x: float) -> float: + """Standard normal CDF, no scipy dependency.""" + return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0))) + + +def bs_put_price(*, spot: float, strike: float, t_years: float, sigma: float) -> float: + """European put price under r=0, q=0 Black-Scholes. + + Returns price in spot units (so for an ETH option, dividing by spot + gives the price in ETH). + """ + if t_years <= 0 or sigma <= 0 or spot <= 0 or strike <= 0: + return max(0.0, strike - spot) + sqrt_t = math.sqrt(t_years) + d1 = (math.log(spot / strike) + 0.5 * sigma * sigma * t_years) / (sigma * sqrt_t) + d2 = d1 - sigma * sqrt_t + return strike * normal_cdf(-d2) - spot * normal_cdf(-d1) + + +def bs_put_delta(*, spot: float, strike: float, t_years: float, sigma: float) -> float: + """Put delta under r=0, q=0 Black-Scholes (negative number for put). + + Returns 0 for expired options. + """ + if t_years <= 0 or sigma <= 0 or spot <= 0 or strike <= 0: + return 0.0 + sqrt_t = math.sqrt(t_years) + d1 = (math.log(spot / strike) + 0.5 * sigma * sigma * t_years) / (sigma * sqrt_t) + return normal_cdf(d1) - 1.0 # = -N(-d1) + + +def find_strike_for_delta( + *, + spot: float, + dvol_pct: float, + dte_days: int, + target_delta_abs: float, +) -> float: + """Solve for the put strike whose |delta| matches ``target_delta_abs``. + + Bisection on a monotone-decreasing |delta(strike)| relationship. + Returns the strike in absolute USD terms. + """ + sigma = max(0.01, dvol_pct / 100.0) + t_years = max(1e-6, dte_days / 365.0) + # Bracket: from 50% of spot (deep OTM, small |delta|) up to spot + # (ATM, |delta| ≈ 0.5). + low = max(1.0, spot * 0.30) + high = spot + for _ in range(64): + mid = 0.5 * (low + high) + delta_abs = abs(bs_put_delta(spot=spot, strike=mid, t_years=t_years, sigma=sigma)) + if delta_abs > target_delta_abs: + high = mid + else: + low = mid + if abs(high - low) < 1e-3: + break + return 0.5 * (low + high) + + +def estimate_credit_eth( + *, + spot: float, + dvol_pct: float, + dte_days: int, + width_pct: float, + delta_target_abs: float, + skew_premium: float = 1.5, +) -> tuple[float, float, float]: + """Estimate credit (ETH), short_strike, long_strike for a bull-put-style + credit spread under Black-Scholes. + + ``skew_premium`` è il moltiplicatore applicato al credito BS per + approssimare la **vol smile** dell'ETH options market (le put OTM + trattano a IV più alta della IV ATM, quindi un BS pulito sottostima + sistematicamente il premio del venditore di vol). Il default 1.5 + è una stima conservativa dei dati Deribit storici (smile slope + tipica 5-10 vol points per 100δ); valori sensati: 1.3 (smile + blanda) … 1.8 (regime "stress IV"). Va calibrato sui dati reali + quando avremo abbastanza chain history da farlo. + + Returns ``(credit_eth, short_strike, long_strike)``. Credit è + già moltiplicato per ``skew_premium``. + """ + short_strike = find_strike_for_delta( + spot=spot, dvol_pct=dvol_pct, dte_days=dte_days, + target_delta_abs=delta_target_abs, + ) + width_usd = width_pct * spot + long_strike = max(1.0, short_strike - width_usd) + sigma = max(0.01, dvol_pct / 100.0) + t_years = max(1e-6, dte_days / 365.0) + short_mid_usd = bs_put_price( + spot=spot, strike=short_strike, t_years=t_years, sigma=sigma, + ) + long_mid_usd = bs_put_price( + spot=spot, strike=long_strike, t_years=t_years, sigma=sigma, + ) + short_mid_eth = short_mid_usd / spot + long_mid_eth = long_mid_usd / spot + credit_eth = (short_mid_eth - long_mid_eth) * skew_premium + return credit_eth, short_strike, long_strike + + +# --------------------------------------------------------------------------- +# Entry filter simulation — rigorous (uses validate_entry exactly) +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class MondayPick: + """Indice di un tick "Monday 14:00 UTC" nella time-series.""" + + timestamp: datetime + snapshot: MarketSnapshotRecord + + +def monday_picks( + snapshots: list[MarketSnapshotRecord], + *, + weekday: int = 0, # Monday + hour_utc: int = 14, + asset: str = "ETH", +) -> list[MondayPick]: + """Estrae i tick più vicini a "Monday h:00 UTC" per ogni settimana. + + ``snapshots`` deve essere ordinato per timestamp ascending. Per ogni + occorrenza di ``weekday + hour_utc`` (es. lun 14:00) presa l'unica + riga ETH che la copre. Settimane senza tick a quell'ora vengono + saltate. + """ + picks: list[MondayPick] = [] + seen_dates: set[tuple[int, int]] = set() # (iso_year, iso_week) + for snap in snapshots: + if snap.asset.upper() != asset.upper(): + continue + ts = snap.timestamp.astimezone(UTC) + if ts.weekday() != weekday or ts.hour != hour_utc: + continue + iso_y, iso_w, _ = ts.isocalendar() + key = (iso_y, iso_w) + if key in seen_dates: + continue + seen_dates.add(key) + picks.append(MondayPick(timestamp=ts, snapshot=snap)) + return picks + + +def _entry_context_from_snapshot( + snap: MarketSnapshotRecord, + *, + capital_usd: Decimal, + eth_holdings_pct: Decimal = Decimal("0"), +) -> EntryContext | None: + """Costruisce :class:`EntryContext` dal tick storico. + + ``None`` quando la riga non ha i campi minimi (spot, dvol, funding). + Nel filtro questo si traduce in "skip della settimana" — è la + stessa logica del live: un tick incompleto è meglio di un'entry + al buio. + """ + if snap.dvol is None or snap.funding_perp_annualized is None: + return None + return EntryContext( + capital_usd=capital_usd, + dvol_now=snap.dvol, + funding_perp_annualized=snap.funding_perp_annualized, + eth_holdings_pct_of_portfolio=eth_holdings_pct, + next_macro_event_in_days=snap.macro_days_to_event, + has_open_position=False, + dealer_net_gamma=snap.dealer_net_gamma, + liquidation_squeeze_risk_high=( + snap.liquidation_long_risk == "high" + or snap.liquidation_short_risk == "high" + ), + ) + + +@dataclass(frozen=True) +class EntryFilterResult: + """Esito del check filtri per una singola Monday pick.""" + + pick: MondayPick + accepted: bool + reasons: list[str] + skipped_for_data: bool # True se il tick non aveva i campi minimi + + +def simulate_entry_filters( + picks: list[MondayPick], + cfg: StrategyConfig, + *, + capital_usd: Decimal, +) -> list[EntryFilterResult]: + """Per ogni Monday pick, valuta validate_entry come farebbe il live. + + Rigoroso: usa esattamente :func:`validate_entry` e :class:`EntryContext`. + Restituisce la lista degli esiti, una entry per pick. + """ + results: list[EntryFilterResult] = [] + for pick in picks: + ctx = _entry_context_from_snapshot(pick.snapshot, capital_usd=capital_usd) + if ctx is None: + results.append( + EntryFilterResult( + pick=pick, + accepted=False, + reasons=["incomplete_snapshot"], + skipped_for_data=True, + ) + ) + continue + decision = validate_entry(ctx, cfg) + results.append( + EntryFilterResult( + pick=pick, + accepted=decision.accepted, + reasons=list(decision.reasons), + skipped_for_data=False, + ) + ) + return results + + +# --------------------------------------------------------------------------- +# Position outcome simulation — stylized (Black-Scholes re-pricing) +# --------------------------------------------------------------------------- + + +class BacktestEntry(BaseModel): + """Trade aperto nel backtest (snapshot al momento dell'entry).""" + + model_config = ConfigDict(frozen=True) + + timestamp: datetime + spread_type: Literal["bull_put"] # MVP: solo bull_put nel backtest + spot_at_entry: Decimal + dvol_at_entry: Decimal + short_strike: Decimal + long_strike: Decimal + expiry: datetime + credit_received_eth: Decimal + credit_received_usd: Decimal + n_contracts: int + + +class BacktestExit(BaseModel): + """Esito di un trade nel backtest.""" + + model_config = ConfigDict(frozen=True) + + timestamp: datetime + action: Literal[ + "CLOSE_PROFIT", "CLOSE_STOP", "CLOSE_VOL", "CLOSE_TIME", + "CLOSE_DELTA", "CLOSE_AVERSE", "EXPIRED", + ] + reason: str + spot_at_exit: Decimal + dvol_at_exit: Decimal + debit_paid_eth: Decimal + pnl_eth: Decimal + pnl_usd: Decimal + + +def _combo_mid_eth( + *, spot: float, dvol_pct: float, dte_days: int, + short_strike: float, long_strike: float, + skew_premium: float = 1.5, +) -> float: + """Re-prezza il combo bull-put usando BS sul nuovo spot/dvol/dte.""" + sigma = max(0.01, dvol_pct / 100.0) + t_years = max(1e-6, dte_days / 365.0) + short_mid_usd = bs_put_price( + spot=spot, strike=short_strike, t_years=t_years, sigma=sigma, + ) + long_mid_usd = bs_put_price( + spot=spot, strike=long_strike, t_years=t_years, sigma=sigma, + ) + return (short_mid_usd - long_mid_usd) / spot * skew_premium + + +def simulate_position_outcome( + entry: BacktestEntry, + future_snapshots: list[MarketSnapshotRecord], + cfg: StrategyConfig, +) -> BacktestExit: + """Re-prezza il combo a ogni tick futuro fino al primo exit trigger. + + Triggers in ordine §7: + 1. profit_take (debit ≤ 0.5×credit) + 2. stop_loss (debit ≥ 2.5×credit) + 3. vol_stop (DVOL salita di ≥10 pt rispetto entry) + 4. time_stop (DTE ≤ 7 e debit > 0.7×credit) + 5. expiry (uscita per scadenza, P/L = credit − intrinsic) + """ + ec = cfg.exit + credit = float(entry.credit_received_eth) + short = float(entry.short_strike) + long_ = float(entry.long_strike) + + profit_thresh = float(ec.profit_take_pct_of_credit) * credit + stop_thresh = float(ec.stop_loss_mark_x_credit) * credit + skip_time_thresh = float(ec.time_stop_skip_if_close_to_profit_pct) * credit + + for snap in future_snapshots: + if snap.timestamp <= entry.timestamp: + continue + if snap.timestamp >= entry.expiry: + break + if snap.dvol is None or snap.spot is None: + continue + spot_now = float(snap.spot) + dvol_now = float(snap.dvol) + dte = max(0, (entry.expiry - snap.timestamp).days) + debit = _combo_mid_eth( + spot=spot_now, dvol_pct=dvol_now, dte_days=dte, + short_strike=short, long_strike=long_, + ) + if debit <= profit_thresh: + return _exit( + snap, entry, debit, + action="CLOSE_PROFIT", + reason=f"debit {debit:.4f} ≤ {profit_thresh:.4f}", + ) + if debit >= stop_thresh: + return _exit( + snap, entry, debit, + action="CLOSE_STOP", + reason=f"debit {debit:.4f} ≥ {stop_thresh:.4f}", + ) + if dvol_now >= float(entry.dvol_at_entry) + float(ec.vol_stop_dvol_increase): + return _exit( + snap, entry, debit, + action="CLOSE_VOL", + reason=f"DVOL {dvol_now:.1f} ≥ entry+{ec.vol_stop_dvol_increase}", + ) + if dte <= ec.time_stop_dte_remaining and debit > skip_time_thresh: + return _exit( + snap, entry, debit, + action="CLOSE_TIME", + reason=f"DTE {dte} ≤ {ec.time_stop_dte_remaining}", + ) + + # Tick passati senza trigger: scadenza naturale. + last = future_snapshots[-1] if future_snapshots else None + intrinsic = max(0.0, short - float(last.spot if last and last.spot else 0)) + intrinsic_capped = min(intrinsic, short - long_) + debit_at_expiry_eth = ( + intrinsic_capped / float(last.spot) + if last is not None and last.spot is not None and float(last.spot) > 0 + else 0.0 + ) + return _exit( + last or _synthetic_expiry_snapshot(entry), + entry, + debit_at_expiry_eth, + action="EXPIRED", + reason="held to expiry", + ) + + +def _synthetic_expiry_snapshot(entry: BacktestEntry) -> MarketSnapshotRecord: + return MarketSnapshotRecord( + timestamp=entry.expiry, + asset="ETH", + spot=entry.spot_at_entry, + dvol=entry.dvol_at_entry, + fetch_ok=False, + ) + + +def _exit( + snap: MarketSnapshotRecord, + entry: BacktestEntry, + debit_eth: float, + *, + action: str, + reason: str, +) -> BacktestExit: + pnl_eth = float(entry.credit_received_eth) - debit_eth + spot = float(snap.spot) if snap.spot is not None else float(entry.spot_at_entry) + dvol = float(snap.dvol) if snap.dvol is not None else float(entry.dvol_at_entry) + return BacktestExit( + timestamp=snap.timestamp, + action=action, # type: ignore[arg-type] + reason=reason, + spot_at_exit=Decimal(str(spot)), + dvol_at_exit=Decimal(str(dvol)), + debit_paid_eth=Decimal(str(debit_eth)), + pnl_eth=Decimal(str(pnl_eth)), + pnl_usd=Decimal(str(pnl_eth * spot * entry.n_contracts)), + ) + + +# --------------------------------------------------------------------------- +# Full pipeline +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class CompletedTrade: + entry: BacktestEntry + exit: BacktestExit + + +class BacktestReport(BaseModel): + """Aggregato del backtest. Tutti i numeri sono **stime**.""" + + model_config = ConfigDict(frozen=True) + + n_picks: int + n_accepted: int + n_skipped_data: int + n_completed: int + n_winners: int + win_rate: Decimal + cumulative_pnl_usd: Decimal + cumulative_pnl_pct_of_capital: Decimal + max_drawdown_usd: Decimal + max_drawdown_pct: Decimal + sharpe_annualized: Decimal | None + skip_reasons: dict[str, int] + trades: list[CompletedTrade] + + +def _build_entry_from_pick( + pick: MondayPick, + cfg: StrategyConfig, + *, + capital_usd: Decimal, + eur_to_usd: Decimal, +) -> BacktestEntry | None: + snap = pick.snapshot + if snap.spot is None or snap.dvol is None: + return None + spot = float(snap.spot) + dvol = float(snap.dvol) + width_pct = float(cfg.structure.spread_width.target_pct_of_spot) + delta_target = float(cfg.structure.short_strike.delta_target) + dte = cfg.structure.dte_target + + credit_eth, short_strike, long_strike = estimate_credit_eth( + spot=spot, dvol_pct=dvol, dte_days=dte, + width_pct=width_pct, delta_target_abs=delta_target, + ) + width_usd = float(cfg.structure.spread_width.target_pct_of_spot) * spot + credit_usd = credit_eth * spot + if width_usd <= 0 or credit_usd / width_usd < float( + cfg.structure.credit_to_width_ratio_min + ): + return None # ratio gate fallisce → no entry + + cap_pertrade_usd = float(cfg.sizing.cap_per_trade_eur) * float(eur_to_usd) + risk_target = min(float(cfg.sizing.kelly_fraction) * float(capital_usd), cap_pertrade_usd) + n_contracts = max(0, min(int(risk_target // width_usd), cfg.sizing.max_contracts_per_trade)) + if n_contracts == 0: + return None + + expiry = pick.timestamp + timedelta(days=dte) + return BacktestEntry( + timestamp=pick.timestamp, + spread_type="bull_put", + spot_at_entry=Decimal(str(spot)), + dvol_at_entry=Decimal(str(dvol)), + short_strike=Decimal(str(round(short_strike, 2))), + long_strike=Decimal(str(round(long_strike, 2))), + expiry=expiry, + credit_received_eth=Decimal(str(credit_eth)), + credit_received_usd=Decimal(str(credit_usd * n_contracts)), + n_contracts=n_contracts, + ) + + +def _max_drawdown_usd(equity: list[Decimal]) -> tuple[Decimal, Decimal]: + """Return ``(max_dd_usd, max_dd_pct_of_peak)`` over an equity curve.""" + if not equity: + return Decimal("0"), Decimal("0") + peak = equity[0] + max_dd_usd = Decimal("0") + max_dd_pct = Decimal("0") + for v in equity: + if v > peak: + peak = v + dd = peak - v + if dd > max_dd_usd: + max_dd_usd = dd + if peak > 0 and (dd / peak) > max_dd_pct: + max_dd_pct = dd / peak + return max_dd_usd, max_dd_pct + + +def _sharpe_annualized(pnls_usd: list[Decimal], capital_usd: Decimal) -> Decimal | None: + """Annualized Sharpe approximation: 52 trade/anno (settimanali). + + Restituisce ``None`` se ci sono <5 trade o stdev = 0. + """ + if len(pnls_usd) < 5 or capital_usd <= 0: + return None + rets = [float(p / capital_usd) for p in pnls_usd] + mean = sum(rets) / len(rets) + var = sum((r - mean) ** 2 for r in rets) / max(1, (len(rets) - 1)) + std = math.sqrt(var) + if std == 0: + return None + sharpe = mean / std * math.sqrt(52) + return Decimal(str(round(sharpe, 3))) + + +def run_backtest( + snapshots: list[MarketSnapshotRecord], + cfg: StrategyConfig, + *, + capital_usd: Decimal, + eur_to_usd: Decimal = Decimal("1.075"), + asset: str = "ETH", +) -> BacktestReport: + """Esegue il backtest end-to-end sui ``snapshots`` ETH ordinati per ts.""" + snapshots = sorted(snapshots, key=lambda s: s.timestamp) + eth_snapshots = [s for s in snapshots if s.asset.upper() == asset.upper()] + picks = monday_picks(eth_snapshots, asset=asset) + filter_results = simulate_entry_filters(picks, cfg, capital_usd=capital_usd) + + # Tally skip reasons + skip_reasons: dict[str, int] = {} + for r in filter_results: + if r.accepted: + continue + for reason in r.reasons: + skip_reasons[reason] = skip_reasons.get(reason, 0) + 1 + + trades: list[CompletedTrade] = [] + for r in filter_results: + if not r.accepted: + continue + entry = _build_entry_from_pick( + r.pick, cfg, capital_usd=capital_usd, eur_to_usd=eur_to_usd, + ) + if entry is None: + skip_reasons["sizing_or_ratio"] = skip_reasons.get("sizing_or_ratio", 0) + 1 + continue + future = [s for s in eth_snapshots if s.timestamp > r.pick.timestamp] + exit_ = simulate_position_outcome(entry, future, cfg) + trades.append(CompletedTrade(entry=entry, exit=exit_)) + + pnls = [t.exit.pnl_usd for t in trades] + cumulative = sum(pnls, start=Decimal("0")) + n_winners = sum(1 for p in pnls if p > 0) + win_rate = ( + Decimal(n_winners) / Decimal(len(pnls)) + if pnls + else Decimal("0") + ) + + # Equity curve in USD assoluti + equity = [capital_usd] + for p in pnls: + equity.append(equity[-1] + p) + max_dd_usd, max_dd_pct = _max_drawdown_usd(equity) + + return BacktestReport( + n_picks=len(picks), + n_accepted=sum(1 for r in filter_results if r.accepted), + n_skipped_data=sum(1 for r in filter_results if r.skipped_for_data), + n_completed=len(trades), + n_winners=n_winners, + win_rate=win_rate, + cumulative_pnl_usd=cumulative, + cumulative_pnl_pct_of_capital=( + cumulative / capital_usd if capital_usd > 0 else Decimal("0") + ), + max_drawdown_usd=max_dd_usd, + max_drawdown_pct=max_dd_pct, + sharpe_annualized=_sharpe_annualized(pnls, capital_usd), + skip_reasons=skip_reasons, + trades=trades, + ) diff --git a/tests/unit/test_backtest.py b/tests/unit/test_backtest.py new file mode 100644 index 0000000..7c3643d --- /dev/null +++ b/tests/unit/test_backtest.py @@ -0,0 +1,259 @@ +"""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, + estimate_credit_eth, + find_strike_for_delta, + monday_picks, + 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) + + +# --------------------------------------------------------------------------- +# Monday 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_monday_picks_extracts_one_per_iso_week() -> None: + monday_2026_05_04 = datetime(2026, 5, 4, 14, 0, tzinfo=UTC) + monday_2026_05_11 = datetime(2026, 5, 11, 14, 0, tzinfo=UTC) + snapshots = [ + _snap(ts=monday_2026_05_04), + _snap(ts=monday_2026_05_04 + timedelta(minutes=15)), # not picked + _snap(ts=monday_2026_05_11), + ] + picks = monday_picks(snapshots) + assert len(picks) == 2 + assert picks[0].timestamp == monday_2026_05_04 + assert picks[1].timestamp == monday_2026_05_11 + + +def test_monday_picks_skips_other_days_and_hours() -> None: + snapshots = [ + _snap(ts=datetime(2026, 5, 4, 13, 0, tzinfo=UTC)), # Monday 13:00 + _snap(ts=datetime(2026, 5, 5, 14, 0, tzinfo=UTC)), # Tuesday 14:00 + ] + assert monday_picks(snapshots) == [] + + +def test_monday_picks_filters_by_asset() -> None: + monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC) + snapshots = [ + _snap(ts=monday, asset="BTC"), + _snap(ts=monday, asset="ETH"), + ] + picks = monday_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 MondayPick + picks = [MondayPick(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 MondayPick + picks = [MondayPick(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 MondayPick + picks = [MondayPick(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 di lunedì alle 14:00: + # offset +1h così vengono ignorati da `monday_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")