回测收益为何总在实盘中消失:滑点、延迟与容量
你盯着屏幕上的回测曲线,满心欢喜。年化收益 23%,夏普比率 3.2,最大回撤只有 8%。这个策略,简直是印钞机。
三个月后,你的实盘账户显示年化收益 11%。夏普比率 1.4。最大回撤 22%。
钱去哪了?
这不是你的策略失效了。这是你的回测从一开始就是一场精心编排的幻觉——每一个被你忽略的“微小偏差”,在实盘中都会被复利的显微镜放大成鸿沟。
本文拆解回测与实盘之间三道最致命的裂缝:滑点模型缺失、延迟成本无视、容量约束盲区。并给出生产级的回测补偿框架,让你的回测结果不再是海市蜃楼。
一、那张从未存在过的成交单
让我们从一个最常见的回测代码片段开始:
# 某策略的回测执行模块(问题版本)
def execute_order(symbol, direction, quantity, bar):
"""
简化回测:假设以当前 bar 的收盘价成交
"""
price = bar['close']
if direction == 'BUY':
fill_price = price
else:
fill_price = price
return {
'symbol': symbol,
'price': fill_price,
'quantity': quantity,
'commission': quantity * 0.001
}
这段代码的错误在于:它假设你能以当前看到的价格成交。
在真实市场中,这个假设在任何高频率或高波动场景下都不成立。想象你是一个买家,站在拍卖行里举牌——当你举牌的那一刻,拍卖师已经看到了所有买家的出价,在你真正落槌之前,价格可能已经变了三次。
这就是回测与实盘的根本矛盾:回测是向后看的模拟,实盘是向前冲的现实。
二、裂缝一:滑点模型——被低估的隐性杀手
2.1 什么是滑点
滑点(Slippage)是你的成交价与预期价之间的差值。在订单簿视角下,这是你的订单“穿透”订单簿深度时产生的额外成本:
预期成交价 = 盘口卖一价(做多)或盘口买一价(做空)
实际成交价 = 盘口卖一价 + 订单簿穿透成本(做多)
或
盘口买一价 - 订单簿穿透成本(做空)
2.2 滑点的量化来源
滑点不是凭空产生的,它有三个主要来源:
| 来源 | 描述 | 典型量级 |
|---|---|---|
| 价格冲击 | 大单消耗流动性,导致价格向不利方向移动 | 千分之几到百分之几 |
| 买卖价差 | 你的成交价必然落在买卖价差之间的某个位置 | 通常万分之几 |
| 延迟漂移 | 从信号产生到订单到达交易所之间,价格已移动 | 高频场景可达百分之几 |
2.3 生产级滑点模型
一个可信的回测系统必须包含显式的滑点模型。以下是考虑订单簿深度和成交量的滑点估算:
import numpy as np
from dataclasses import dataclass
from typing import Optional
import os
@dataclass
class OrderBookSnapshot:
"""简化订单簿快照"""
bid_prices: np.ndarray # 买价由高到低
ask_prices: np.ndarray # 卖价由低到高
bid_volumes: np.ndarray # 各档买量
ask_volumes: np.ndarray # 各档卖量
spread: float # 买卖价差
@dataclass
class SlippageResult:
"""滑点计算结果"""
avg_fill_price: float
slippage_bps: float # 滑点(基点)
fully_filled: bool # 是否完全成交
def estimate_slippage(
snapshot: OrderBookSnapshot,
side: str, # 'BUY' or 'SELL'
order_quantity: float,
market_impact_coeff: float = 0.1,
fixed_slippage_bps: float = 0.5
) -> SlippageResult:
"""
估算订单滑点
参数:
snapshot: 订单簿快照
side: 交易方向
order_quantity: 订单数量
market_impact_coeff: 市场冲击系数(与订单簿深度相关)
fixed_slippage_bps: 固定滑点(基点)
返回:
SlippageResult: 包含实际成交均价和滑点
"""
if side == 'BUY':
prices = snapshot.ask_prices
volumes = snapshot.ask_volumes
best_price = prices[0]
else:
prices = snapshot.bid_prices
volumes = snapshot.bid_volumes
best_price = prices[0]
# 基础滑点 = 买卖价差的一半(期望值)
base_slippage = snapshot.spread / 2
# 计算订单穿透深度
cumulative_volume = np.cumsum(volumes)
fill_mask = cumulative_volume < order_quantity
if not np.any(fill_mask):
# 订单量超出订单簿深度,冲击极大
avg_price = prices[-1] # 吃穿到最后一档
market_impact = (avg_price - best_price) / best_price * 10000
else:
# 完全成交在各档
n_filled = np.sum(fill_mask) + 1
partial_volume = order_quantity - cumulative_volume[min(np.sum(fill_mask) - 1, len(fill_mask) - 1)] if np.sum(fill_mask) > 0 else order_quantity
# 加权平均成交价
fill_volumes = np.concatenate([volumes[fill_mask], [partial_volume]])
avg_price = np.average(prices[:len(fill_volumes)], weights=fill_volumes)
# 市场冲击 = 成交量加权平均价与最佳价的偏离
market_impact = (avg_price - best_price) / best_price * 10000
# 总滑点 = 固定滑点 + 市场冲击 * 系数
total_slippage_bps = fixed_slippage_bps + market_impact * market_impact_coeff
# ⚠️ 注意:这是一个简化的线性冲击模型
# 生产环境中,冲击系数应基于历史成交数据校准
# 不同流动性标的(大盘股 vs 小盘股)的冲击系数差异可达 10 倍以上
return SlippageResult(
avg_fill_price=avg_price,
slippage_bps=total_slippage_bps,
fully_filled=np.sum(fill_mask) >= len(volumes) - 1
)
2.4 为什么你的回测滑点总是不够
很多量化新手只在回测中加一个固定滑点,比如 0.05%。这几乎总是低估了真实滑点,原因有三:
- 固定滑点掩盖了流动性差异:苹果股票和一只仙股的滑点能差 100 倍
- 没有考虑订单大小:1手和100手的滑点完全不同
- 忽略了波动率缩放:财报季的滑点是平静期的 3-5 倍
正确的做法是让滑点随市场状态和订单规模动态变化。
三、裂缝二:延迟成本——信号与执行的时间鸿沟
3.1 延迟的三层含义
在量化交易中,“延迟”不是一个单一概念,它有三层含义:
| 延迟类型 | 定义 | 典型量级 | 可控性 |
|---|---|---|---|
| 信号延迟 | 数据源到你的终端 | 毫秒到秒级 | 取决于数据源 |
| 执行延迟 | 你的订单到交易所 | 微秒到毫秒级 | 取决于基础设施 |
| 市场响应延迟 | 价格对你订单的响应 | 取决于市场微观结构 | 不可控 |
3.2 延迟成本的几何累积
延迟成本的可怕之处在于它是几何累积的。假设你在 t=0 时刻产生一个买入信号,预期买入价为 P₀。
如果不考虑延迟,你回测中的成交价是 P₀。但如果存在 Δt 秒延迟,在这段时间内:
Δt 秒后的价格 ≈ P₀ × (1 + μ × Δt + σ × √Δt × N(0,1))
其中 μ 是漂移率,σ 是波动率。这意味着延迟造成的价格不确定性随延迟时间的平方根增长。
3.3 生产级延迟模拟器
以下代码展示如何将延迟成本显式建模到回测框架中:
import time
import random
from typing import Callable, Dict, Any
from enum import Enum
from dataclasses import dataclass, field
import numpy as np
class OrderType(Enum):
MARKET = "market"
LIMIT = "limit"
STOP = "stop"
@dataclass
class LatencyProfile:
"""
延迟特征描述
不同执行方式对应不同的延迟分布
"""
signal_to_strategy_us: int = 100_000 # 信号到策略处理(100ms)
strategy_to_order_us: int = 50_000 # 策略到下单指令(50ms)
order_to_exchange_us: int = 20_000 # 订单到交易所(下单方式决定)
# 不同下单方式的交易所延迟
exchange_delays: Dict[str, int] = field(default_factory=lambda: {
"market": 50_000, # 市价单:50ms
"smart": 100_000, # 智能路由:100ms
"direct": 5_000, # 直连交易所:5ms
})
def total_latency(self, order_type: str = "smart") -> int:
"""计算总延迟(微秒)"""
exchange_delay = self.exchange_delays.get(order_type, 100_000)
return (
self.signal_to_strategy_us
+ self.strategy_to_order_us
+ exchange_delay
)
def total_latency_seconds(self, order_type: str = "smart") -> float:
return self.total_latency(order_type) / 1_000_000
@dataclass
class PricePoint:
"""带时间戳的价格点"""
timestamp: float
price: float
volume: float
class LatencySimulator:
"""
延迟模拟器:将延迟成本显式建模
⚠️ 生产级注意事项:
1. 历史回测中的延迟应基于实际 benchmark 数据校准
2. 不同市场、不同交易时段的延迟分布差异显著
3. 高频策略需要考虑延迟的 jitter(抖动)效应
"""
def __init__(
self,
latency_profile: LatencyProfile,
volatility_per_second: float, # 每秒波动率
drift_per_second: float = 0.0, # 每秒漂移率
jitter_ratio: float = 0.2 # 抖动比例
):
self.profile = latency_profile
self.volatility_per_second = volatility_per_second
self.drift_per_second = drift_per_second
self.jitter_ratio = jitter_ratio
def simulate_execution_price(
self,
signal_price: float,
signal_time: float,
side: str,
order_type: str = "smart"
) -> tuple[float, float, float]:
"""
模拟延迟后的执行价格
返回:
(execution_price, latency_seconds, slippage_bps)
"""
# 计算延迟(带抖动)
base_latency = self.profile.total_latency_seconds(order_type)
jitter = base_latency * self.jitter_ratio * (2 * random.random() - 1)
actual_latency = max(0, base_latency + jitter)
# 价格移动模拟
# 使用几何布朗运动的离散近似
dt = actual_latency
# 漂移项
drift_effect = self.drift_per_second * dt
# 波动项
vol_effect = self.volatility_per_second * np.sqrt(dt) * np.random.normal()
# 总价格变动
price_change_ratio = drift_effect + vol_effect
execution_price = signal_price * (1 + price_change_ratio)
# 滑点(简化:买卖方向不同)
if side == 'BUY':
slippage_bps = abs(price_change_ratio) * 10000 + 0.5 # 加上基础滑点
else:
slippage_bps = abs(price_change_ratio) * 10000 + 0.5
return execution_price, actual_latency, slippage_bps
def create_latency_simulator(
annual_volatility: float,
annual_drift: float = 0.0,
order_type: str = "smart"
) -> LatencySimulator:
"""
工厂函数:创建延迟模拟器
参数:
annual_volatility: 年化波动率(如 0.25 表示 25%)
annual_drift: 年化漂移率
order_type: 下单方式
"""
# 转换为每秒的波动率和漂移
seconds_per_year = 252 * 6.5 * 3600 # 交易日 * 每日小时 * 每小时秒数
vol_per_second = annual_volatility / np.sqrt(seconds_per_year)
drift_per_second = annual_drift / seconds_per_year
return LatencySimulator(
latency_profile=LatencyProfile(),
volatility_per_second=vol_per_second,
drift_per_second=drift_per_second
)
3.4 延迟对策略的差异化影响
延迟成本对不同类型策略的影响差异巨大:
| 策略类型 | 典型持仓周期 | 延迟敏感度 | 典型延迟容忍 |
|---|---|---|---|
| 高频做市 | < 1 秒 | 极高 | < 1ms |
| 统计套利 | 秒到分钟 | 高 | < 100ms |
| 趋势跟踪 | 小时到天 | 中 | < 1 秒 |
| 价值投资 | 天到月 | 低 | < 1 分钟 |
如果你的趋势跟踪策略声称回测年化 23%,但实际用了 200ms 的直连交易所——这可能没问题。但如果同样的延迟发生在统计套利策略上,结果可能是灾难性的。
四、裂缝三:容量约束——流动性陷阱
4.1 容量约束的三重含义
容量(Capacity)是策略能承载的最大资金量。超过这个量级,策略收益会因为市场冲击而崩溃。容量约束有三个层次:
| 层次 | 描述 | 信号 |
|---|---|---|
| 市场规模 | 日均成交量(ADV)的上限 | 你的订单 > ADV 的 1% |
| 流动性深度 | 订单簿能承接的订单量 | 大单导致价差扩大 |
| 市场冲击 | 成交对价格的反身性影响 | 买入后价格持续上涨 |
4.2 阿尔梅斯-格兰杰容量模型
学术界常用的容量估算模型是阿尔梅斯-格兰杰(Almgren-Chriss)框架,它将最优执行问题形式化为:
最小化:E[成本] + λ × Var[成本]
其中:
E[成本] = 冲击成本 + 时间成本
Var[成本] = 波动率成本
λ = 风险厌恶系数
简化形式下,单笔订单的最优执行价格为:
def calculate_optimal_execution_price(
order_quantity: float,
current_price: float,
adv: float, # 平均日成交量
volatility_daily: float, # 日波动率
time_horizon_days: float, # 执行周期(天)
risk_aversion: float = 0.1, # 风险厌恶系数
participation_rate: float = 0.1 # 目标参与率(相对于ADV)
) -> dict:
"""
简化容量估算:考虑执行冲击的最优价格
⚠️ 简化模型,仅供概念演示
生产环境需使用更复杂的微观结构模型
"""
# 冲击系数(需根据历史数据校准)
# 这里使用线性冲击假设
market_impact_coeff = 0.1 # 订单量/ADV 每增加 1%,价格移动 0.1%
# 计算建议参与率
safe_participation = min(participation_rate, adv * 0.05) # 不超过 ADV 的 5%
# 单日可执行量
daily_capacity = adv * safe_participation
# 需要的执行天数
execution_days = max(1, order_quantity / daily_capacity)
# 市场冲击估算
participation_ratio = order_quantity / (adv * execution_days)
estimated_impact_pct = market_impact_coeff * participation_ratio * 100
# 考虑冲击的执行成本
avg_execution_price = current_price * (1 - estimated_impact_pct / 10000) if participation_ratio > 0 else current_price
# 波动率风险成本(时间越长,风险越大)
vol_risk_cost = volatility_daily * np.sqrt(execution_days) * np.sqrt(time_horizon_days) * risk_aversion
return {
"safe_daily_quantity": daily_capacity,
"execution_days": execution_days,
"estimated_impact_bps": estimated_impact_pct,
"adjusted_execution_price": avg_execution_price,
"volatility_risk_cost": vol_risk_cost,
"capacity_warning": order_quantity > adv * 0.2 # 超过 ADV 20% 需警告
}
4.3 回测中的容量盲区
大多数回测引擎默认你的订单不会影响市场。这是一个极其危险的假设。
考虑这个场景:
- 策略:突破 20 日高点时买入
- 标的:某小盘股,日均成交量 50 万股
- 策略资金:1 亿元,平均买入价 10 元
- 需要买入:1000 万股 = ADV 的 20%
你的回测假设以收盘价 10 元成交 1000 万股。但实际上:
- 你的订单会推动价格上涨
- 价格上涨会吸引其他趋势跟踪者
- 你的实际平均成交价可能是 10.8 元
- 这 8% 的冲击成本直接吞噬了你的收益
五、回测补偿框架:让幻觉变成真实
5.1 三层补偿模型
针对上述三道裂缝,我们提出一个三层补偿框架:
┌─────────────────────────────────────────────────────────────┐
│ 回测补偿框架 │
├─────────────────────────────────────────────────────────────┤
│ Layer 1: 数据层补偿 │
│ ├── 引入 tick 级数据或尽可能精细的数据源 │
│ ├── 清洗前视偏差(未来函数) │
│ └── 使用存活者偏差校正后的历史数据 │
├─────────────────────────────────────────────────────────────┤
│ Layer 2: 执行层建模 │
│ ├── 动态滑点模型(基于订单簿深度 + 波动率) │
│ ├── 延迟成本模拟(基于信号类型 + 执行方式) │
│ └── 容量约束嵌入(基于 ADV + 持仓周期) │
├─────────────────────────────────────────────────────────────┤
│ Layer 3: 结果层校准 │
│ ├── 蒙特卡洛模拟(考虑参数不确定性) │
│ ├── 参数敏感度分析(微笑曲线) │
│ └── 样本外验证(Walk-Forward) │
└─────────────────────────────────────────────────────────────┘
5.2 生产级回测引擎骨架
以下是整合三层补偿的回测引擎核心结构:
import pandas as pd
from typing import List, Dict, Optional, Any
from dataclasses import dataclass, field
from abc import ABC, abstractmethod
import os
@dataclass
class ExecutionParams:
"""执行参数"""
# 滑点参数
base_slippage_bps: float = 0.5
market_impact_coeff: float = 0.1
# 延迟参数
signal_latency_ms: int = 100
execution_latency_ms: int = 50
volatility_scaling: bool = True
# 容量参数
max_participation_rate: float = 0.05 # 不超过 ADV 的 5%
capacity_warning_threshold: float = 0.20 # 超过 20% 报警
# 成本参数
commission_rate: float = 0.001 # 佣金(千分之一)
stamp_duty: float = 0.001 # 印花税(仅卖出)
spread_cost_bps: float = 0.3 # 价差成本
@dataclass
class BacktestResult:
"""回测结果"""
total_return: float
sharpe_ratio: float
max_drawdown: float
max_drawdown_duration: int # 回撤持续天数
win_rate: float
profit_loss_ratio: float
total_trades: int
# 新增:成本分解
slippage_cost_total: float = 0.0
delay_cost_total: float = 0.0
commission_total: float = 0.0
capacity_breach_count: int = 0
def summary(self) -> str:
return f"""
=== 回测结果摘要 ===
总收益率: {self.total_return*100:.2f}%
夏普比率: {self.sharpe_ratio:.2f}
最大回撤: {self.max_drawdown*100:.2f}%
胜率: {self.win_rate*100:.2f}%
盈亏比: {self.profit_loss_ratio:.2f}
交易次数: {self.total_trades}
=== 成本分解 ===
滑点成本: {self.slippage_cost_total:.2f}
延迟成本: {self.delay_cost_total:.2f}
佣金+税费: {self.commission_total:.2f}
容量越限次数: {self.capacity_breach_count}
"""
class CostEstimator:
"""
成本估算器:整合滑点、延迟、容量三维度
⚠️ 重要:所有系数需要基于历史数据校准
下面的默认值适用于大盘股,低流动性标的需调整
"""
def __init__(self, params: ExecutionParams):
self.params = params
def estimate_slippage(
self,
price: float,
order_value: float,
adv: float,
volatility_daily: float
) -> float:
"""
估算滑点成本
"""
# 流动性比率
liquidity_ratio = order_value / adv if adv > 0 else 1.0
# 波动率缩放因子(高波动期滑点更大)
vol_scale = 1.0 + volatility_daily * 5 # 简化:波动率翻倍,滑点增加 25%
# 动态滑点 = 基础滑点 + 流动性冲击
slippage_bps = (
self.params.base_slippage_bps
+ liquidity_ratio * self.params.market_impact_coeff * 100
) * vol_scale
return price * order_value * slippage_bps / 10000
def estimate_delay_cost(
self,
signal_price: float,
execution_price: float,
order_value: float
) -> float:
"""
估算延迟成本(成交价差)
"""
# 延迟成本 = 信号价格与执行价格的差值
if execution_price > signal_price: # 做多
return (execution_price - signal_price) * order_value
else: # 做空
return (signal_price - execution_price) * order_value
def estimate_commission(
self,
order_value: float,
is_sell: bool
) -> float:
"""
估算佣金和税费
"""
commission = order_value * self.params.commission_rate
tax = order_value * self.params.stamp_duty if is_sell else 0.0
spread = order_value * self.params.spread_cost_bps / 10000
return commission + tax + spread
def check_capacity(
self,
order_value: float,
adv: float
) -> tuple[bool, float]:
"""
检查容量约束
返回:
(是否合规, 参与率)
"""
participation_rate = order_value / adv if adv > 0 else 0.0
compliant = participation_rate <= self.params.max_participation_rate
return compliant, participation_rate
class BacktestEngine:
"""
生产级回测引擎
特性:
1. 集成滑点、延迟、容量三维度成本估算
2. 全量记录成本分解
3. 容量越限报警
4. 支持 Walk-Forward 验证
⚠️ 当前实现为骨架,仅供架构参考
实际使用需要接入 TickDB 等数据源获取 tick 级数据
"""
def __init__(
self,
data_provider: Any, # 数据源接口
cost_estimator: CostEstimator,
params: ExecutionParams
):
self.data = data_provider
self.cost = cost_estimator
self.params = params
# 统计
self.trades: List[Dict] = []
self.cost_breakdown = {
'slippage': [],
'delay': [],
'commission': []
}
self.capacity_breaches = 0
def run(
self,
strategy: Any,
start_date: str,
end_date: str,
initial_capital: float = 1000000.0
) -> BacktestResult:
"""
执行回测
"""
# 获取历史数据
bars = self.data.get_bars(start_date, end_date)
# 初始化状态
capital = initial_capital
position = 0
equity_curve = []
trades = []
# 逐 bar 回测
for i, bar in enumerate(bars):
# 策略信号
signal = strategy.generate_signal(bar, position)
if signal and signal.action in ('BUY', 'SELL'):
# 获取标的信息
symbol = bar['symbol']
adv = self.data.get_adv(symbol, bar['date']) # 平均成交量
volatility = self.data.get_volatility(symbol, bar['date'])
# 订单价值
order_value = capital * signal.position_size if signal.position_size else capital * 0.1
order_quantity = order_value / bar['close']
# 1. 滑点估算
slippage_cost = self.cost.estimate_slippage(
bar['close'], order_value, adv, volatility
)
# 2. 延迟模拟(简化:使用波动率缩放的固定延迟)
latency_ms = self.params.signal_latency_ms + self.params.execution_latency_ms
if self.params.volatility_scaling:
latency_ms *= (1 + volatility * 3)
latency_delay = bar['close'] * order_value * (latency_ms / 1000) * volatility / 100
# ⚠️ 这是简化模型,实际应使用 LatencySimulator
# 3. 容量检查
compliant, participation = self.cost.check_capacity(order_value, adv)
if not compliant:
self.capacity_breaches += 1
# 超出部分按更差的成交价处理
excess_ratio = (participation - self.params.max_participation_rate) / participation
excess_cost = order_value * excess_ratio * 0.01 # 假设超出部分额外损失 1%
latency_delay += excess_cost
# 4. 执行价格(含滑点和延迟影响)
execution_price = bar['close']
if signal.action == 'BUY':
execution_price *= (1 + self.params.base_slippage_bps / 10000)
else:
execution_price *= (1 - self.params.base_slippage_bps / 10000)
# 5. 佣金计算
commission = self.cost.estimate_commission(
order_value, signal.action == 'SELL'
)
# 记录成本
self.cost_breakdown['slippage'].append(slippage_cost)
self.cost_breakdown['delay'].append(latency_delay)
self.cost_breakdown['commission'].append(commission)
# 更新持仓和资金
if signal.action == 'BUY':
position += order_quantity
capital -= (order_value + slippage_cost + latency_delay + commission)
else:
position -= order_quantity
capital += (order_value - slippage_cost - latency_delay - commission)
# 记录权益
equity = capital + position * bar['close']
equity_curve.append(equity)
# 计算统计指标
return self._calculate_metrics(equity_curve, trades)
def _calculate_metrics(
self,
equity_curve: List[float],
trades: List[Dict]
) -> BacktestResult:
"""计算回测指标"""
equity = np.array(equity_curve)
returns = np.diff(equity) / equity[:-1]
# 基础指标
total_return = (equity[-1] - equity[0]) / equity[0]
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252) if np.std(returns) > 0 else 0
# 最大回撤
cummax = np.maximum.accumulate(equity)
drawdowns = (cummax - equity) / cummax
max_drawdown = np.max(drawdowns)
return BacktestResult(
total_return=total_return,
sharpe_ratio=sharpe,
max_drawdown=max_drawdown,
max_drawdown_duration=0, # 简化
win_rate=0, # 需从 trades 计算
profit_loss_ratio=0,
total_trades=len(trades),
slippage_cost_total=sum(self.cost_breakdown['slippage']),
delay_cost_total=sum(self.cost_breakdown['delay']),
commission_total=sum(self.cost_breakdown['commission']),
capacity_breach_count=self.capacity_breaches
)
六、实战对比:含补偿 vs 不含补偿
为了直观展示三层补偿的效果,我们用一组虚构但具有代表性的参数做对比:
| 指标 | 不含补偿回测 | 含三层补偿回测 | 差异原因 |
|---|---|---|---|
| 年化收益率 | 23.4% | 12.1% | 11.3% 被成本吞噬 |
| 夏普比率 | 3.2 | 1.6 | 成本降低有效收益 |
| 最大回撤 | 8.2% | 14.7% | 未考虑流动性导致的假回撤 |
| 胜率 | 62% | 58% | 延迟导致部分信号失效 |
| 交易次数 | 847 次/年 | 812 次/年 | 容量限制减少实际执行 |
关键结论:如果你发现回测和实盘差了 50% 以上,问题很可能不是策略失效,而是回测从一开始就没有正确计入成本。
七、让回测成为真正的预测,而非幻想
回测不是历史回放的录像机。它是一个概率模型,模型里塞什么假设,就会输出什么结果。
当你发现回测与实盘存在巨大鸿沟时,不要急着否定策略。先问自己四个问题:
- 数据够细吗? 用日线回测日内策略,就像用年鉴预测季度财报——分辨率本身就是误差来源。
- 滑点够真吗? 固定滑点是偷懒,动态滑点才是负责任。
- 延迟够诚吗? 你的策略在高延迟下还能赚钱吗?
- 容量够大吗? 你的资金量级配得上这个策略吗?
一个好的回测系统不是为了证明策略有效,而是诚实地告诉你策略在什么条件下有效、失效概率有多大。
这才是量化交易的正确姿势。
下一步行动
如果你希望亲手实现本文的成本补偿框架:
- 访问 tickdb.ai 注册(免费,无需信用卡)
- 在控制台生成 API Key,获取 tick 级历史数据用于回测校准
- 将本文代码中的
data_provider接口替换为 TickDB 数据源 - 用历史数据回溯校准你的滑点系数和延迟参数
如果你正在寻找更细粒度的历史数据源:
TickDB 提供 10 年级别的美股历史 K 线数据,可用于跨周期策略回测验证。联系 [email protected] 了解机构级数据方案。
如果你想系统学习量化回测方法论:
建议阅读《Algorithmic Trading》(Ernest Chan)和《Trading and Exchanges》(Larry Harris),这两本书对回测局限性有深入讨论。
风险提示:本文所展示的回测补偿模型仅为概念演示,实际系数(滑点系数、市场冲击系数、延迟分布)需要基于真实历史数据和策略特性进行校准。历史回测结果不代表未来收益,任何策略在实际部署前都应进行充分的样本外验证和压力测试。
本文不构成任何投资建议。市场有风险,投资需谨慎。