Skip to content

18.2 模拟盘与实盘过渡

策略从回测通过到实盘上线,中间需要经过模拟盘(paper trading)阶段的验证和过渡。这个过程看似简单,实则是策略生命周期中最容易被低估的环节——许多在回测中表现优异的策略,到了实盘却大幅缩水甚至亏损,根源就在于忽视了模拟盘阶段暴露出的问题。

模拟盘的必要性和局限

为什么必须做模拟盘

模拟盘是用真实的市场行情数据驱动策略产生交易信号,但不实际执行交易、不实际占用资金的过程。它的核心价值在于:

验证数据管道:回测使用的是历史数据库中的数据,而模拟盘使用的是实时行情数据流。两者之间可能存在格式差异、时间戳对齐问题、数据缺失等。模拟盘能够在不承担资金风险的情况下暴露这些工程问题。

检验信号稳定性:回测中信号值是基于完整的历史数据计算的(例如均线计算有完整的回溯窗口),而实盘中信号是在数据流不断到达的过程中增量计算的。模拟盘可以验证信号在增量计算环境下的稳定性。

发现实盘特有的问题:例如交易所的临时停牌、涨跌停板的不可交易性、数据的修正和重述等——这些在回测中常常被忽略或简化,但在实盘中是真实存在的。

模拟盘的局限

模拟盘虽然重要,但有其不可克服的局限:

缺乏市场冲击:模拟盘不实际发送订单,因此信号对市场的影响被完全忽略。对于容量较小的策略,这个问题不严重;但对于交易量较大的信号,市场冲击可能是 Alpha 的主要侵蚀因素。

成交假设过于乐观:模拟盘的成交价通常取某个参考价(如信号触发后的第一个开盘价或中间价),但实盘中可能无法以该价格成交,尤其是对于流动性较差的股票。

缺少对手方博弈:实盘中你的订单会影响订单簿,其他市场参与者会对此做出反应。这是模拟盘完全无法模拟的。

心理因素缺失:模拟盘不涉及真实资金,没有心理压力。实盘中面对连续回撤时的心态变化可能影响判断和执行纪律。

名词解释:模拟盘(Paper Trading)

模拟盘是指在不使用真实资金的情况下,模拟交易策略执行的过程。策略接收实时市场数据,产生交易信号,记录虚拟交易和持仓变动,但不实际向交易所发送订单。模拟盘是连接回测和实盘的桥梁。

从模拟到实盘的检查清单

以下是一份经过业界验证的检查清单,建议在从模拟盘转实盘前逐项确认:

数据层面

  • [ ] 实时数据源的字段和格式与回测数据一致
  • [ ] 数据的延迟在策略可接受范围内(例如日频策略不应使用有15分钟延迟的数据)
  • [ ] 处理了数据缺失和异常值的情况(如股票停牌、除权除息)
  • [ ] 复权方式在实盘和回测中保持一致
  • [ ] 数据时区处理正确(A股使用北京时间,加密货币使用UTC时间)

信号层面

  • [ ] 信号值在回测环境和模拟盘环境中的差异在可接受范围内(建议做逐日比对)
  • [ ] 信号生成没有使用未来数据(前视偏差检查)
  • [ ] 信号的处理时效性满足要求(在信号有效的时间窗口内完成计算)
  • [ ] 信号异常值有兜底处理机制(如单日信号突变超过阈值时发出告警)

执行层面

  • [ ] 订单生成逻辑在回测和模拟盘中的输出一致
  • [ ] 处理了不可交易场景(涨跌停、停牌、除权日等)
  • [ ] 设置了订单价格和数量的合理上下限
  • [ ] 对撤单和部分成交有处理逻辑

风控层面

  • [ ] 单笔订单的最大金额限制
  • [ ] 单只股票的最大持仓比例限制
  • [ ] 行业/板块的集中度限制
  • [ ] 日内最大亏损限制(止损逻辑)
  • [ ] 系统异常时的紧急停止机制(Kill Switch)

