feat(backtest): v1.14.0 — pluggable slippage models + execution simulation

- SlippageModel ABC with 4 built-in models (Fixed, Percent, SquareRoot, Volume)
- ExecutionModel ABC with 4 built-in models (Immediate, TWAP, VWAP, Limit)
- OrderSimulator integration with SlippageModel
- BacktestEngine integration with SlippageModel + ExecutionModel
- Full backward compatibility (all existing code unchanged)
- 544 tests passing (46 new)

Co-Authored-By: Claude <noreply@anthropic.com>

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "easy-tdx"
-version = "1.13.0"
+version = "1.14.0"
 description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步"
 readme = "README.md"
 requires-python = ">=3.10"

src/easy_tdx/backtest/engine.py

@@ -6,7 +6,7 @@ Coordinates Strategy → OrderSimulator → PortfolioTracker → PerformanceAnal
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import pandas as pd
 
@@ -16,6 +16,10 @@ from easy_tdx.backtest.portfolio import PortfolioTracker
 from easy_tdx.backtest.strategy import Strategy
 from easy_tdx.backtest.types import BacktestResult, Signal, Trade
 
+if TYPE_CHECKING:
+    from easy_tdx.backtest.execution import ExecutionModel
+    from easy_tdx.backtest.slippage import SlippageModel
+
 
 @dataclass
 class _StopCondition:
@@ -57,6 +61,8 @@ class BacktestEngine:
         reject_policy: str = "reduce",
         benchmark: pd.DataFrame | None = None,
         chanlun_level: str | None = None,
+        slippage_model: SlippageModel | None = None,
+        execution_model: ExecutionModel | None = None,
     ):
         """Initialize engine.
 
@@ -73,6 +79,10 @@ class BacktestEngine:
             benchmark: Benchmark data for performance comparison
             chanlun_level: Auto-compute chanlun analysis at this level
                 (e.g. 'DAILY', '30MIN'). Strategy accesses via self.chanlun.
+            slippage_model: Pluggable slippage model (overrides flat slippage
+                when provided).
+            execution_model: Pluggable execution model (bypasses OrderSimulator
+                when provided).
         """
         self._strategy_cls = strategy if isinstance(strategy, type) else type(strategy)
         self._strategy_instance = strategy if isinstance(strategy, Strategy) else None
@@ -87,6 +97,8 @@ class BacktestEngine:
         self._reject_policy = reject_policy
         self._benchmark = benchmark
         self._chanlun_level = chanlun_level
+        self._slippage_model = slippage_model
+        self._execution_model = execution_model
 
     def run(self, df: pd.DataFrame, chanlun_result: Any | None = None) -> BacktestResult:
         """Run backtest.
@@ -113,21 +125,29 @@ class BacktestEngine:
         signals = self._generate_signals(df, chanlun_result)
 
         # Step 2: Order simulation
-        simulator = OrderSimulator(
-            df,
-            execution=self._execution,
-            position_mode=self._position_mode,
-            reject_policy=self._reject_policy,
-            commission=self._commission,
-            min_commission=self._min_commission,
-            stamp_tax=self._stamp_tax,
-            slippage=self._slippage,
-        )
-        trades = simulator.simulate(
-            signals=signals,
-            cash=self._cash,
-            position=0.0,
-        )
+        if self._execution_model is not None:
+            # ExecutionModel path
+            trades = self._execute_with_model(signals, df)
+            future_leak = False
+        else:
+            # OrderSimulator path (default)
+            simulator = OrderSimulator(
+                df,
+                execution=self._execution,
+                position_mode=self._position_mode,
+                reject_policy=self._reject_policy,
+                commission=self._commission,
+                min_commission=self._min_commission,
+                stamp_tax=self._stamp_tax,
+                slippage=self._slippage,
+                slippage_model=self._slippage_model,
+            )
+            trades = simulator.simulate(
+                signals=signals,
+                cash=self._cash,
+                position=0.0,
+            )
+            future_leak = simulator.future_leak_warning
 
         # Step 3: Portfolio tracking
         trades = self._compute_pnls(trades)
@@ -149,7 +169,7 @@ class BacktestEngine:
             "execution": self._execution,
             "position_mode": self._position_mode,
             "reject_policy": self._reject_policy,
-            "future_leak_warning": simulator.future_leak_warning,
+            "future_leak_warning": future_leak,
         }
 
         return BacktestResult(
@@ -160,6 +180,57 @@ class BacktestEngine:
             config=config,
         )
 
+    def _execute_with_model(self, signals: list[Signal], df: pd.DataFrame) -> list[Trade]:
+        """Use ExecutionModel to process signals."""
+        assert self._execution_model is not None
+        all_trades: list[Trade] = []
+        cash = self._cash
+        position = 0.0
+
+        for signal in signals:
+            bar_idx = self._find_bar_index(df, signal.datetime)
+            if bar_idx is None:
+                continue
+            sub_trades = self._execution_model.execute(
+                signal=signal,
+                df=df,
+                bar_idx=bar_idx,
+                cash=cash,
+                position=position,
+                position_mode=self._position_mode,
+                commission=self._commission,
+                min_commission=self._min_commission,
+                stamp_tax=self._stamp_tax,
+                slippage_model=self._slippage_model,
+            )
+            for t in sub_trades:
+                if not t.rejected:
+                    if t.direction == "BUY":
+                        cash -= t.size * t.price + t.commission + t.slippage
+                        position += t.size
+                    else:
+                        cash += t.size * t.price - t.commission - t.slippage
+                        position -= t.size
+            all_trades.extend(sub_trades)
+        return all_trades
+
+    @staticmethod
+    def _find_bar_index(df: pd.DataFrame, datetime_val: int) -> int | None:
+        """Find bar index by datetime value."""
+        dt_col = df["datetime"]
+        try:
+            idx = (dt_col == datetime_val).idxmax() if (dt_col == datetime_val).any() else None
+            if idx is not None:
+                return int(idx)
+        except (TypeError, ValueError):
+            pass
+        if hasattr(dt_col, "dt"):
+            dt_ints = dt_col.dt.strftime("%Y%m%d").astype(int)
+            mask = dt_ints == datetime_val
+            if mask.any():
+                return int(mask.idxmax())
+        return None
+
     def _generate_signals(self, df: pd.DataFrame, chanlun_result: Any | None) -> list[Signal]:
         """Generate signals from strategy.
 
