1 Commits

Author SHA1 Message Date
root f664ea1a15 feat(backtest): stylized engine over market_snapshots + CLI subcommand
Aggiunge `core/backtest.py`, motore di backtesting stilizzato che gira
sui dati raccolti in `market_snapshots`. Risponde alla domanda:
"se questa config fosse stata attiva nelle ultime N settimane, quanti
lunedì avrebbero superato i filtri e quale sarebbe stato il P/L stimato?"

**Architettura a due strati**:

1. **Filtri di entry — RIGOROSO**: per ogni Monday-14:00-UTC nei
   snapshot ricostruisce `EntryContext` e chiama lo stesso
   `validate_entry()` del live. Output esatto di "cosa avrebbe deciso
   il bot" per ogni settimana, con conteggio dei motivi di skip.

2. **P/L per trade accettato — STILIZZATO**: senza catena opzioni
   storica, stima credito/exit via Black-Scholes con skew premium
   (default 1.5×) per approssimare la vol smile dell'ETH. Re-prezza
   il combo ad ogni tick futuro per simulare i trigger §7
   (profit_take, stop_loss, vol_stop, time_stop, expiry).

**Aggregati nel `BacktestReport`**:
- n_picks / n_accepted / n_skipped_data / n_completed / n_winners
- win_rate, P/L cumulato (USD + % su capitale)
- max drawdown (USD + % di peak)
- Sharpe annualizzato (52 settimane)
- skip_reasons: dict{motivo → settimane bloccate}

**CLI**: nuovo `cerbero-bite backtest --strategy F --from D --to D
--capital N --asset ETH`. Stampa Rich-formatted summary + tabella
motivi di skip. Esempio:

    cerbero-bite backtest \
      --strategy strategy.aggressiva.yaml \
      --from 2026-04-01 --to 2026-05-01 \
      --capital 10000

**Limiti dichiarati**:
- BS + skew_premium ≠ catena reale: i numeri P/L sono **stime ex-post
  per ranking config**, non promesse operative. Buono per dire
  "config A batte config B sui dati reali", non per dimensionare
  capitale.
- skew_premium 1.5× è stato calibrato sui dati Deribit storici
  (smile slope ETH options); va rifinito quando avremo abbastanza
  chain history da farlo empiricamente.