监控层面

  • [ ] 关键指标(收益、回撤、换手率)的实时监控
  • [ ] 预期外的信号/订单/持仓变化告警
  • [ ] 日内 P&L 的实时计算和展示
  • [ ] 与回测基准的偏差监控

小资金渐进式上线策略

策略从模拟盘直接跳到满仓实盘是极其危险的。推荐采用渐进式上线(Gradual Rollout)策略:

第一阶段:微资金验证(1-2周)

用最小交易单位的资金运行策略(例如 A 股的一手即 100 股,或加密货币的最小合约张数)。这一阶段的目标是验证整个交易链路——从行情接收、信号生成、订单发送、成交回报到持仓管理——的每一个环节都能正常工作,而不追求收益。

验证重点

  • 订单是否能成功发送到交易所并被接收?
  • 成交回报的解析是否正确?
  • 持仓和资金数据是否与预期一致?

第二阶段:小资金运行(2-4周)

将资金规模提升到目标规模的 5%-10%。这一阶段开始观察策略在真实市场环境中的表现,重点关注实盘与模拟盘的收益差异。

每日核对项目

  • 实盘收益 vs 模拟盘收益的差异(预期在交易成本和滑点范围内)
  • 成交价格与信号参考价格的偏差
  • 实际换手率与预期的偏差
  • 是否有未预期的交易限制(如某些股票融券不可用)

第三阶段:逐步加仓(4-8周)

按照 10% → 25% → 50% → 75% → 100% 的节奏逐步放大规模。每次加仓之间至少间隔一周,确保上一档运行稳定后再进入下一档。

$$\text{加仓条件} = \begin{cases} \text{滚动夏普比率} \geq \text{回测夏普比率} \times 0.7 \\ \text{最大回撤} \leq \text{回测最大回撤} \times 1.5 \\ \text{跟踪误差} \leq 3\% \end{cases}$$

如果在某一档出现了异常(如夏普比率大幅下降、回撤超出预期),则暂停加仓,甚至回退到前一档进行排查。

实盘差异来源

实盘与回测的差异是量化交易中永恒的课题。理解差异的来源有助于合理设定预期并在策略设计中提前规避。

滑点(Slippage)

滑点是指预期成交价格与实际成交价格之间的差异。滑点来源包括:

  • 买卖价差(Bid-Ask Spread):市价买单通常以卖一价成交,市价卖单以买一价成交。中间的价差在半仓换手时产生成本。
  • 价格变动:从信号产生到订单到达交易所之间存在时间延迟,价格可能已经移动。
  • 市场冲击:大单推动价格向不利方向移动。

估算滑点的经验公式:

$$\text{滑点(bps)} \approx \text{Spread}_{50\%} + \sigma \times \sqrt{\frac{T}{V}} \times \text{IC}$$

其中 Spread50% 是半价差,σ 是波动率,T/V 是参与率(交易量占区间总成交量的比例),IC 是冲击系数。

延迟(Latency)

延迟是信号从生成到订单执行之间的时间差。延迟来源包括:

  • 数据延迟:交易所→行情服务商→策略系统的网络传输时间
  • 计算延迟:信号生成的时间复杂度
  • 通信延迟:策略系统→交易网关→交易所的网络往返时间
  • 交易所延迟:订单匹配引擎的处理时间

对于高频策略,微秒级别的延迟至关重要;对于日频策略,秒级别的延迟通常可以接受。

容量(Capacity)约束

容量不足会导致策略在资金规模增大时 Alpha 快速衰减。常见表现包括:

  • 线性衰减:Alpha 与规模的倒数大致线性相关,表明市场冲击是主要限制因素。
  • 门槛效应:规模超过某个临界点后 Alpha 突然大幅下降,通常意味着部分交易已经完全无法在期望价格附近完成。

其他差异来源

  • 税收和费用:印花税、佣金、过户费等。A股卖出时征收 0.1% 的印花税。
  • 融券限制:做空策略中,融券可获得性和融券费率可能与回测假设有显著差异。
  • 停复牌:回测中可能简单跳过了停牌期间的信号,但实盘中停牌期间的持仓无法操作。
  • 除权除息:分红、送转股、配股等事件的现金流和时间点可能与回测假设不同。

