"""Event-driven bar backtester. Timeline per bar i (closed bars only): • At close(i) the strategy's target exposure τ(i) is known (features use rows ≤ i). • The resulting order is *submitted* after close(i) and executed at open(i+1+latency) — never on bar i. • Fill price = open ± half-spread ± slippage ± impact ; fee = taker/maker × notional ; partial fills capped at max_participation × volume(i+1) ; qty rounded DOWN to lot_decimals ; orders below ordermin/costmin are dropped. • Every order passes the independent RiskGateway (same class used by paper/live engines). • Equity is marked at close(i+1). Fees, slippage and rejected/partial orders are all accounted in the ledger. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Literal import numpy as np import pandas as pd from ktrade.backtest.costs import CostModel, fee_rate, fill_price, round_lot from ktrade.risk.gateway import MarketState, OrderIntent, PortfolioSnapshot, RiskGateway, RiskLimits from ktrade.strategy.base import Strategy @dataclass class TradeRecord: ts: int side: str qty: float price: float ref_price: float fee: float slippage_cost: float notional: float reason: str strategy_target: float partial: bool = False @dataclass class BacktestResult: equity: pd.DataFrame # ts, equity, cash, position, exposure, price, target trades: list[TradeRecord] rejected: list[dict[str, Any]] fees_paid: float slippage_paid: float turnover: float bars: int diagnostics: dict[str, Any] = field(default_factory=dict) def run_backtest( strategy: Strategy, df: pd.DataFrame, *, interval_min: int, initial_cash: float = 10_000.0, cost_model: CostModel | None = None, risk_limits: RiskLimits | None = None, lot_decimals: int = 8, ordermin: float = 0.0, costmin: float = 0.0, rebalance_threshold: float = 0.05, ) -> BacktestResult: cm = cost_model or CostModel() limits = risk_limits or RiskLimits(max_order_notional=1e12, max_position_notional=1e12, max_gross_exposure=1.0, max_concentration=1.0, daily_loss_limit=1.0, max_drawdown_limit=1.0, stale_data_seconds=1e12, max_open_orders=10, max_participation=cm.max_participation) gateway = RiskGateway(limits) df = df.sort_values("ts").reset_index(drop=True).copy() df.attrs["interval_min"] = interval_min res = strategy.run(df) target = res.target.to_numpy(dtype=float) ts = df["ts"].to_numpy(dtype=np.int64) o = df["open"].to_numpy(dtype=float) c = df["close"].to_numpy(dtype=float) v = df["volume"].to_numpy(dtype=float) n = len(df) cash = initial_cash pos = 0.0 fees = 0.0 slip = 0.0 traded_notional = 0.0 peak = initial_cash trades: list[TradeRecord] = [] rejected: list[dict[str, Any]] = [] rows = [] fr = fee_rate(cm) pending: list[tuple[int, float]] = [] # (execute_at_index, target) for i in range(n): # 1) execute orders scheduled for this bar's open (decided at an earlier close) due = [t for (idx, t) in pending if idx == i] pending = [(idx, t) for (idx, t) in pending if idx != i] if due: tgt = due[-1] ref = o[i] equity_open = cash + pos * ref desired_qty = tgt * equity_open / ref delta = desired_qty - pos # avoid churning: skip if the change in exposure is below the threshold if abs(delta * ref) >= rebalance_threshold * equity_open and abs(delta) > 0: side: Literal["buy", "sell"] = "buy" if delta > 0 else "sell" qty = round_lot(abs(delta), lot_decimals) if side == "sell": qty = min(qty, pos) max_fill = cm.max_participation * v[i] if v[i] > 0 else qty partial = qty > max_fill qty = round_lot(min(qty, max_fill), lot_decimals) px = fill_price(side, ref, qty, v[i], cm) if qty > 0 else ref if side == "buy": # cap by cash incl. fee qty = round_lot(min(qty, cash / (px * (1 + fr))), lot_decimals) intent = OrderIntent(symbol="BT", side=side, qty=qty, strategy=strategy.meta.name) pf = PortfolioSnapshot(equity=equity_open, cash=cash, positions={"BT": pos}, marks={"BT": ref}, peak_equity=peak) mkt = MarketState(last_price=ref, recent_bar_volume=v[i] if v[i] > 0 else None, ordermin=ordermin, costmin=costmin) dec = gateway.evaluate(intent, pf, mkt) if qty > 0 and dec.allowed: notional = qty * px fee = notional * fr if side == "buy": cash -= notional + fee pos += qty else: cash += notional - fee pos -= qty fees += fee sc = abs(px - ref) * qty slip += sc traded_notional += notional trades.append(TradeRecord(ts=int(ts[i]), side=side, qty=qty, price=px, ref_price=ref, fee=fee, slippage_cost=sc, notional=notional, reason="rebalance", strategy_target=tgt, partial=partial)) if partial: pending.append((i + 1, tgt)) # keep working the remainder next bar elif qty > 0: rejected.append({"ts": int(ts[i]), "side": side, "qty": qty, "reasons": dec.reasons}) # 2) mark at close equity = cash + pos * c[i] peak = max(peak, equity) rows.append((int(ts[i]), equity, cash, pos, (pos * c[i]) / equity if equity > 0 else 0.0, c[i], float(target[i]) if not np.isnan(target[i]) else 0.0)) # 3) decision at close(i) → execution at open(i+1+latency) if i + 1 + cm.latency_bars < n and not np.isnan(target[i]): pending.append((i + 1 + cm.latency_bars, float(target[i]))) eq = pd.DataFrame(rows, columns=["ts", "equity", "cash", "position", "exposure", "price", "target"]) avg_equity = float(eq["equity"].mean()) if n else initial_cash bars_per_year = 365 * 24 * 60 / interval_min turnover = (traded_notional / avg_equity) / max(n / bars_per_year, 1e-9) if avg_equity > 0 else 0.0 return BacktestResult(equity=eq, trades=trades, rejected=rejected, fees_paid=fees, slippage_paid=slip, turnover=turnover, bars=n, diagnostics={"strategy": res.diagnostics, "n_partial": sum(t.partial for t in trades)})