@@ -385,6 +456,7 @@ class BacktestEngine:
                     "size",
                     "price",
                     "commission",
+                    "slippage",
                     "pnl",
                     "rejected",
                 ]
@@ -397,6 +469,7 @@ class BacktestEngine:
                 "size": t.size,
                 "price": t.price,
                 "commission": t.commission,
+                "slippage": t.slippage,
                 "pnl": t.pnl,
                 "rejected": t.rejected,
             }
@@ -427,6 +500,7 @@ class BacktestEngine:
                     "size",
                     "price",
                     "commission",
+                    "slippage",
                     "pnl",
                     "rejected",
                 ]

src/easy_tdx/backtest/execution.py

@@ -0,0 +1,636 @@
+"""可插拔执行仿真引擎。"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+import numpy as np
+import pandas as pd
+
+from easy_tdx.backtest.types import Trade
+
+if TYPE_CHECKING:
+    from easy_tdx.backtest.slippage import SlippageModel
+    from easy_tdx.backtest.types import Signal
+
+
+class ExecutionModel(ABC):
+    """执行仿真基类。"""
+
+    @abstractmethod
+    def execute(
+        self,
+        signal: Signal,
+        df: pd.DataFrame,
+        bar_idx: int,
+        cash: float,
+        position: float,
+        position_mode: str,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+        slippage_model: SlippageModel | None,
+    ) -> list[Trade]:
+        """将信号转换为一笔或多笔成交。"""
+        ...
+
+    def _calc_commission(
+        self,
+        size: float,
+        price: float,
+        is_sell: bool,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+    ) -> float:
+        """计算手续费。"""
+        comm = max(size * price * commission, min_commission)
+        if is_sell:
+            comm += size * price * stamp_tax
+        return comm
+
+    def _calc_slippage(
+        self,
+        size: float,
+        price: float,
+        is_sell: bool,
+        slippage_model: SlippageModel | None,
+        df: pd.DataFrame,
+    ) -> float:
+        """计算滑点。"""
+        if slippage_model is None:
+            return 0.0
+        volume = float(df["volume"].iloc[-1]) if "volume" in df.columns else 0.0
+        volatility = self._estimate_volatility(df)
+        return slippage_model.compute(
+            price=price,
+            size=size,
+            volume=volume,
+            volatility=volatility,
+            direction="SELL" if is_sell else "BUY",
+        )
+
+    def _estimate_volatility(self, df: pd.DataFrame) -> float:
+        """从收盘价估计近期年化波动率。"""
+        if "close" not in df.columns or len(df) < 2:
+            return 0.0
+        close = df["close"].to_numpy()
+        returns = np.diff(close) / close[:-1]
+        if len(returns) < 2:
+            return 0.0
+        return float(np.std(returns)) * np.sqrt(252)
+
+    def _calc_buy_size(
+        self,
+        signal_size: float,
+        price: float,
+        cash: float,
+        position_mode: str,
+        commission: float,
+    ) -> float:
+        """计算买入数量。"""
+        if position_mode == "full" or signal_size == 0:
+            max_cost = price * (1 + commission)
+            max_shares = int(cash / max_cost / 100) * 100
+            return float(max_shares)
+        elif position_mode == "percent":
+            target_value = cash * signal_size
+            return float(int(target_value / price / 100) * 100)
+        return signal_size
+
+    def _get_datetime_int(self, df: pd.DataFrame, idx: int) -> int:
+        """获取指定 index 的 datetime int。"""
+        dt_raw = df["datetime"].iloc[idx]
+        if hasattr(dt_raw, "strftime"):
+            return int(dt_raw.strftime("%Y%m%d"))
+        return int(dt_raw)
+
+
+class ImmediateExecution(ExecutionModel):
+    """即时成交（向后兼容，与现有 OrderSimulator 行为一致）。"""
+
+    def execute(
+        self,
+        signal: Signal,
+        df: pd.DataFrame,
+        bar_idx: int,
+        cash: float,
+        position: float,
+        position_mode: str,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+        slippage_model: SlippageModel | None,
+    ) -> list[Trade]:
+        exec_idx = bar_idx + 1
+        if exec_idx >= len(df):
+            return []
+
+        price = float(df["open"].iloc[exec_idx])
+
+        if signal.direction == "BUY":
+            size = self._calc_buy_size(
+                signal.size,
+                price,
+                cash,
+                position_mode,
+                commission,
+            )
+            if size <= 0:
+                return []
+            comm = self._calc_commission(
+                size,
+                price,
+                False,
+                commission,
+                min_commission,
+                stamp_tax,
+            )
+            slip = self._calc_slippage(size, price, False, slippage_model, df)
+            return [
+                Trade(
+                    datetime=self._get_datetime_int(df, exec_idx),
+                    direction="BUY",
+                    size=size,
+                    price=price,
+                    commission=comm,
+                    slippage=slip,
+                )
+            ]
+        elif signal.direction == "SELL":
+            size = signal.size if signal.size > 0 else position
+            if size <= 0:
+                return []
+            if size > position:
+                size = position
+            comm = self._calc_commission(
+                size,
+                price,
+                True,
+                commission,
+                min_commission,
+                stamp_tax,
+            )
+            slip = self._calc_slippage(size, price, True, slippage_model, df)
+            return [
+                Trade(
+                    datetime=self._get_datetime_int(df, exec_idx),
+                    direction="SELL",
+                    size=size,
+                    price=price,
+                    commission=comm,
+                    slippage=slip,
+                )
+            ]
+
+        return []
+
+
+class TWAPExecution(ExecutionModel):
+    """时间加权平均价格执行。
+
+    将订单均匀拆分为 n_bars 份，在连续 n_bars 根 K 线上执行。
+    """
+
+    def __init__(self, n_bars: int = 5) -> None:
+        self._n_bars = max(1, n_bars)
+
+    def execute(
+        self,
+        signal: Signal,
+        df: pd.DataFrame,
+        bar_idx: int,
+        cash: float,
+        position: float,
+        position_mode: str,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+        slippage_model: SlippageModel | None,
+    ) -> list[Trade]:
+        if signal.direction == "BUY":
+            return self._execute_buy(
+                signal,
+                df,
+                bar_idx,
+                cash,
+                position_mode,
+                commission,
+                min_commission,
+                stamp_tax,
+                slippage_model,
+            )
+        return self._execute_sell(
+            signal,
+            df,
+            bar_idx,
+            position,
+            commission,
+            min_commission,
+            stamp_tax,
+            slippage_model,
+        )
+
+    def _execute_buy(
+        self,
+        signal: Signal,
+        df: pd.DataFrame,
+        bar_idx: int,
+        cash: float,
+        position_mode: str,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+        slippage_model: SlippageModel | None,
+    ) -> list[Trade]:
+        first_price = float(df["open"].iloc[bar_idx + 1]) if bar_idx + 1 < len(df) else 0
+        if first_price <= 0:
+            return []
+        total_size = self._calc_buy_size(
+            signal.size,
+            first_price,
+            cash,
+            position_mode,
+            commission,
+        )
+        if total_size <= 0:
+            return []
+
+        sub_size = int(total_size / self._n_bars / 100) * 100
+        if sub_size <= 0:
+            sub_size = 100
+
+        trades: list[Trade] = []
+        for i in range(self._n_bars):
+            exec_idx = bar_idx + 1 + i
+            if exec_idx >= len(df):
+                break
+            price = float(df["close"].iloc[exec_idx])
+            remaining = total_size - sum(t.size for t in trades)
+            actual_size = min(sub_size, remaining)
+            actual_size = int(actual_size / 100) * 100
+            if actual_size <= 0:
+                break
+            comm = self._calc_commission(
+                actual_size,
+                price,
+                False,
+                commission,
+                min_commission,
+                stamp_tax,
+            )
+            slip = self._calc_slippage(actual_size, price, False, slippage_model, df)
+            trades.append(
+                Trade(
+                    datetime=self._get_datetime_int(df, exec_idx),
+                    direction="BUY",
+                    size=float(actual_size),
+                    price=price,
+                    commission=comm,
+                    slippage=slip,
+                )
+            )
+        return trades
+
+    def _execute_sell(
+        self,
+        signal: Signal,
+        df: pd.DataFrame,
+        bar_idx: int,
+        position: float,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+        slippage_model: SlippageModel | None,
+    ) -> list[Trade]:
+        total_size = signal.size if signal.size > 0 else position
+        if total_size <= 0:
+            return []
+
+        sub_size = int(total_size / self._n_bars / 100) * 100
+        if sub_size <= 0:
+            sub_size = 100
+
+        trades: list[Trade] = []
+        for i in range(self._n_bars):
+            exec_idx = bar_idx + 1 + i
+            if exec_idx >= len(df):
+                break
+            price = float(df["close"].iloc[exec_idx])
+            remaining = total_size - sum(t.size for t in trades)
+            actual_size = min(sub_size, remaining)
+            actual_size = int(actual_size / 100) * 100
+            if actual_size <= 0:
+                break
+            comm = self._calc_commission(
+                actual_size,
+                price,
+                True,
+                commission,
+                min_commission,
+                stamp_tax,
+            )
+            slip = self._calc_slippage(actual_size, price, True, slippage_model, df)
+            trades.append(
+                Trade(
+                    datetime=self._get_datetime_int(df, exec_idx),
+                    direction="SELL",
+                    size=float(actual_size),
+                    price=price,
+                    commission=comm,
+                    slippage=slip,
+                )
+            )
+        return trades
+
+
+class VWAPExecution(ExecutionModel):
+    """成交量加权平均价格执行。
+
+    按历史成交量分布比例拆分订单。
+    """
+
+    def __init__(self, n_bars: int = 5, volume_lookback: int = 20) -> None:
+        self._n_bars = max(1, n_bars)
+        self._volume_lookback = max(1, volume_lookback)
+
+    def execute(
+        self,
+        signal: Signal,
+        df: pd.DataFrame,
+        bar_idx: int,
+        cash: float,
+        position: float,
+        position_mode: str,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+        slippage_model: SlippageModel | None,
+    ) -> list[Trade]:
+        if signal.direction == "BUY":
+            return self._execute_buy(
+                signal,
+                df,
+                bar_idx,
+                cash,
+                position_mode,
+                commission,
+                min_commission,
+                stamp_tax,
+                slippage_model,
+            )
+        return self._execute_sell(
+            signal,
+            df,
+            bar_idx,
+            position,
+            commission,
+            min_commission,
+            stamp_tax,
+            slippage_model,
+        )
+
+    def _get_volume_weights(self, df: pd.DataFrame, bar_idx: int) -> list[float]:
+        """获取成交量权重分布。"""
+        start = max(0, bar_idx - self._volume_lookback + 1)
+        lookback = df.iloc[start : bar_idx + 1]
+        if "volume" not in lookback.columns or len(lookback) == 0:
+            return [1.0 / self._n_bars] * self._n_bars
+
+        volumes = lookback["volume"].to_numpy()
+        total_vol = float(volumes.sum())
+        if total_vol <= 0:
+            return [1.0 / self._n_bars] * self._n_bars
+
+        weights: list[float] = []
+        for i in range(self._n_bars):
+            idx = max(0, len(volumes) - 1 - (i % max(1, len(volumes))))
+            weights.append(float(volumes[idx]) / total_vol)
+        total_w = sum(weights)
+        if total_w <= 0:
+            return [1.0 / self._n_bars] * self._n_bars
+        return [w / total_w for w in weights]
+
+    def _execute_buy(
+        self,
+        signal: Signal,
+        df: pd.DataFrame,
+        bar_idx: int,
+        cash: float,
+        position_mode: str,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+        slippage_model: SlippageModel | None,
+    ) -> list[Trade]:
+        first_price = float(df["open"].iloc[bar_idx + 1]) if bar_idx + 1 < len(df) else 0
+        if first_price <= 0:
+            return []
+        total_size = self._calc_buy_size(
+            signal.size,
+            first_price,
+            cash,
+            position_mode,
+            commission,
+        )
+        if total_size <= 0:
+            return []
+
+        weights = self._get_volume_weights(df, bar_idx)
+        trades: list[Trade] = []
+        for i in range(self._n_bars):
+            exec_idx = bar_idx + 1 + i
+            if exec_idx >= len(df):
+                break
+            price = float(df["close"].iloc[exec_idx])
+            w = weights[i] if i < len(weights) else 1.0 / self._n_bars
+            target = int(total_size * w / 100) * 100
+            remaining = total_size - sum(t.size for t in trades)
+            actual_size = min(target, remaining)
+            actual_size = int(actual_size / 100) * 100
+            if actual_size <= 0:
+                continue
+            comm = self._calc_commission(
+                actual_size,
+                price,
+                False,
+                commission,
+                min_commission,
+                stamp_tax,
+            )
+            slip = self._calc_slippage(actual_size, price, False, slippage_model, df)
+            trades.append(
+                Trade(
+                    datetime=self._get_datetime_int(df, exec_idx),
+                    direction="BUY",
+                    size=float(actual_size),
+                    price=price,
+                    commission=comm,
+                    slippage=slip,
+                )
+            )
+        return trades
+
+    def _execute_sell(
+        self,
+        signal: Signal,
+        df: pd.DataFrame,
+        bar_idx: int,
+        position: float,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+        slippage_model: SlippageModel | None,
+    ) -> list[Trade]:
+        total_size = signal.size if signal.size > 0 else position
+        if total_size <= 0:
+            return []
+
+        weights = self._get_volume_weights(df, bar_idx)
+        trades: list[Trade] = []
+        for i in range(self._n_bars):
+            exec_idx = bar_idx + 1 + i
+            if exec_idx >= len(df):
+                break
+            price = float(df["close"].iloc[exec_idx])
+            w = weights[i] if i < len(weights) else 1.0 / self._n_bars
+            target = int(total_size * w / 100) * 100
+            remaining = total_size - sum(t.size for t in trades)
+            actual_size = min(target, remaining)
+            actual_size = int(actual_size / 100) * 100
+            if actual_size <= 0:
+                continue
+            comm = self._calc_commission(
+                actual_size,
+                price,
+                True,
+                commission,
+                min_commission,
+                stamp_tax,
+            )
+            slip = self._calc_slippage(actual_size, price, True, slippage_model, df)
+            trades.append(
+                Trade(
+                    datetime=self._get_datetime_int(df, exec_idx),
+                    direction="SELL",
+                    size=float(actual_size),
+                    price=price,
+                    commission=comm,
+                    slippage=slip,
+                )
+            )
+        return trades
+
+
+class LimitExecution(ExecutionModel):
+    """限价单执行。
+
+    在目标价位挂单，仅当 bar_low <= price（买入）或 bar_high >= price（卖出）时成交。
+    无限价时退化为 ImmediateExecution。
+    """
+
+    def __init__(self, ttl_bars: int = 5) -> None:
+        self._ttl_bars = max(1, ttl_bars)
+        self._fallback = ImmediateExecution()
+
+    def execute(
+        self,
+        signal: Signal,
+        df: pd.DataFrame,
+        bar_idx: int,
+        cash: float,
+        position: float,
+        position_mode: str,
+        commission: float,
+        min_commission: float,
+        stamp_tax: float,
+        slippage_model: SlippageModel | None,
+    ) -> list[Trade]:
+        if signal.price is None:
+            return self._fallback.execute(
+                signal,
+                df,
+                bar_idx,
+                cash,
+                position,
+                position_mode,
+                commission,
+                min_commission,
+                stamp_tax,
+                slippage_model,
+            )
+
+        target_price = signal.price
+
+        for i in range(self._ttl_bars):
+            exec_idx = bar_idx + 1 + i
+            if exec_idx >= len(df):
+                break
+            row = df.iloc[exec_idx]
+            triggered = False
+            if signal.direction == "BUY" and float(row["low"]) <= target_price:
+                triggered = True
+            elif signal.direction == "SELL" and float(row["high"]) >= target_price:
+                triggered = True
+
+            if triggered:
+                if signal.direction == "BUY":
+                    size = self._calc_buy_size(
+                        signal.size,
+                        target_price,
+                        cash,
+                        position_mode,
+                        commission,
+                    )
+                    if size <= 0:
+                        return []
+                    comm = self._calc_commission(
+                        size,
+                        target_price,
+                        False,
+                        commission,
+                        min_commission,
+                        stamp_tax,
+                    )
+                    slip = self._calc_slippage(
+                        size,
+                        target_price,
+                        False,
+                        slippage_model,
+                        df,
+                    )
+                else:
+                    size = signal.size if signal.size > 0 else position
+                    if size <= 0:
+                        return []
+                    if size > position:
+                        size = position
+                    comm = self._calc_commission(
+                        size,
+                        target_price,
+                        True,
+                        commission,
+                        min_commission,
+                        stamp_tax,
+                    )
+                    slip = self._calc_slippage(
+                        size,
+                        target_price,
+                        True,
+                        slippage_model,
+                        df,
+                    )
+
+                return [
+                    Trade(
+                        datetime=self._get_datetime_int(df, exec_idx),
+                        direction=signal.direction,
+                        size=float(size),
+                        price=target_price,
+                        commission=comm,
+                        slippage=slip,
+                    )
+                ]
+
+        return []

src/easy_tdx/backtest/orders.py

@@ -6,11 +6,16 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from typing import TYPE_CHECKING
 
+import numpy as np
 import pandas as pd
 
 from easy_tdx.backtest.types import Signal, Trade
 
+if TYPE_CHECKING:
+    from easy_tdx.backtest.slippage import SlippageModel
+
 
 @dataclass
 class OrderSimulator:
@@ -39,6 +44,7 @@ class OrderSimulator:
     min_commission: float = 5.0
     stamp_tax: float = 0.001
     slippage: float = 0.0
+    slippage_model: SlippageModel | None = None
     future_leak_warning: bool = False
 
     def simulate(
@@ -288,6 +294,37 @@ class OrderSimulator:
 
         return commission
 
+    def _compute_slippage(self, size: float, price: float, is_sell: bool) -> float:
+        """计算滑点成本。"""
+        if self.slippage_model is not None:
+            volume = self._get_current_volume()
+            volatility = self._estimate_volatility()
+            return self.slippage_model.compute(
+                price=price,
+                size=size,
+                volume=volume,
+                volatility=volatility,
+                direction="SELL" if is_sell else "BUY",
+            )
+        return size * self.slippage
+
+    def _get_current_volume(self) -> float:
+        """获取最后一根K线的成交量。"""
+        if "volume" in self.df.columns and len(self.df) > 0:
+            return float(self.df["volume"].iloc[-1])
+        return 0.0
+
+    def _estimate_volatility(self) -> float:
+        """从收盘价估计年化波动率。"""
+        if "close" not in self.df.columns or len(self.df) < 2:
+            return 0.0
+        close = self.df["close"].to_numpy()
+        returns = np.diff(close) / close[:-1]
+        if len(returns) < 2:
+            return 0.0
+        daily_vol = float(np.std(returns))
+        return daily_vol * np.sqrt(252)
+
     def _execute_buy(
         self,
         signal: Signal,
@@ -338,7 +375,7 @@ class OrderSimulator:
 
         # 计算费用
         commission = self._calculate_commission(size, price, is_sell=False)
-        slippage = size * self.slippage
+        slippage = self._compute_slippage(size, price, is_sell=False)
 
         # 检查资金是否足够
         total_cost = size * price + commission + slippage
@@ -361,7 +398,7 @@ class OrderSimulator:
                     reduced_size = int(available_cash / price / 100) * 100
                     if reduced_size > 0:
                         commission = self._calculate_commission(reduced_size, price, is_sell=False)
-                        slippage = reduced_size * self.slippage
+                        slippage = self._compute_slippage(reduced_size, price, is_sell=False)
                         return Trade(
                             datetime=self.df.iloc[exec_idx]["datetime"],
                             direction="BUY",
@@ -446,7 +483,7 @@ class OrderSimulator:
 
         # 计算费用
         commission = self._calculate_commission(size, price, is_sell=True)
-        slippage = size * self.slippage
+        slippage = self._compute_slippage(size, price, is_sell=True)
 
         return Trade(
             datetime=self.df.iloc[exec_idx]["datetime"],

src/easy_tdx/backtest/slippage.py

@@ -0,0 +1,129 @@
+"""可插拔滑点模型。"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+import numpy as np
+
+
+class SlippageModel(ABC):
+    """滑点模型基类。
+
+    所有滑点模型必须实现 compute() 方法，返回总滑点成本（金额）。
+    """
+
+    @abstractmethod
+    def compute(
+        self,
+        price: float,
+        size: float,
+        volume: float,
+        volatility: float,
+        direction: str,
+    ) -> float:
+        """计算滑点成本。
+
+        Args:
+            price: 成交价格
+            size: 订单数量（股）
+            volume: 当日成交量（股），0 表示无数据
+            volatility: 近期年化波动率，0 表示无数据
+            direction: 交易方向 BUY / SELL
+
+        Returns:
+            总滑点成本（金额，非比率）
+        """
+        ...
+
+
+class FixedSlippage(SlippageModel):
+    """固定每股滑点（向后兼容）。"""
+
+    def __init__(self, per_share: float = 0.01) -> None:
+        self._per_share = per_share
+
+    def compute(
+        self,
+        price: float,
+        size: float,
+        volume: float,
+        volatility: float,
+        direction: str,
+    ) -> float:
+        return size * self._per_share
+
+
+class PercentSlippage(SlippageModel):
+    """按成交金额百分比滑点。"""
+
+    def __init__(self, rate: float = 0.001) -> None:
+        self._rate = rate
+
+    def compute(
+        self,
+        price: float,
+        size: float,
+        volume: float,
+        volatility: float,
+        direction: str,
+    ) -> float:
+        return price * size * self._rate
+
+
+class SquareRootSlippage(SlippageModel):
+    """方根市场冲击模型（Almgren-Chriss 简化版）。
+
+    impact = σ × √(participation_rate) × price × size × impact_coeff
+
+    当 volume=0 或 volatility=0 时退化为 PercentSlippage(rate=0.001)。
+    """
+
+    def __init__(self, impact_coeff: float = 0.1) -> None:
+        self._impact_coeff = impact_coeff
+        self._fallback = PercentSlippage(rate=0.001)
+
+    def compute(
+        self,
+        price: float,
+        size: float,
+        volume: float,
+        volatility: float,
+        direction: str,
+    ) -> float:
+        if size <= 0:
+            return 0.0
+        if volume <= 0 or volatility <= 0:
+            return self._fallback.compute(price, size, volume, volatility, direction)
+        participation_rate = min(size / volume, 1.0)
+        impact = volatility * np.sqrt(participation_rate) * price * size * self._impact_coeff
+        return float(impact)
+
+
+class VolumeSlippage(SlippageModel):
+    """成交量比例滑点。
+
+    cost = (base_bps / 10000) × (size / volume) × price × size
+
+    当 volume=0 时退化为 PercentSlippage(rate=base_bps/10000)。
+    """
+
+    def __init__(self, base_bps: float = 10.0) -> None:
+        self._base_bps = base_bps
+        self._fallback = PercentSlippage(rate=base_bps / 10000.0)
+
+    def compute(
+        self,
+        price: float,
+        size: float,
+        volume: float,
+        volatility: float,
+        direction: str,
+    ) -> float:
+        if size <= 0:
+            return 0.0
+        if volume <= 0:
+            return self._fallback.compute(price, size, volume, volatility, direction)
+        rate = self._base_bps / 10000.0
+        participation = min(size / volume, 1.0)
+        return rate * participation * price * size