下面是一个模拟盘监控脚本的框架:

PYTHON219 行 · 7.1 KB
📄此处有展示代码219 行 · 7.1 KB展开 ▼
python
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class PaperTradingMonitor:
    """模拟盘监控与实盘差异追踪"""
    
    def __init__(self, backtest_pnl: pd.Series, strategy_config: dict):
        """
        Parameters
        ----------
        backtest_pnl : pd.Series
            回测的日度 PnL 序列,index 为日期
        strategy_config : dict
            策略配置,包含换手率、预期夏普、最大回撤等参数
        """
        self.backtest_pnl = backtest_pnl
        self.config = strategy_config
        self.paper_pnl = []
        self.live_pnl = []
        self.alerts = []
        
    def update_paper_pnl(self, date: str, pnl: float, signal_details: dict):
        """更新模拟盘的日度 PnL"""
        self.paper_pnl.append({
            'date': date,
            'pnl': pnl,
            **signal_details
        })
        
    def update_live_pnl(self, date: str, pnl: float, execution_details: dict):
        """更新实盘的日度 PnL"""
        self.live_pnl.append({
            'date': date,
            'pnl': pnl,
            **execution_details
        })
        
    def check_signal_divergence(self, paper_signals: pd.Series, 
                                 backtest_signals: pd.Series,
                                 threshold: float = 0.05) -> list:
        """
        检查模拟盘信号与回测信号的偏离
        
        Parameters
        ----------
        threshold : float
            允许的最大日度信号相关系数下降幅度
        """
        common_dates = paper_signals.index.intersection(backtest_signals.index)
        if len(common_dates) < 20:
            return []
        
        # 滚动相关系数
        rolling_corr = paper_signals[common_dates].rolling(20).corr(
            backtest_signals[common_dates]
        )
        
        alerts = []
        for date, corr in rolling_corr.dropna().items():
            if corr < 1.0 - threshold:
                alert = f"[{date}] 信号相关系数异常: {corr:.3f} < {1-threshold:.3f}"
                logger.warning(alert)
                alerts.append(alert)
        
        return alerts
    
    def track_pnl_divergence(self) -> pd.DataFrame:
        """追踪实盘与模拟盘 PnL 的差异"""
        if not self.paper_pnl or not self.live_pnl:
            return pd.DataFrame()
        
        df_paper = pd.DataFrame(self.paper_pnl).set_index('date')
        df_live = pd.DataFrame(self.live_pnl).set_index('date')
        
        # 合并
        df = pd.DataFrame({
            'paper_pnl': df_paper['pnl'].cumsum(),
            'live_pnl': df_live['pnl'].cumsum()
        })
        
        df['pnl_diff'] = df['live_pnl'] - df['paper_pnl']
        df['diff_pct'] = df['pnl_diff'] / df['paper_pnl'].abs().replace(0, np.nan)
        
        # 检查差异是否持续扩大
        if len(df) >= 10:
            recent_diff_trend = df['pnl_diff'].iloc[-10:].diff().mean()
            if recent_diff_trend < 0:
                alert = f"实盘与模拟盘差异持续扩大: 近10日均差{recent_diff_trend:.4f}"
                logger.warning(alert)
                self.alerts.append(alert)
        
        return df
    
    def execution_quality_report(self) -> dict:
        """执行质量报告"""
        if not self.live_pnl:
            return {}
        
        df = pd.DataFrame(self.live_pnl)
        
        # 滑点分析
        if 'slippage_bps' in df.columns:
            avg_slippage = df['slippage_bps'].mean()
            max_slippage = df['slippage_bps'].max()
        else:
            avg_slippage = max_slippage = None
        
        # 成交率分析
        if 'fill_rate' in df.columns:
            avg_fill_rate = df['fill_rate'].mean()
        else:
            avg_fill_rate = None
        
        # 撤单率
        if 'cancel_rate' in df.columns:
            avg_cancel = df['cancel_rate'].mean()
        else:
            avg_cancel = None
        
        report = {
            'trading_days': len(df),
            'avg_slippage_bps': avg_slippage,
            'max_slippage_bps': max_slippage,
            'avg_fill_rate': avg_fill_rate,
            'avg_cancel_rate': avg_cancel,
            'total_alerts': len(self.alerts)
        }
        
        return report
    
    def go_live_checklist(self) -> dict:
        """生成上线前检查报告"""
        checks = {
            'data_pipeline': {
                'status': 'PASS',
                'details': {
                    'data_freshness': 'Real-time',
                    'missing_rate': '0.02%',
                    'outlier_handled': True
                }
            },
            'signal_stability': {
                'status': 'PASS',
                'details': {
                    'corr_vs_backtest': 0.995,
                    'timestamp_aligned': True
                }
            },
            'execution_readiness': {
                'status': 'WARN',
                'details': {
                    'limit_up_down_handled': True,
                    'position_limits_set': True,
                    'kill_switch_configured': False  # 需要修复
                }
            },
            'risk_management': {
                'status': 'PASS',
                'details': {
                    'max_order_size': '1M CNY',
                    'max_position': '10%',
                    'sector_limit': '30%'
                }
            }
        }
        return checks


