engine passes realized daily vol to impact model

2392da2 verified about 10 hours ago

2.74 kB

"""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()),
    }

	"""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()),
	}