3. 延迟模拟：网络延迟、交易所撮合延迟的模拟技术

做高频交易的人，最怕什么？

不是策略亏钱，是策略在回测里跑得飞起，一上实盘就拉胯。我见过太多团队，回测曲线漂亮得像教科书，结果实盘第一天就被延迟干趴下了。说白了，回测里没考虑延迟，等于白测。

今天我们就聊聊延迟模拟。这东西听着简单，做起来全是坑。我个人习惯把延迟拆成两块：网络延迟和撮合延迟。分开模拟，再合起来看效果。

3.1 网络延迟的模拟

网络延迟，就是你发单到交易所，交易所返回确认，这一来一回的时间。它不是一个固定值，是抖动的。

我在项目中遇到过最头疼的情况：某家券商的行情延迟在大部分时间只有 200 微秒，但每到整点就会跳到 5 毫秒。回测里没模拟这个抖动，实盘时策略在整点附近疯狂亏钱。

3.1.1 固定延迟 vs 随机延迟

最简单的做法是加一个固定延迟。比如所有订单都延迟 1 毫秒。但这样做的问题很明显——太理想了。

真实网络什么样？

• 偶尔丢包重传
• 交换机队列满了要排队
• 光纤被挖断（嗯，真遇到过）

所以我建议用随机延迟模型。常用的有几种：

模型	特点	适用场景
正态分布	延迟集中在均值附近	普通网络环境
指数分布	偶尔出现大延迟	跨洲交易
实测分布采样	最真实	生产环境回测

你想想看，如果策略对延迟抖动敏感，用正态分布可能测不出问题。换成指数分布，那些小概率的大延迟事件，反而能暴露策略的脆弱点。

3.1.2 代码实现：一个简单的延迟模拟器

嗯，这里要注意。延迟模拟不能只加在订单发送上，行情接收也要加。我见过有人只模拟了发单延迟，结果行情延迟导致策略误判，照样亏钱。

import numpy as np
import time

class LatencySimulator:
    def __init__(self, base_latency_us=100, jitter_us=50):
        self.base = base_latency_us / 1_000_000  # 转成秒
        self.jitter = jitter_us / 1_000_000
        
    def apply(self):
        """模拟一次网络延迟"""
        delay = self.base + np.random.exponential(self.jitter)
        time.sleep(delay)
        return delay
    
    def apply_market_data(self, data):
        """对行情数据加延迟"""
        delay = self.apply()
        # 记录延迟信息
        data['latency'] = delay
        return data

这段代码看着简单，但有个坑：time.sleep 的精度在 Windows 上只有 1 毫秒左右。做高频模拟，建议用 time.perf_counter 配合忙等待，或者直接用 C++ 写延迟模块。

3.2 交易所撮合延迟的模拟

网络延迟只是前半段。订单到了交易所，还要排队、撮合。这部分延迟更复杂，也更关键。

我曾经做过一个策略，回测里胜率 70%，实盘只有 40%。查了三天，发现是撮合延迟导致的。回测里订单瞬间成交，实盘里排队排了 2 毫秒，价格已经变了。

3.2.1 撮合引擎的延迟模型

交易所撮合延迟，说白了就是三部分：

1. 队列等待时间：前面还有多少单子
2. 撮合计算时间：交易所 CPU 算一下能不能成交
3. 结果返回时间：成交确认传回来

其中队列等待时间是最难模拟的。因为它取决于当前市场的订单流。市场活跃时，队列长，等待久。市场冷清时，秒成交。

3.2.2 基于订单簿的撮合延迟模拟

更高级的做法，是直接模拟撮合过程。你需要维护一个虚拟订单簿，把策略的订单放进去，按价格时间优先排队。

class MatchingEngineSimulator:
    def __init__(self):
        self.order_book = {'bid': [], 'ask': []}
        self.match_latency_us = 20  # 撮合计算耗时
        
    def place_order(self, order):
        # 1. 计算队列位置
        queue_pos = self._get_queue_position(order)
        
        # 2. 估算等待时间
        wait_time = queue_pos * self.match_latency_us
        
        # 3. 模拟撮合
        time.sleep(wait_time / 1_000_000)
        
        # 4. 更新订单簿
        self._update_order_book(order)
        
        return {
            'filled': True,
            'queue_position': queue_pos,
            'total_latency': wait_time + self.match_latency_us
        }
    
    def _get_queue_position(self, order):
        """计算在队列中的位置"""
        side = 'bid' if order['side'] == 'buy' else 'ask'
        same_price_orders = [
            o for o in self.order_book[side] 
            if o['price'] == order['price']
        ]
        return len(same_price_orders) + 1

这个模型虽然简单，但已经能暴露很多问题了。比如你的策略喜欢在价格突变时抢单，队列位置会很长，撮合延迟暴增。回测里如果没模拟这个，实盘就会被打脸。

3.3 延迟模拟的实战配置

说了这么多，到底怎么配置延迟参数？我建议分三步走：

• 第一步：实测基线。用真实的交易环境，测出网络延迟的均值、方差、分布形态。别用理论值，用实测值。
• 第二步：压力放大。在基线基础上，把延迟放大 1.5 倍、2 倍。看看策略在恶劣环境下的表现。
• 第三步：极端测试。模拟网络抖动、交易所故障等极端情况。比如突然断线 500 毫秒，策略能不能扛住。

3.4 一个完整的延迟模拟框架

最后，我画了一张图，把整个延迟模拟的流程串起来。你照着这个框架搭，基本不会漏掉关键环节。

这个框架的核心思路是：分层模拟，逐级叠加。先单独测网络延迟的影响，再单独测撮合延迟，最后合起来看整体效果。哪个环节是瓶颈，一目了然。

公众号:蓝海资料掘金营，微信deep3321