# 使用示例
if __name__ == '__main__':
    # 构造模拟回测 PnL
    dates = pd.date_range('2024-01-01', '2024-06-30', freq='B')
    np.random.seed(42)
    bt_pnl = pd.Series(
        np.random.randn(len(dates)).cumsum() * 1000,
        index=dates, name='backtest_pnl'
    )
    
    config = {
        'turnover': 0.3,
        'expected_sharpe': 1.5,
        'max_drawdown_limit': 0.15,
        'strategy_name': 'Alpha_001'
    }
    
    monitor = PaperTradingMonitor(bt_pnl, config)
    
    # 模拟每日更新
    for i, (date, pnl) in enumerate(bt_pnl.items()):
        monitor.update_paper_pnl(date.strftime('%Y-%m-%d'), pnl, {'n_signals': 12})
        # 实盘 PnL 加入一些噪声和滑点成本
        live_pnl = pnl * (1 + np.random.randn() * 0.02) - abs(pnl) * 0.0005
        monitor.update_live_pnl(
            date.strftime('%Y-%m-%d'), live_pnl,
            {'slippage_bps': abs(np.random.randn()) * 2,
             'fill_rate': 0.95 + np.random.rand() * 0.05,
             'cancel_rate': np.random.rand() * 0.05}
        )
    
    # 生成报告
    div_df = monitor.track_pnl_divergence()
    exec_report = monitor.execution_quality_report()
    checklist = monitor.go_live_checklist()
    
    print("===== 执行质量报告 =====")
    for k, v in exec_report.items():
        print(f"{k}: {v}")
    
    print("\n===== 上线检查清单 =====")
    for section, result in checklist.items():
        print(f"  {section}: {result['status']}")
        for check, value in result['details'].items():
            print(f"    - {check}: {value}")
点击展开可浏览运行结果
===== 执行质量报告 =====
trading_days: 130
avg_slippage_bps: 1.528256534292446
max_slippage_bps: 7.705462981309442
avg_fill_rate: 0.9749626718887566
avg_cancel_rate: 0.027734022900485045
total_alerts: 1

===== 上线检查清单 =====
  data_pipeline: PASS
    - data_freshness: Real-time
    - missing_rate: 0.02%
    - outlier_handled: True
  signal_stability: PASS
    - corr_vs_backtest: 0.995
    - timestamp_aligned: True
  execution_readiness: WARN
    - limit_up_down_handled: True
    - position_limits_set: True
    - kill_switch_configured: False
  risk_management: PASS
    - max_order_size: 1M CNY
    - max_position: 10%
    - sector_limit: 30%

模拟盘阶段的严谨与否,往往决定了实盘上线后头三个月是"波澜不惊"还是"鸡飞狗跳"。花在模拟盘上的每一分精力,都是在为实盘的稳定运行购买保险。