diff --git a/src/analysis/backtest_execution_model.py b/src/analysis/backtest_execution_model.py new file mode 100644 index 0000000..24798dc --- /dev/null +++ b/src/analysis/backtest_execution_model.py @@ -0,0 +1,103 @@ +"""Conservative backtest execution model.""" + +from __future__ import annotations + +from dataclasses import dataclass +import math +from random import Random +from typing import Literal + + +OrderSide = Literal["BUY", "SELL"] + + +@dataclass(frozen=True) +class ExecutionRequest: + side: OrderSide + session_id: str + qty: int + reference_price: float + + +@dataclass(frozen=True) +class ExecutionAssumptions: + slippage_bps_by_session: dict[str, float] + failure_rate_by_session: dict[str, float] + partial_fill_rate_by_session: dict[str, float] + partial_fill_min_ratio: float = 0.3 + partial_fill_max_ratio: float = 0.8 + seed: int = 0 + + +@dataclass(frozen=True) +class ExecutionResult: + status: Literal["FILLED", "PARTIAL", "REJECTED"] + filled_qty: int + avg_price: float + slippage_bps: float + reason: str + + +class BacktestExecutionModel: + """Execution simulator with conservative unfavorable fill assumptions.""" + + def __init__(self, assumptions: ExecutionAssumptions) -> None: + self.assumptions = assumptions + self._rng = Random(assumptions.seed) + if assumptions.partial_fill_min_ratio <= 0 or assumptions.partial_fill_max_ratio > 1: + raise ValueError("partial fill ratios must be within (0,1]") + if assumptions.partial_fill_min_ratio > assumptions.partial_fill_max_ratio: + raise ValueError("partial_fill_min_ratio must be <= partial_fill_max_ratio") + for sess, bps in assumptions.slippage_bps_by_session.items(): + if not math.isfinite(bps) or bps < 0: + raise ValueError(f"slippage_bps must be finite and >= 0 for session={sess}") + for sess, rate in assumptions.failure_rate_by_session.items(): + if not math.isfinite(rate) or rate < 0 or rate > 1: + raise ValueError(f"failure_rate must be in [0,1] for session={sess}") + for sess, rate in assumptions.partial_fill_rate_by_session.items(): + if not math.isfinite(rate) or rate < 0 or rate > 1: + raise ValueError(f"partial_fill_rate must be in [0,1] for session={sess}") + + def simulate(self, request: ExecutionRequest) -> ExecutionResult: + if request.qty <= 0: + raise ValueError("qty must be positive") + if request.reference_price <= 0: + raise ValueError("reference_price must be positive") + + slippage_bps = self.assumptions.slippage_bps_by_session.get(request.session_id, 0.0) + failure_rate = self.assumptions.failure_rate_by_session.get(request.session_id, 0.0) + partial_rate = self.assumptions.partial_fill_rate_by_session.get(request.session_id, 0.0) + + if self._rng.random() < failure_rate: + return ExecutionResult( + status="REJECTED", + filled_qty=0, + avg_price=0.0, + slippage_bps=slippage_bps, + reason="execution_failure", + ) + + slip_mult = 1.0 + (slippage_bps / 10000.0 if request.side == "BUY" else -slippage_bps / 10000.0) + exec_price = request.reference_price * slip_mult + + if self._rng.random() < partial_rate: + ratio = self._rng.uniform( + self.assumptions.partial_fill_min_ratio, + self.assumptions.partial_fill_max_ratio, + ) + filled = max(1, min(request.qty - 1, int(request.qty * ratio))) + return ExecutionResult( + status="PARTIAL", + filled_qty=filled, + avg_price=exec_price, + slippage_bps=slippage_bps, + reason="partial_fill", + ) + + return ExecutionResult( + status="FILLED", + filled_qty=request.qty, + avg_price=exec_price, + slippage_bps=slippage_bps, + reason="filled", + ) diff --git a/tests/test_backtest_execution_model.py b/tests/test_backtest_execution_model.py new file mode 100644 index 0000000..fb2fa58 --- /dev/null +++ b/tests/test_backtest_execution_model.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +import pytest + +from src.analysis.backtest_execution_model import ( + BacktestExecutionModel, + ExecutionAssumptions, + ExecutionRequest, +) + + +def test_buy_uses_unfavorable_slippage_direction() -> None: + model = BacktestExecutionModel( + ExecutionAssumptions( + slippage_bps_by_session={"US_PRE": 50.0}, + failure_rate_by_session={"US_PRE": 0.0}, + partial_fill_rate_by_session={"US_PRE": 0.0}, + seed=1, + ) + ) + out = model.simulate( + ExecutionRequest(side="BUY", session_id="US_PRE", qty=10, reference_price=100.0) + ) + assert out.status == "FILLED" + assert out.avg_price == pytest.approx(100.5) + + +def test_sell_uses_unfavorable_slippage_direction() -> None: + model = BacktestExecutionModel( + ExecutionAssumptions( + slippage_bps_by_session={"US_PRE": 50.0}, + failure_rate_by_session={"US_PRE": 0.0}, + partial_fill_rate_by_session={"US_PRE": 0.0}, + seed=1, + ) + ) + out = model.simulate( + ExecutionRequest(side="SELL", session_id="US_PRE", qty=10, reference_price=100.0) + ) + assert out.status == "FILLED" + assert out.avg_price == pytest.approx(99.5) + + +def test_failure_rate_can_reject_order() -> None: + model = BacktestExecutionModel( + ExecutionAssumptions( + slippage_bps_by_session={"KRX_REG": 10.0}, + failure_rate_by_session={"KRX_REG": 1.0}, + partial_fill_rate_by_session={"KRX_REG": 0.0}, + seed=42, + ) + ) + out = model.simulate( + ExecutionRequest(side="BUY", session_id="KRX_REG", qty=10, reference_price=100.0) + ) + assert out.status == "REJECTED" + assert out.filled_qty == 0 + + +def test_partial_fill_applies_when_rate_is_one() -> None: + model = BacktestExecutionModel( + ExecutionAssumptions( + slippage_bps_by_session={"KRX_REG": 0.0}, + failure_rate_by_session={"KRX_REG": 0.0}, + partial_fill_rate_by_session={"KRX_REG": 1.0}, + partial_fill_min_ratio=0.4, + partial_fill_max_ratio=0.4, + seed=0, + ) + ) + out = model.simulate( + ExecutionRequest(side="BUY", session_id="KRX_REG", qty=10, reference_price=100.0) + ) + assert out.status == "PARTIAL" + assert out.filled_qty == 4 + assert out.avg_price == 100.0 + + +@pytest.mark.parametrize("bad_slip", [-1.0, float("nan"), float("inf")]) +def test_invalid_slippage_is_rejected(bad_slip: float) -> None: + with pytest.raises(ValueError, match="slippage_bps"): + BacktestExecutionModel( + ExecutionAssumptions( + slippage_bps_by_session={"US_PRE": bad_slip}, + failure_rate_by_session={"US_PRE": 0.0}, + partial_fill_rate_by_session={"US_PRE": 0.0}, + ) + ) + + +@pytest.mark.parametrize("bad_rate", [-0.1, 1.1, float("nan")]) +def test_invalid_failure_or_partial_rates_are_rejected(bad_rate: float) -> None: + with pytest.raises(ValueError, match="failure_rate"): + BacktestExecutionModel( + ExecutionAssumptions( + slippage_bps_by_session={"US_PRE": 10.0}, + failure_rate_by_session={"US_PRE": bad_rate}, + partial_fill_rate_by_session={"US_PRE": 0.0}, + ) + ) + with pytest.raises(ValueError, match="partial_fill_rate"): + BacktestExecutionModel( + ExecutionAssumptions( + slippage_bps_by_session={"US_PRE": 10.0}, + failure_rate_by_session={"US_PRE": 0.0}, + partial_fill_rate_by_session={"US_PRE": bad_rate}, + ) + )