diff --git a/pyproject.toml b/pyproject.toml index 1aa321c..126bd4d 100644 --- a/pyproject.toml +++ b/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" diff --git a/src/easy_tdx/backtest/engine.py b/src/easy_tdx/backtest/engine.py index 005570a..943c893 100644 --- a/src/easy_tdx/backtest/engine.py +++ b/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", ] diff --git a/src/easy_tdx/backtest/execution.py b/src/easy_tdx/backtest/execution.py new file mode 100644 index 0000000..f0e4e57 --- /dev/null +++ b/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 [] diff --git a/src/easy_tdx/backtest/orders.py b/src/easy_tdx/backtest/orders.py index 638e948..dfd2712 100644 --- a/src/easy_tdx/backtest/orders.py +++ b/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"], diff --git a/src/easy_tdx/backtest/slippage.py b/src/easy_tdx/backtest/slippage.py new file mode 100644 index 0000000..70a0dd8 --- /dev/null +++ b/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 diff --git a/tests/unit/test_backtest_engine.py b/tests/unit/test_backtest_engine.py index 55fefdc..25a39f2 100644 --- a/tests/unit/test_backtest_engine.py +++ b/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 diff --git a/tests/unit/test_backtest_execution.py b/tests/unit/test_backtest_execution.py new file mode 100644 index 0000000..0d742fe --- /dev/null +++ b/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 diff --git a/tests/unit/test_backtest_orders.py b/tests/unit/test_backtest_orders.py index 7aca070..72879da 100644 --- a/tests/unit/test_backtest_orders.py +++ b/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) diff --git a/tests/unit/test_backtest_slippage.py b/tests/unit/test_backtest_slippage.py new file mode 100644 index 0000000..b2b2e2e --- /dev/null +++ b/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)