fix: validate execution assumption ranges in backtest model

src/analysis/backtest_cost_guard.py

@@ -0,0 +1,52 @@
+"""Backtest cost/slippage/failure validation guard."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import math
+
+
+@dataclass(frozen=True)
+class BacktestCostModel:
+    commission_bps: float | None = None
+    slippage_bps_by_session: dict[str, float] | None = None
+    failure_rate_by_session: dict[str, float] | None = None
+    unfavorable_fill_required: bool = True
+
+
+def validate_backtest_cost_model(
+    *,
+    model: BacktestCostModel,
+    required_sessions: list[str],
+) -> None:
+    """Raise ValueError when required cost assumptions are missing/invalid."""
+    if (
+        model.commission_bps is None
+        or not math.isfinite(model.commission_bps)
+        or model.commission_bps < 0
+    ):
+        raise ValueError("commission_bps must be provided and >= 0")
+    if not model.unfavorable_fill_required:
+        raise ValueError("unfavorable_fill_required must be True")
+
+    slippage = model.slippage_bps_by_session or {}
+    failure = model.failure_rate_by_session or {}
+
+    missing_slippage = [s for s in required_sessions if s not in slippage]
+    if missing_slippage:
+        raise ValueError(
+            f"missing slippage_bps_by_session for sessions: {', '.join(missing_slippage)}"
+        )
+
+    missing_failure = [s for s in required_sessions if s not in failure]
+    if missing_failure:
+        raise ValueError(
+            f"missing failure_rate_by_session for sessions: {', '.join(missing_failure)}"
+        )
+
+    for sess, bps in slippage.items():
+        if not math.isfinite(bps) or bps < 0:
+            raise ValueError(f"slippage bps must be >= 0 for session={sess}")
+    for sess, rate in failure.items():
+        if not math.isfinite(rate) or rate < 0 or rate > 1:
+            raise ValueError(f"failure rate must be within [0,1] for session={sess}")

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",
+        )

tests/test_backtest_cost_guard.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import pytest
+
+from src.analysis.backtest_cost_guard import BacktestCostModel, validate_backtest_cost_model
+
+
+def test_valid_backtest_cost_model_passes() -> None:
+    model = BacktestCostModel(
+        commission_bps=5.0,
+        slippage_bps_by_session={"KRX_REG": 10.0, "US_PRE": 50.0},
+        failure_rate_by_session={"KRX_REG": 0.01, "US_PRE": 0.08},
+        unfavorable_fill_required=True,
+    )
+    validate_backtest_cost_model(model=model, required_sessions=["KRX_REG", "US_PRE"])
+
+
+def test_missing_required_slippage_session_raises() -> None:
+    model = BacktestCostModel(
+        commission_bps=5.0,
+        slippage_bps_by_session={"KRX_REG": 10.0},
+        failure_rate_by_session={"KRX_REG": 0.01, "US_PRE": 0.08},
+        unfavorable_fill_required=True,
+    )
+    with pytest.raises(ValueError, match="missing slippage_bps_by_session.*US_PRE"):
+        validate_backtest_cost_model(model=model, required_sessions=["KRX_REG", "US_PRE"])
+
+
+def test_missing_required_failure_rate_session_raises() -> None:
+    model = BacktestCostModel(
+        commission_bps=5.0,
+        slippage_bps_by_session={"KRX_REG": 10.0, "US_PRE": 50.0},
+        failure_rate_by_session={"KRX_REG": 0.01},
+        unfavorable_fill_required=True,
+    )
+    with pytest.raises(ValueError, match="missing failure_rate_by_session.*US_PRE"):
+        validate_backtest_cost_model(model=model, required_sessions=["KRX_REG", "US_PRE"])
+
+
+def test_invalid_failure_rate_range_raises() -> None:
+    model = BacktestCostModel(
+        commission_bps=5.0,
+        slippage_bps_by_session={"KRX_REG": 10.0},
+        failure_rate_by_session={"KRX_REG": 1.2},
+        unfavorable_fill_required=True,
+    )
+    with pytest.raises(ValueError, match="failure rate must be within"):
+        validate_backtest_cost_model(model=model, required_sessions=["KRX_REG"])
+
+
+def test_unfavorable_fill_requirement_cannot_be_disabled() -> None:
+    model = BacktestCostModel(
+        commission_bps=5.0,
+        slippage_bps_by_session={"KRX_REG": 10.0},
+        failure_rate_by_session={"KRX_REG": 0.02},
+        unfavorable_fill_required=False,
+    )
+    with pytest.raises(ValueError, match="unfavorable_fill_required must be True"):
+        validate_backtest_cost_model(model=model, required_sessions=["KRX_REG"])
+
+
+@pytest.mark.parametrize("bad_commission", [float("nan"), float("inf"), float("-inf")])
+def test_non_finite_commission_rejected(bad_commission: float) -> None:
+    model = BacktestCostModel(
+        commission_bps=bad_commission,
+        slippage_bps_by_session={"KRX_REG": 10.0},
+        failure_rate_by_session={"KRX_REG": 0.02},
+        unfavorable_fill_required=True,
+    )
+    with pytest.raises(ValueError, match="commission_bps"):
+        validate_backtest_cost_model(model=model, required_sessions=["KRX_REG"])
+
+
+@pytest.mark.parametrize("bad_slippage", [float("nan"), float("inf"), float("-inf")])
+def test_non_finite_slippage_rejected(bad_slippage: float) -> None:
+    model = BacktestCostModel(
+        commission_bps=5.0,
+        slippage_bps_by_session={"KRX_REG": bad_slippage},
+        failure_rate_by_session={"KRX_REG": 0.02},
+        unfavorable_fill_required=True,
+    )
+    with pytest.raises(ValueError, match="slippage bps"):
+        validate_backtest_cost_model(model=model, required_sessions=["KRX_REG"])

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},
+            )
+        )