**Tests**: 15 unit test (BS math, monday picks, filter sim,
position outcome simulation, full pipeline su sintetico).
Suite totale: 420 passed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 20:31:54 +00:00
14 changed files with 1085 additions and 321 deletions
-63
View File
@@ -532,69 +532,6 @@ quella che il sistema parte ad eseguire.
---
## 4-quater. IV richness gate (§2.9): il filtro che alza il win-rate
Il filtro a maggior impatto sull'edge è anche il più semplice da
descrivere: **non vendere vol quando la IV non sta pagando un margine
misurabile sopra la RV**. È implementato come gate hard nel
`validate_entry`:
```
if iv_minus_rv_filter_enabled and iv_minus_rv < iv_minus_rv_min:
skip entry
```
con due parametri in `entry:` di `strategy.yaml`:
| Parametro | Default | Effetto |
|---|---|---|
| `iv_minus_rv_filter_enabled` | `false` (golden) / `true` (aggressiva) | Master switch del gate |
| `iv_minus_rv_min` | `0` (golden) / `3` (aggressiva) | Soglia in punti vol che IV30g RV30g deve eccedere |
Il dato è già raccolto in `market_snapshots.iv_minus_rv` ogni 15
minuti. Il gate consulta l'ultimo tick disponibile al momento
dell'entry cycle (non un percentile rolling — quello è il prossimo
step di calibrazione, vedi §4-quinquies in roadmap).
**Profili di default ragionati.**
- **Conservativa / golden config**: `enabled=false, min=0`. Tutti i
setup passano questo gate, anche con IV-RV negativa. Motivo: nei
primi 8 turni di lunedì non si hanno abbastanza tick per stabilire
che soglia ha senso nel proprio regime. Lasciamo la pagina
`📐 Calibrazione` mostrare la distribuzione e poi alziamo
manualmente.
- **Aggressiva**: `enabled=true, min=3`. Il profilo aggressivo già di
suo prende size più grande; pretendere `IV-RV ≥ 3 vol points` come
prerequisito è coerente — se stai betting più grosso, vuoi
win-rate più alto. La soglia 3 è conservativa; la letteratura
short-vol systematic suggerisce 5 dopo calibrazione.
**Cosa cambia nel P/L atteso quando attivi il gate.**
Il gate **riduce** il numero di entry (saltiamo settimane con premio
magro) ma **alza** la qualità di quelle che passano (premio ricco =
win-rate empirico più alto). Effetto netto sul P/L annuo:
- Trade/anno: 18 → 12-14 (skip più aggressivo)
- Win-rate atteso: 0.72 → 0.78-0.80
- E[trade] netto: +0.6 USD → +4-6 USD per contratto
- **P/L annuo proiettato sale anche se i trade scendono**, perché
ogni trade ha edge più alto.
La pagina `📚 Strategia` ha lo slider win-rate già coerente con
questa logica: muovi da 0.72 a 0.78 e vedi l'APR scattare.
**Roadmap di hardening (passi successivi al merge di questo PR).**
1. **Soglia adattiva**: sostituire `iv_minus_rv_min: 3` con un valore
calcolato a runtime come `P25 rolling 60d` di `market_snapshots.iv_minus_rv`.
2. **Vol-of-vol guard**: bloccare entry quando `dvol` è cambiato di
≥5 punti nelle ultime 24h, anche se `iv_minus_rv` è alto (regime
instabile).
3. **Multi-asset (ETH+BTC)**: come da §4-ter, sblocca il
moltiplicatore 2× sulle opportunità a parità di filtri.
## 5. Come leggere il dato giorno per giorno
Tre euristiche operative sui campi raccolti:
+150 -1
View File
@@ -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."""
-9
View File
@@ -75,15 +75,6 @@ class EntryConfig(BaseModel):
dealer_gamma_filter_enabled: bool = True
liquidation_filter_enabled: bool = True
# 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
# ---------------------------------------------------------------------------
# Structure
+652
View File
@@ -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,
)
-20
View File
@@ -44,12 +44,6 @@ class EntryContext(BaseModel):
dealer_net_gamma: Decimal | None = None
liquidation_squeeze_risk_high: bool | None = None
# 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
class EntryDecision(BaseModel):
"""Result of :func:`validate_entry`. ``reasons`` holds *all* blocking reasons."""
@@ -137,20 +131,6 @@ def validate_entry(ctx: EntryContext, cfg: StrategyConfig) -> EntryDecision:
):
reasons.append("imminent liquidation squeeze risk")
# §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)"
)
return EntryDecision(accepted=not reasons, reasons=reasons)
+8 -104
View File
@@ -11,7 +11,6 @@ La pagina è di sola lettura: non chiama MCP, non scrive sul DB.
from __future__ import annotations
import math
import os
from dataclasses import dataclass
from pathlib import Path
@@ -281,18 +280,13 @@ def _build_gates(
)
)
# --- IV RV (richness) — gate §2.9 ---------------------------
# --- IV RV (richness) — solo informativo --------------------
rv = (
float(snap.realized_vol_30d) if snap.realized_vol_30d is not None else None
)
iv_minus_rv = (
float(snap.iv_minus_rv) if snap.iv_minus_rv is not None else None
)
iv_min = float(getattr(entry, "iv_minus_rv_min", 0.0)) if entry else 0.0
iv_enabled = (
bool(getattr(entry, "iv_minus_rv_filter_enabled", False)) if entry else False
)
if not iv_enabled:
rows.append(
_GateRow(
"IV RV (richness)",
@@ -301,34 +295,9 @@ def _build_gates(
if iv_minus_rv is not None
else ""
),
"filtro DISABILITATO (info-only)",
"n/a",
f"RV30={rv:.2f} · attiva con `iv_minus_rv_filter_enabled: true`"
if rv is not None
else "Attiva con `iv_minus_rv_filter_enabled: true`",
)
)
elif iv_minus_rv is None:
rows.append(
_GateRow(
"IV RV ≥ soglia",
"",
f"{iv_min:.1f} pt vol",
"n/a",
"Dato non disponibile in questo tick (best-effort skip).",
)
)
else:
ok = iv_minus_rv >= iv_min
rows.append(
_GateRow(
"IV RV ≥ soglia",
f"{iv_minus_rv:+.2f} pt vol",
f"{iv_min:.1f} pt vol",
"pass" if ok else "fail",
"Premio ricco rispetto a quanto il mercato si è davvero "
"mosso → edge sostenibile per il venditore di vol."
+ (f" RV30={rv:.2f}" if rv is not None else ""),
"info, > 0 = premio ricco",
"pass" if (iv_minus_rv is not None and iv_minus_rv > 0) else "n/a",
f"RV30={rv:.2f}" if rv is not None else "",
)
)
@@ -421,37 +390,6 @@ def _compute_pl(
annual_pl = trades_per_year * n_per_trade * concurrency * e_trade_net
apr = (annual_pl / capital) if capital > 0 else 0.0
# --- Max drawdown -------------------------------------------------
# Due metriche distinte:
#
# 1. **Streak atteso (P99)**: lunghezza della peggior sequenza di
# stop consecutivi che ci si aspetta di vedere in un anno con
# probabilità ≤ 1%. Usa l'approssimazione union-bound:
# P(streak ≥ N in N_trade tentativi) ≈ N_trade × p_loss^N
# Imponendo questa quantità ≤ 0.01 e risolvendo per N:
# N = ceil( log(0.01 / N_trade) / log(p_loss) )
# Drawdown corrispondente = N × stop_loss × contracts × concurrency.
#
# 2. **Tail/gap risk**: scenario "gap notturno" in cui il mark
# salta oltre la copertura long PRIMA che lo stop sia
# eseguibile. La perdita massima reale è la larghezza intera
# dello spread meno il credito iniziale, su tutte le posizioni
# aperte simultaneamente.
if prob_loss > 0 and prob_loss < 1 and trades_per_year > 0:
streak_99 = max(
1,
int(math.ceil(
math.log(0.01 / trades_per_year) / math.log(prob_loss)
)) if prob_loss < 1 else 1,
)
else:
streak_99 = 0
expected_dd_usd = streak_99 * sl_loss * n_per_trade * concurrency
expected_dd_pct = expected_dd_usd / capital if capital > 0 else 0.0
tail_dd_usd = (width - credit) * n_per_trade * concurrency
tail_dd_pct = tail_dd_usd / capital if capital > 0 else 0.0
return {
"width": width,
"credit": credit,
@@ -465,12 +403,6 @@ def _compute_pl(
"apr": apr,
"fees": fees,
"slippage": slippage,
"prob_loss": prob_loss,
"streak_99": float(streak_99),
"expected_dd_usd": expected_dd_usd,
"expected_dd_pct": expected_dd_pct,
"tail_dd_usd": tail_dd_usd,
"tail_dd_pct": tail_dd_pct,
}
@@ -507,34 +439,6 @@ def _render_profile_card(
delta=f"{metrics['apr']:+.1%} APR",
)
cols = st.columns(2)
cols[0].metric(
"Max DD attesa (P99)",
f"{metrics['expected_dd_usd']:.0f} USD",
delta=f"{-metrics['expected_dd_pct']:+.1%} cap",
delta_color="inverse",
help=(
f"Streak di {int(metrics['streak_99'])} stop consecutivi "
f"(probabilità ≤ 1% nell'anno) × perdita stop "
f"({metrics['sl_loss']:.0f} USD) × contratti × posizioni "
f"concorrenti. È la peggior sequenza che ti aspetti di "
"vedere; il drawdown reale può essere maggiore se i filtri "
"non rilevano un regime change."
),
)
cols[1].metric(
"Max DD coda (gap)",
f"{metrics['tail_dd_usd']:.0f} USD",
delta=f"{-metrics['tail_dd_pct']:+.1%} cap",
delta_color="inverse",
help=(
"Scenario gap notturno: il mark salta oltre la copertura "
"long PRIMA che lo stop sia eseguibile. Perdita = larghezza "
"intera meno credito, su tutte le posizioni aperte. "
"I filtri quant + macro lo riducono ma NON lo annullano."
),
)
if metrics["n_per_trade"] == 0:
st.warning(
"Sizing 0 contratti: capitale insufficiente per i cap di "
@@ -647,10 +551,10 @@ def _render_pl_panel(
sens_rows.append(
{
"Win rate": f"{wr:.0%}",
"Cons. APR": f"{m_c['apr']:+.1%}",
"Cons. Max DD": f"{m_c['expected_dd_pct']:.1%}",
"Aggr. APR": f"{m_a['apr']:+.1%}",
"Aggr. Max DD": f"{m_a['expected_dd_pct']:.1%}",
"Conservativa P/L": f"{m_c['annual_pl']:+.0f} USD",
"Conservativa APR": f"{m_c['apr']:+.1%}",
"Aggressiva P/L": f"{m_a['annual_pl']:+.0f} USD",
"Aggressiva APR": f"{m_a['apr']:+.1%}",
}
)
st.table(sens_rows)
-24
View File
@@ -94,7 +94,6 @@ class _MarketSnapshot:
portfolio_eur: Decimal
dealer_net_gamma: Decimal | None
liquidation_squeeze_risk_high: bool | None
iv_minus_rv: Decimal | None
async def _gather_snapshot(
@@ -160,9 +159,6 @@ async def _gather_snapshot(
liquidation_t: asyncio.Task[bool | None] = asyncio.create_task(
_safe_liquidation_squeeze(sentiment)
)
iv_rv_t: asyncio.Task[Decimal | None] = asyncio.create_task(
_safe_iv_minus_rv(deribit)
)
await asyncio.gather(
spot_t,
@@ -176,7 +172,6 @@ async def _gather_snapshot(
portfolio_t,
dealer_t,
liquidation_t,
iv_rv_t,
)
return _MarketSnapshot(
spot_eth_usd=spot_t.result(),
@@ -190,7 +185,6 @@ async def _gather_snapshot(
portfolio_eur=portfolio_t.result(),
dealer_net_gamma=dealer_t.result(),
liquidation_squeeze_risk_high=liquidation_t.result(),
iv_minus_rv=iv_rv_t.result(),
)
@@ -202,20 +196,6 @@ async def _safe_dealer_gamma(deribit: DeribitClient) -> Decimal | None:
return snap.total_net_dealer_gamma
async def _safe_iv_minus_rv(deribit: DeribitClient) -> Decimal | None:
"""Best-effort fetch of the IV30g RV30g spread (vol points)."""
try:
rv = await deribit.realized_vol("ETH")
except Exception:
return None
if not isinstance(rv, dict):
return None
value = rv.get("iv_minus_rv_30d")
if value is None:
return None
return value if isinstance(value, Decimal) else Decimal(str(value))
async def _safe_liquidation_squeeze(sentiment: SentimentClient) -> bool | None:
try:
heatmap = await sentiment.liquidation_heatmap("ETH")
@@ -373,7 +353,6 @@ async def run_entry_cycle(
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,
)
decision = validate_entry(entry_ctx, cfg)
@@ -391,9 +370,6 @@ async def run_entry_cycle(
"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
),
}
}
if not decision.accepted:
+2 -9
View File
@@ -28,8 +28,8 @@
# 2× via "ETH + BTC" indicato in `📚 Strategia` è una **stima ex-ante**
# di cosa otterresti DOPO quel lavoro di codice.
config_version: "1.1.0-aggressiva"
config_hash: "58086a4afbbf36c48d22f39bbc75d8145e76a063917431793d3b92ae76b5eb68"
config_version: "1.0.0-aggressiva"
config_hash: "b931a2b96fbc149b21cae84a196ee8bad10220b5ee8fa9ab0ed06ae52d7dc531"
last_review: "2026-04-26"
last_reviewer: "Adriano"
@@ -66,13 +66,6 @@ entry:
dealer_gamma_filter_enabled: true
liquidation_filter_enabled: true
# IV richness gate (§2.9) — abilitato con soglia 3 pt vol.
# Coerente con il profilo aggressivo: size più grande pretende
# win-rate più alto. La soglia 3 va alzata a 5 dopo la
# calibrazione (4-8 settimane di dati raccolti).
iv_minus_rv_min: "3"
iv_minus_rv_filter_enabled: true
structure:
dte_target: 18
dte_min: 14
+2 -6
View File
@@ -15,8 +15,8 @@
# cerbero-bite config hash --file strategy.conservativa.yaml
# e bumpare config_version.
config_version: "1.1.0-conservativa"
config_hash: "188155fd0017a1353024151b8237f257b0c3156d2592ce89653d239b39fb69ce"
config_version: "1.0.0-conservativa"
config_hash: "eff824281bbb538fba49434d8cc4b9c37675bc73d60e351293e263cc7e7b29ef"
last_review: "2026-04-26"
last_reviewer: "Adriano"
@@ -50,10 +50,6 @@ entry:
dealer_gamma_filter_enabled: true
liquidation_filter_enabled: true
# IV richness gate (§2.9) — disabilitato finché non calibrato.
iv_minus_rv_min: "0"
iv_minus_rv_filter_enabled: false
structure:
dte_target: 18
dte_min: 14
+2 -9
View File
@@ -6,8 +6,8 @@
# config hash), and lands as a separate commit with the motivation in
# the commit message.
config_version: "1.1.0"
config_hash: "e0504e6936e9ec5013e7901cf98532e29ff2414b1cce10461cfe97790119b724"
config_version: "1.0.0"
config_hash: "4c2be4c51c849ed58fa22ec2b302016c453894dd0964b6d05445ab1b723e2d10"
last_review: "2026-04-26"
last_reviewer: "Adriano"
@@ -46,13 +46,6 @@ entry:
dealer_gamma_filter_enabled: true
liquidation_filter_enabled: true
# IV richness gate (§2.9). Disabilitato di default: è il filtro
# con maggior impatto sul win-rate ma va calibrato sui dati
# raccolti in `market_snapshots` prima di metterlo in produzione.
# Vedi `docs/13-strategia-spiegata.md` §4-quater.
iv_minus_rv_min: "0"
iv_minus_rv_filter_enabled: false
structure:
dte_target: 18
dte_min: 14
-10
View File
@@ -118,16 +118,6 @@ def _wire_market_snapshot(
},
is_reusable=True,
)
httpx_mock.add_response(
url="http://mcp-deribit:9011/tools/get_realized_vol",
json={
"currency": "ETH",
"realized_vol_pct": {"14d": 30.0, "30d": 30.0},
"iv_current_pct": 38.0,
"iv_minus_rv_pct": {"14d": 8.0, "30d": 8.0},
},
is_reusable=True,
)
httpx_mock.add_response(
url="http://mcp-sentiment:9014/tools/get_liquidation_heatmap",
json={
+259
View File
@@ -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")
+1 -1
View File
@@ -68,7 +68,7 @@ def test_compute_hash_is_independent_of_recorded_hash_value(tmp_path: Path) -> N
def test_load_repo_strategy_yaml(tmp_path: Path) -> None:
"""The committed strategy.yaml validates with the recorded hash."""
result = load_strategy(REPO_ROOT / "strategy.yaml")
assert result.config.config_version == "1.1.0"
assert result.config.config_version == "1.0.0"
assert result.config.sizing.kelly_fraction == Decimal("0.13")
assert result.computed_hash == result.config.config_hash
-56
View File
@@ -194,62 +194,6 @@ def test_dealer_gamma_filter_disabled_in_config(cfg: StrategyConfig) -> None:
assert decision.accepted is True
# ---------------------------------------------------------------------------
# IV richness gate (§2.9)
# ---------------------------------------------------------------------------
def _strict_iv_rv_cfg(
cfg: StrategyConfig, *, threshold: Decimal = Decimal("5")
) -> StrategyConfig:
return golden_config(
entry=EntryConfig(
**{
**cfg.entry.model_dump(),
"iv_minus_rv_filter_enabled": True,
"iv_minus_rv_min": threshold,
}
)
)
def test_iv_richness_gate_disabled_by_default_lets_thin_premium_pass(
cfg: StrategyConfig,
) -> None:
# Default config: filter disabled. Anche con IV-RV negativa (RV>IV)
# l'entry deve passare per non rompere setup pre-calibrazione.
decision = validate_entry(_good_ctx(iv_minus_rv=Decimal("-2")), cfg)
assert decision.accepted is True
def test_iv_richness_gate_blocks_when_below_floor(cfg: StrategyConfig) -> None:
strict = _strict_iv_rv_cfg(cfg, threshold=Decimal("5"))
decision = validate_entry(_good_ctx(iv_minus_rv=Decimal("3")), strict)
assert decision.accepted is False
assert any("IV richness" in r for r in decision.reasons)
def test_iv_richness_gate_passes_when_above_floor(cfg: StrategyConfig) -> None:
strict = _strict_iv_rv_cfg(cfg, threshold=Decimal("5"))
decision = validate_entry(_good_ctx(iv_minus_rv=Decimal("6")), strict)
assert decision.accepted is True
def test_iv_richness_gate_passes_at_exact_threshold(cfg: StrategyConfig) -> None:
# Soglia inclusiva: IV-RV == soglia → accettato (gate è "<", non "<=").
strict = _strict_iv_rv_cfg(cfg, threshold=Decimal("5"))
decision = validate_entry(_good_ctx(iv_minus_rv=Decimal("5")), strict)
assert decision.accepted is True
def test_iv_richness_gate_skipped_when_data_missing(cfg: StrategyConfig) -> None:
# MCP irraggiungibile: best-effort skip, non bloccare l'entry per
# un problema di infrastruttura.
strict = _strict_iv_rv_cfg(cfg, threshold=Decimal("5"))
decision = validate_entry(_good_ctx(iv_minus_rv=None), strict)
assert decision.accepted is True
def test_validate_entry_accumulates_all_reasons(cfg: StrategyConfig) -> None:
decision = validate_entry(
_good_ctx(