"""Execution cost & fill model shared by the backtester and the paper simulator.""" from __future__ import annotations import math from decimal import ROUND_DOWN, Decimal from pydantic import BaseModel, Field class CostModel(BaseModel): taker_fee: float = Field(0.0040, description="Kraken spot taker fee at the lowest tier (verify with TradeVolume)") maker_fee: float = Field(0.0025) spread_bps: float = Field(5.0, description="Assumed quoted spread when no book data (bps of mid)") slippage_bps: float = Field(2.0, description="Extra adverse move per order") impact_coeff: float = Field(0.1, description="Impact (bps) = coeff × sqrt(qty / bar_volume) × 1e4") max_participation: float = Field(0.05, description="Max fraction of bar volume filled per bar (partial fills)") latency_bars: int = Field(0, description="Extra bars of delay beyond next-open execution") use_maker: bool = False fee_multiplier: float = 1.0 # stress: fee increases spread_multiplier: float = 1.0 # stress: spread widening def round_lot(qty: float, lot_decimals: int) -> float: if qty <= 0 or not math.isfinite(qty): return 0.0 q = Decimal(str(qty)).quantize(Decimal(1).scaleb(-lot_decimals), rounding=ROUND_DOWN) return float(q) def fill_price(side: str, ref_price: float, qty: float, bar_volume: float, cm: CostModel) -> float: half_spread = ref_price * (cm.spread_bps * cm.spread_multiplier) / 2e4 slip = ref_price * cm.slippage_bps / 1e4 impact = ref_price * cm.impact_coeff * math.sqrt(qty / bar_volume) if bar_volume > 0 else 0.0 adverse = (0.0 if cm.use_maker else half_spread) + slip + impact return ref_price + adverse if side == "buy" else max(ref_price - adverse, 1e-12) def fee_rate(cm: CostModel) -> float: return (cm.maker_fee if cm.use_maker else cm.taker_fee) * cm.fee_multiplier