Merge feat/backtest-engine
This commit is contained in:
root
+150
-1
@@ -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."""
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
Reference in New Issue
Block a user