tests/unit/test_backtest_engine.py

@@ -7,6 +7,8 @@ import pandas as pd
 
 from easy_tdx import MyTT
 from easy_tdx.backtest.engine import BacktestEngine
+from easy_tdx.backtest.execution import TWAPExecution
+from easy_tdx.backtest.slippage import FixedSlippage
 from easy_tdx.backtest.strategy import Strategy
 
 
@@ -552,3 +554,73 @@ def test_chanlun_manual_result_overrides_auto():
 
     # Strategy should have received the manual result, not auto-computed one
     assert CheckerStrategy.received == manual_result
+
+
+# ── SlippageModel + ExecutionModel Integration ───────────────────────────────
+
+
+class TestEngineSlippageModel:
+    """BacktestEngine with SlippageModel integration."""
+
+    def test_engine_with_slippage_model(self) -> None:
+        """Engine uses SlippageModel."""
+
+        class SimpleBuy(Strategy):
+            def init(self) -> None:
+                pass
+
+            def next(self) -> None:
+                if self._bar_index == 0:
+                    self.buy(size=100)
+
+        df = _make_df(20)
+        engine = BacktestEngine(
+            SimpleBuy,
+            cash=100000,
+            slippage_model=FixedSlippage(per_share=0.05),
+        )
+        result = engine.run(df)
+        buy_trades = result.trades[result.trades["direction"] == "BUY"]
+        if len(buy_trades) > 0:
+            assert buy_trades.iloc[0]["slippage"] > 0
+
+
+class TestEngineExecutionModel:
+    """BacktestEngine with ExecutionModel integration."""
+
+    def test_engine_with_twap(self) -> None:
+        """Engine uses TWAP execution."""
+
+        class SimpleBuy(Strategy):
+            def init(self) -> None:
+                pass
+
+            def next(self) -> None:
+                if self._bar_index == 0:
+                    self.buy(size=300)
+
+        df = _make_df(20)
+        engine = BacktestEngine(
+            SimpleBuy,
+            cash=100000,
+            execution_model=TWAPExecution(n_bars=3),
+        )
+        result = engine.run(df)
+        buy_trades = result.trades[result.trades["direction"] == "BUY"]
+        assert len(buy_trades) >= 1
+
+    def test_engine_backward_compatible(self) -> None:
+        """No new params: behavior unchanged."""
+
+        class SimpleBuy(Strategy):
+            def init(self) -> None:
+                pass
+
+            def next(self) -> None:
+                if self._bar_index == 0:
+                    self.buy(size=100)
+
+        df = _make_df(20)
+        engine = BacktestEngine(SimpleBuy, cash=100000)
+        result = engine.run(df)
+        assert len(result.trades) >= 1

