以下是一个强大的 Python 实现，展示了如何在自定义的 Pandas 回测系统中，对基于成交量的动态滑点进行建模。
与其对每笔交易应用一成不变且不切实际的固定惩罚，这段代码使用了一种非线性市场冲击模型（通常被称为“平方根定律”变体）。该模型会根据两个核心变量动态调整执行摩擦：
- 相对于市场流动性的订单规模： 订单规模越大，吃掉的订单簿深度就越深，遭受的滑点惩罚也就越严重。
- 历史波动率： 在市场剧烈波动期间，由于买卖价差自然拉大，滑点也会随之增加。
Python
import numpy as np
import pandas as pd
def apply_dynamic_slippage(
df: pd.DataFrame,
order_size_col: str,
market_vol_col: str,
market_return_col: str,
participation_rate: float = 0.10,
impact_exponent: float = 0.5,
) -> pd.DataFrame:
"""模拟基于成交量和波动率驱动的非线性市场冲击模型，
以便在回测执行中进行精准的惩罚。
参数:
-----------
df : pd.DataFrame
回测时间序列的 DataFrame。
order_size_col : str
代表策略在该 K 线（Bar）上计划交易的股票/单位数量的列名。（正数代表买入，负数代表卖出）。
market_vol_col : str
该 K 线上的历史市场总成交量的列名。
market_return_col : str
代表资产 volatility 的列名（例如：对数收益率的滚动标准差，或滚动 ATR 百分比）。
participation_rate : float, 默认 0.10
缩放因子，决定订单成交量对订单簿价差的冲击激进程度。
impact_exponent : float, 默认 0.5
平方根幂次因子。微观结构经验数据表明，市场冲击大致与订单规模的平方根（0.5）成正比。
返回:
--------
pd.DataFrame
增加了 'slippage_pct'（滑点百分比）和 'execution_price'（执行价格）列的 DataFrame。
"""
# 创建深拷贝以防止修改原始 DataFrame
backtest_df = df.copy()
# 计算成交量参与率: (你的订单规模 / 该 K 线市场总成交量)
# 使用绝对值，因为买单和卖单都会消耗流动性
backtest_df["volume_participation"] = backtest_df[order_size_col].abs() / (
backtest_df[market_vol_col] + 1e-8
)
# 非线性市场冲击公式:
# 滑点 % = 参与率 * (成交量参与率)^冲击指数 * 市场波动率
backtest_df["slippage_pct"] = (
participation_rate
* (backtest_df["volume_participation"] ** impact_exponent)
* backtest_df[market_return_col]
)
# 用 0 填充 NaN 值（适用于策略未发出任何订单的 K 线）
backtest_df["slippage_pct"] = backtest_df["slippage_pct"].fillna(0.0)
# 应用执行方向的约束条件:
# 买单（正订单）的成交价格要高于（HIGHER）打印出的市场价。
# 卖单（负订单）的成交价格要低于（LOWER）打印出的市场价。
backtest_df["execution_price"] = np.where(
backtest_df[order_size_col] > 0,
backtest_df["close"] * (1 + backtest_df["slippage_pct"]), # 买入惩罚
np.where(
backtest_df[order_size_col] < 0,
backtest_df["close"] * (1 - backtest_df["slippage_pct"]), # 卖出惩罚
backtest_df["close"], # 未发出交易信号
),
)
return backtest_df
# ==========================================
# 使用示例及样本数据模拟
# ==========================================
if __name__ == "__main__":
# 生成模拟的 15 分钟 K 线市场数据
np.random.seed(42)
dates = pd.date_range(start="2026-07-10 09:30", periods=5, freq="15min")
data = {
"close": [150.00, 150.50, 152.00, 151.20, 151.80],
"market_volume": [
50000,
12000,
85000,
22000,
45000,
], # 注意：第 2 根 K 线流动性极低
"rolling_volatility": [
0.012,
0.015,
0.025,
0.018,
0.011,
], # 第 3 根 K 线波动率极高
"strategy_order": [
2500,
2500,
-5000,
0,
100,
], # 正数 = 买入，负数 = 卖出
}
raw_backtest = pd.DataFrame(data, index=dates)
# 通过防御性滑点引擎进行处理
realistic_backtest = apply_dynamic_slippage(
df=raw_backtest,
order_size_col="strategy_order",
market_vol_col="market_volume",
market_return_col="rolling_volatility",
)
# 打印专注于执行偏差的结果
display_cols = [
"close",
"strategy_order",
"volume_participation",
"slippage_pct",
"execution_price",
]
print(realistic_backtest[display_cols].to_string())
代码输出逻辑详解
- 第 1 根 K 线（标准买入）： 在 50,000 的市场总成交量中买入 2,500 股，产生了 5% 的成交量参与率。最终的成交价格相比于 150.00 美元的基准价格轻微上滑至 150.40 美元。
- 第 2 根 K 线（低流动性陷阱）： 在流动性枯竭、市场总成交量仅有 12,000 股的情况下，买入完全相同的 2,500 股，导致成交量参与率飙升至接近 21%。尽管当时市场价格仅上涨了 0.50 美元，但由于缺乏可用流动性，你激进的订单属性导致执行价格被进一步大幅推高。
- 第 3 根 K 线（高波动率下的流动性横扫）： 在高波动率（0.025）的窗口期大举卖出 5,000 股。这加剧了微观结构上的业绩损耗，迫使最终的成交价被严重惩罚，远低于历史基准收盘价 152.00 美元。