File size: 2,739 Bytes
c45a387 2392da2 c45a387 2392da2 c45a387 2392da2 c45a387 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """Backtest engine: net PnL from positions, bars and cost models.
PnL at bar t (bar covers [t, t+freq)) accrues from holding the position
decided at t through the bar. Decision-time discipline is the strategy's
responsibility (use shift_bars in crypto_eval.strategies); a strategy
that peeks is caught downstream by the purge/embargo audit and by the
forward-return structure here.
Fees + slippage are charged on position changes (turnover), funding is
accrued at funding ticks per costs.funding_cost.
"""
from __future__ import annotations
import numpy as np
import pandas as pd
from .costs import funding_cost, fee, slippage_cost
def _realized_daily_vol_bps(close: pd.Series) -> float:
rets = close.pct_change().dropna()
ann = rets.std(ddof=1) * np.sqrt(24 * 365)
return float(ann / np.sqrt(365) * 1e4)
def run_backtest(
bars: pd.DataFrame,
positions: pd.Series,
funding: pd.Series,
fill_side: str = "taker",
adv_notional: float = 5e7, # Binance BTCUSDT spot ADV, conservative
initial_capital: float = 1_000_000.0,
) -> dict:
if not positions.index.equals(bars.index):
raise ValueError("positions index must equal bars index")
if positions.isna().any():
raise ValueError("positions contain NaN (indeterminate signal)")
if isinstance(funding, pd.DataFrame):
funding = funding["funding_rate"]
close = bars["close"]
pos = positions
# per-bar PnL: position decided at t earns the bar's return
bar_ret = close.pct_change()
# position at t held over [t, t+1h): PnL uses bar_ret at t+1
pnl_per_bar = pos * bar_ret.shift(-1) * close
# the final bar has no realized forward return; treat as flat/zero
pnl_per_bar = pnl_per_bar.fillna(0.0)
# turnover: position changes at t fill at bar t's close
dpos = pos.diff().fillna(pos.iloc[0])
turnover_notional = (dpos.abs() * close)
daily_vol_bps = _realized_daily_vol_bps(close)
fees = turnover_notional.map(lambda n: fee(n, fill_side))
slips = turnover_notional.map(
lambda n: slippage_cost(n, adv_notional, daily_vol_bps))
fund = funding_cost(pos, funding, close)
total_cost = fees + slips + fund
net_pnl = pnl_per_bar - total_cost
net_ret = net_pnl / initial_capital
return {
"net_pnl": net_pnl,
"net_return": net_ret,
"gross_pnl": pnl_per_bar,
"fees": fees,
"slippage": slips,
"funding": fund,
"turnover_notional": turnover_notional,
"total_fees": float(fees.sum()),
"total_slippage": float(slips.sum()),
"total_funding": float(fund.sum()),
"gross_total_pnl": float(pnl_per_bar.sum()),
"net_total_pnl": float(net_pnl.sum()),
}
|