tests/unit/test_backtest_execution.py

@@ -0,0 +1,420 @@
+"""执行仿真引擎单元测试。"""
+
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+from easy_tdx.backtest.execution import (
+    ExecutionModel,
+    ImmediateExecution,
+    LimitExecution,
+    TWAPExecution,
+    VWAPExecution,
+)
+from easy_tdx.backtest.types import Signal
+
+
+def _make_df(n: int = 20) -> pd.DataFrame:
+    """构造测试用K线数据。"""
+    data = {
+        "datetime": [20240101 + i for i in range(n)],
+        "open": [100.0 + i for i in range(n)],
+        "close": [101.0 + i for i in range(n)],
+        "high": [102.0 + i for i in range(n)],
+        "low": [99.0 + i for i in range(n)],
+        "volume": [10000] * n,
+    }
+    return pd.DataFrame(data)
+
+
+class TestExecutionBase:
+    """基类验证。"""
+
+    def test_cannot_instantiate_abc(self) -> None:
+        with pytest.raises(TypeError):
+            ExecutionModel()  # type: ignore[abstract]
+
+
+class TestImmediateExecution:
+    """即时成交。"""
+
+    def test_buy_signal(self) -> None:
+        df = _make_df(10)
+        model = ImmediateExecution()
+        signal = Signal(datetime=20240101, direction="BUY", size=100)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=20000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 1
+        assert trades[0].direction == "BUY"
+        assert trades[0].price == 101.0
+
+    def test_sell_signal(self) -> None:
+        df = _make_df(10)
+        model = ImmediateExecution()
+        signal = Signal(datetime=20240101, direction="SELL", size=100)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=0,
+            position=200,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 1
+        assert trades[0].direction == "SELL"
+
+    def test_signal_at_last_bar(self) -> None:
+        df = _make_df(10)
+        model = ImmediateExecution()
+        signal = Signal(datetime=20240109, direction="BUY", size=100)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=9,
+            cash=20000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 0
+
+    def test_with_slippage_model(self) -> None:
+        from easy_tdx.backtest.slippage import FixedSlippage
+
+        df = _make_df(10)
+        model = ImmediateExecution()
+        signal = Signal(datetime=20240101, direction="BUY", size=100)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=20000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=FixedSlippage(per_share=0.01),
+        )
+        assert len(trades) == 1
+        assert trades[0].slippage == pytest.approx(1.0)
+
+    def test_commission_on_buy(self) -> None:
+        df = _make_df(10)
+        model = ImmediateExecution()
+        signal = Signal(datetime=20240101, direction="BUY", size=100)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=20000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 1
+        assert trades[0].commission >= 5.0
+
+    def test_stamp_tax_on_sell(self) -> None:
+        df = _make_df(10)
+        model = ImmediateExecution()
+        signal = Signal(datetime=20240101, direction="SELL", size=100)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=0,
+            position=200,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 1
+        assert trades[0].commission > 10.0
+
+    def test_full_position_buy(self) -> None:
+        df = _make_df(10)
+        model = ImmediateExecution()
+        signal = Signal(datetime=20240101, direction="BUY", size=0)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=20000,
+            position=0,
+            position_mode="full",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 1
+        assert trades[0].size == 100
+
+
+class TestTWAPExecution:
+    """时间加权平均价格执行。"""
+
+    def test_split_buy_into_3_bars(self) -> None:
+        df = _make_df(20)
+        model = TWAPExecution(n_bars=3)
+        signal = Signal(datetime=20240101, direction="BUY", size=300)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=100000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 3
+        total_size = sum(t.size for t in trades)
+        assert total_size <= 300
+        prices = [t.price for t in trades]
+        assert prices[0] != prices[1]
+
+    def test_split_sell_into_2_bars(self) -> None:
+        df = _make_df(20)
+        model = TWAPExecution(n_bars=2)
+        signal = Signal(datetime=20240101, direction="SELL", size=200)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=0,
+            position=500,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 2
+        assert sum(t.size for t in trades) == 200.0
+
+    def test_truncates_at_data_end(self) -> None:
+        df = _make_df(5)
+        model = TWAPExecution(n_bars=10)
+        signal = Signal(datetime=20240101, direction="BUY", size=1000)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=100000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) <= 4
+
+    def test_full_position_mode(self) -> None:
+        df = _make_df(20)
+        model = TWAPExecution(n_bars=3)
+        signal = Signal(datetime=20240101, direction="BUY", size=0)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=60000,
+            position=0,
+            position_mode="full",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 3
+        assert all(t.size > 0 for t in trades)
+
+
+class TestVWAPExecution:
+    """成交量加权平均价格执行。"""
+
+    def test_basic_buy(self) -> None:
+        df = _make_df(20)
+        model = VWAPExecution(n_bars=3, volume_lookback=10)
+        signal = Signal(datetime=20240101, direction="BUY", size=300)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=5,
+            cash=100000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 3
+        total_size = sum(t.size for t in trades)
+        assert total_size <= 300
+
+    def test_volume_weighted_split(self) -> None:
+        df = _make_df(20)
+        df.loc[6, "volume"] = 50000
+        df.loc[7, "volume"] = 50000
+        df.loc[8, "volume"] = 50000
+        model = VWAPExecution(n_bars=3, volume_lookback=5)
+        signal = Signal(datetime=20240105, direction="BUY", size=300)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=5,
+            cash=100000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 3
+        sizes = [t.size for t in trades]
+        assert sum(sizes) <= 300
+
+    def test_truncates_at_data_end(self) -> None:
+        df = _make_df(5)
+        model = VWAPExecution(n_bars=10, volume_lookback=3)
+        signal = Signal(datetime=20240101, direction="BUY", size=1000)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=100000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) <= 4
+
+
+class TestLimitExecution:
+    """限价单执行。"""
+
+    def test_buy_limit_filled(self) -> None:
+        df = _make_df(20)
+        model = LimitExecution(ttl_bars=5)
+        signal = Signal(datetime=20240101, direction="BUY", size=100, price=100.0)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=20000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 1
+        assert trades[0].price == 100.0
+        assert trades[0].direction == "BUY"
+
+    def test_sell_limit_filled(self) -> None:
+        df = _make_df(20)
+        model = LimitExecution(ttl_bars=5)
+        signal = Signal(datetime=20240101, direction="SELL", size=100, price=105.0)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=0,
+            position=200,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 1
+        assert trades[0].price == 105.0
+
+    def test_limit_not_triggered(self) -> None:
+        df = _make_df(10)
+        model = LimitExecution(ttl_bars=3)
+        signal = Signal(datetime=20240101, direction="BUY", size=100, price=50.0)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=20000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 0
+
+    def test_no_price_falls_back_to_immediate(self) -> None:
+        df = _make_df(10)
+        model = LimitExecution(ttl_bars=5)
+        signal = Signal(datetime=20240101, direction="BUY", size=100, price=None)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=20000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 1
+        assert trades[0].price == 101.0
+
+    def test_ttl_expires(self) -> None:
+        df = _make_df(20)
+        model = LimitExecution(ttl_bars=2)
+        signal = Signal(datetime=20240101, direction="BUY", size=100, price=98.0)
+        trades = model.execute(
+            signal=signal,
+            df=df,
+            bar_idx=0,
+            cash=20000,
+            position=0,
+            position_mode="fixed",
+            commission=0.0003,
+            min_commission=5.0,
+            stamp_tax=0.001,
+            slippage_model=None,
+        )
+        assert len(trades) == 0

tests/unit/test_backtest_orders.py

@@ -3,8 +3,10 @@
 from __future__ import annotations
 
 import pandas as pd
+import pytest
 
 from easy_tdx.backtest.orders import OrderSimulator
+from easy_tdx.backtest.slippage import FixedSlippage, PercentSlippage
 from easy_tdx.backtest.types import Signal
 
 # ── Test Fixtures ─────────────────────────────────────────────────────────────
@@ -390,3 +392,75 @@ class TestEdgeCases:
         # 简化：只验证成交记录
         assert len(trades) == 1
         assert trades[0].size == 100
+
+
+# ── Test SlippageModel Integration ─────────────────────────────────────────────
+
+
+class TestSlippageModelIntegration:
+    """测试 OrderSimulator 与 SlippageModel 集成。"""
+
+    def test_fixed_slippage_model(self) -> None:
+        """FixedSlippage 与旧 slippage 参数等价。"""
+        df = _make_df(10)
+        sim = OrderSimulator(
+            df,
+            execution="next_open",
+            slippage_model=FixedSlippage(per_share=0.01),
+        )
+        signals = [_buy_signal(0, size=100)]
+        trades = sim.simulate(signals, cash=20000, position=0)
+        assert len(trades) == 1
+        assert trades[0].slippage == pytest.approx(1.0)
+
+    def test_percent_slippage_model(self) -> None:
+        """PercentSlippage 计算。"""
+        df = _make_df(10)
+        sim = OrderSimulator(
+            df,
+            execution="next_open",
+            slippage_model=PercentSlippage(rate=0.001),
+        )
+        signals = [_buy_signal(0, size=100)]
+        trades = sim.simulate(signals, cash=20000, position=0)
+        assert len(trades) == 1
+        # price=101 (next_open), 101 × 100 × 0.001 = 10.1
+        assert trades[0].slippage == pytest.approx(10.1)
+
+    def test_slippage_model_overrides_slippage_param(self) -> None:
+        """slippage_model 优先于 slippage 参数。"""
+        df = _make_df(10)
+        sim = OrderSimulator(
+            df,
+            execution="next_open",
+            position_mode="fixed",
+            slippage=999.0,
+            slippage_model=FixedSlippage(per_share=0.01),
+        )
+        signals = [_buy_signal(0, size=100)]
+        trades = sim.simulate(signals, cash=20000, position=0)
+        assert len(trades) == 1
+        assert trades[0].slippage == pytest.approx(1.0)
+
+    def test_sell_with_slippage_model(self) -> None:
+        """卖出时也使用滑点模型。"""
+        df = _make_df(10)
+        sim = OrderSimulator(
+            df,
+            execution="next_open",
+            slippage_model=FixedSlippage(per_share=0.02),
+        )
+        signals = [_sell_signal(0, size=100)]
+        trades = sim.simulate(signals, cash=0, position=100)
+        assert len(trades) == 1
+        # position_mode=full, size=0 → sell all position=100
+        assert trades[0].slippage == pytest.approx(2.0)
+
+    def test_no_slippage_model_uses_old_param(self) -> None:
+        """不提供 model 时使用旧 slippage 参数。"""
+        df = _make_df(10)
+        sim = OrderSimulator(df, execution="next_open", slippage=0.05)
+        signals = [_buy_signal(0, size=100)]
+        trades = sim.simulate(signals, cash=20000, position=0)
+        assert len(trades) == 1
+        assert trades[0].slippage == pytest.approx(5.0)

tests/unit/test_backtest_slippage.py

@@ -0,0 +1,161 @@
+"""滑点模型单元测试。"""
+
+from __future__ import annotations
+
+import pytest
+
+from easy_tdx.backtest.slippage import (
+    FixedSlippage,
+    PercentSlippage,
+    SlippageModel,
+    SquareRootSlippage,
+    VolumeSlippage,
+)
+
+
+class TestSlippageBase:
+    """基类验证。"""
+
+    def test_cannot_instantiate_abc(self) -> None:
+        """不能直接实例化 ABC。"""
+        with pytest.raises(TypeError):
+            SlippageModel()  # type: ignore[abstract]
+
+    def test_subclass_must_implement_compute(self) -> None:
+        """子类必须实现 compute。"""
+
+        class BadModel(SlippageModel):
+            pass
+
+        with pytest.raises(TypeError):
+            BadModel()  # type: ignore[abstract]
+
+
+class TestFixedSlippage:
+    """固定每股滑点。"""
+
+    def test_zero_per_share(self) -> None:
+        """per_share=0 时无滑点。"""
+        model = FixedSlippage(per_share=0.0)
+        cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY")
+        assert cost == 0.0
+
+    def test_basic(self) -> None:
+        """基本计算：100 股 × 0.01 元/股 = 1.0。"""
+        model = FixedSlippage(per_share=0.01)
+        cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY")
+        assert cost == pytest.approx(1.0)
+
+    def test_large_size(self) -> None:
+        """大单。"""
+        model = FixedSlippage(per_share=0.05)
+        cost = model.compute(
+            price=50.0, size=10000, volume=500000, volatility=0.2, direction="SELL"
+        )
+        assert cost == pytest.approx(500.0)
+
+    def test_direction_irrelevant(self) -> None:
+        """方向不影响固定滑点。"""
+        model = FixedSlippage(per_share=0.01)
+        buy_cost = model.compute(
+            price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY"
+        )
+        sell_cost = model.compute(
+            price=10.0, size=100, volume=10000, volatility=0.3, direction="SELL"
+        )
+        assert buy_cost == sell_cost
+
+
+class TestPercentSlippage:
+    """按成交金额百分比滑点。"""
+
+    def test_zero_rate(self) -> None:
+        """rate=0 时无滑点。"""
+        model = PercentSlippage(rate=0.0)
+        cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY")
+        assert cost == 0.0
+
+    def test_basic(self) -> None:
+        """10元 × 100股 × 0.001 = 1.0。"""
+        model = PercentSlippage(rate=0.001)
+        cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY")
+        assert cost == pytest.approx(1.0)
+
+    def test_high_price(self) -> None:
+        """高价股。"""
+        model = PercentSlippage(rate=0.002)
+        cost = model.compute(price=100.0, size=500, volume=20000, volatility=0.25, direction="BUY")
+        # 100 × 500 × 0.002 = 100.0
+        assert cost == pytest.approx(100.0)
+
+
+class TestSquareRootSlippage:
+    """方根市场冲击模型。"""
+
+    def test_zero_size(self) -> None:
+        """size=0 时无冲击。"""
+        model = SquareRootSlippage(impact_coeff=0.1)
+        cost = model.compute(price=10.0, size=0, volume=10000, volatility=0.3, direction="BUY")
+        assert cost == 0.0
+
+    def test_small_participation_rate(self) -> None:
+        """低参与率（小单），冲击成本低。"""
+        model = SquareRootSlippage(impact_coeff=0.1)
+        # size=100, volume=1000000, participation_rate=0.0001
+        cost = model.compute(
+            price=10.0, size=100, volume=1_000_000, volatility=0.3, direction="BUY"
+        )
+        # σ=0.3, √(0.0001)=0.01, impact = 0.3 × 0.01 × 10 × 100 × 0.1 = 0.3
+        assert cost == pytest.approx(0.3)
+
+    def test_high_participation_rate(self) -> None:
+        """高参与率（大单），冲击成本高。"""
+        model = SquareRootSlippage(impact_coeff=0.1)
+        cost = model.compute(
+            price=10.0, size=100_000, volume=200_000, volatility=0.3, direction="BUY"
+        )
+        small_cost = model.compute(
+            price=10.0, size=100, volume=1_000_000, volatility=0.3, direction="BUY"
+        )
+        assert cost > small_cost * 10
+
+    def test_zero_volume_fallback(self) -> None:
+        """volume=0 时退化为 PercentSlippage(rate=0.001)。"""
+        model = SquareRootSlippage(impact_coeff=0.1)
+        cost = model.compute(price=10.0, size=100, volume=0, volatility=0.3, direction="BUY")
+        assert cost == pytest.approx(1.0)
+
+    def test_zero_volatility_fallback(self) -> None:
+        """volatility=0 时退化为 PercentSlippage(rate=0.001)。"""
+        model = SquareRootSlippage(impact_coeff=0.1)
+        cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.0, direction="BUY")
+        assert cost == pytest.approx(1.0)
+
+
+class TestVolumeSlippage:
+    """成交量比例滑点。"""
+
+    def test_zero_size(self) -> None:
+        model = VolumeSlippage(base_bps=10.0)
+        cost = model.compute(price=10.0, size=0, volume=10000, volatility=0.3, direction="BUY")
+        assert cost == 0.0
+
+    def test_basic(self) -> None:
+        model = VolumeSlippage(base_bps=10.0)
+        cost = model.compute(price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY")
+        assert cost == pytest.approx(0.01)
+
+    def test_high_participation(self) -> None:
+        model = VolumeSlippage(base_bps=10.0)
+        cost_high = model.compute(
+            price=10.0, size=5000, volume=10000, volatility=0.3, direction="BUY"
+        )
+        cost_low = model.compute(
+            price=10.0, size=100, volume=10000, volatility=0.3, direction="BUY"
+        )
+        assert cost_high > cost_low
+
+    def test_zero_volume_fallback(self) -> None:
+        model = VolumeSlippage(base_bps=10.0)
+        cost = model.compute(price=10.0, size=100, volume=0, volatility=0.3, direction="BUY")
+        assert cost == pytest.approx(1.0)