第26章做市商模拟环境

第26章做市商模拟环境：仿真市场生成、对手方行为建模、压力测试场景设计

做市商策略上线前，最怕什么？

怕实盘直接亏钱。

我见过太多团队，策略回测跑得漂亮，一上实盘就崩。原因很简单——回测环境太「干净」了。真实市场有滑点、有对手方博弈、有极端行情，这些在普通回测里根本模拟不出来。

所以这一章，我们来聊聊怎么搭建一个靠谱的做市商模拟环境。说白了，就是给你的策略造一个「虚拟战场」。

26.1 仿真市场生成：从零造一个交易所

仿真市场的核心，是生成可信的订单簿和价格序列。我个人习惯用以下三种方法：

26.1.1 基于历史数据重放

最简单，也最真实。直接把历史Level2数据喂进去，逐笔回放。优点是市场微观结构完全真实，缺点是数据量有限，没法测试「没发生过」的场景。

class HistoricalReplayer:
    def __init__(self, data_path):
        self.data = pd.read_parquet(data_path)
        self.index = 0
    
    def next_tick(self):
        """返回下一个tick的订单簿快照"""
        if self.index >= len(self.data):
            return None
        tick = self.data.iloc[self.index]
        self.index += 1
        return {
            'bids': tick['bid_prices'],  # 数组
            'asks': tick['ask_prices'],
            'bid_sizes': tick['bid_sizes'],
            'ask_sizes': tick['ask_sizes'],
            'timestamp': tick['timestamp']
        }

26.1.2 随机过程生成

当你想测试「如果波动率突然翻倍会怎样」时，历史数据就不够用了。这时候需要用随机过程来生成价格路径。

我常用的是Heston模型，因为它能模拟波动率聚集效应——就是那种「跌的时候波动更大」的现象。

def generate_price_path(S0, T, N, kappa, theta, sigma_v, rho, v0):
    """
    Heston模型生成价格路径
    S0: 初始价格
    T: 时间长度(秒)
    N: 步数
    """
    dt = T / N
    prices = np.zeros(N)
    vols = np.zeros(N)
    prices[0] = S0
    vols[0] = v0
    
    for i in range(1, N):
        z1 = np.random.normal()
        z2 = np.random.normal()
        # 相关性处理
        dw1 = z1 * np.sqrt(dt)
        dw2 = (rho * z1 + np.sqrt(1 - rho**2) * z2) * np.sqrt(dt)
        
        # 方差过程（保证非负）
        vols[i] = max(vols[i-1] + kappa * (theta - vols[i-1]) * dt 
                     + sigma_v * np.sqrt(max(vols[i-1], 0)) * dw2, 0)
        
        # 价格过程
        prices[i] = prices[i-1] * np.exp(
            (0 - 0.5 * max(vols[i-1], 0)) * dt 
            + np.sqrt(max(vols[i-1], 0)) * dw1
        )
    
    return prices, vols

26.1.3 订单簿生成器

有了价格路径，还得有订单簿。我一般用「泊松到达 + 价格偏移」的方式来模拟限价单的到达和撤销。

class OrderBookSimulator:
    def __init__(self, mid_price, spread, depth=10):
        self.mid = mid_price
        self.spread = spread
        self.depth = depth
        self.bids = {}  # 价格 -> 数量
        self.asks = {}
        self._init_book()
    
    def _init_book(self):
        """初始化一个均匀分布的订单簿"""
        for i in range(1, self.depth + 1):
            bid_price = self.mid - self.spread/2 - i * 0.01
            ask_price = self.mid + self.spread/2 + i * 0.01
            self.bids[bid_price] = np.random.exponential(10)
            self.asks[ask_price] = np.random.exponential(10)
    
    def step(self, new_mid, dt):
        """模拟一个时间步的订单簿变化"""
        # 价格漂移
        shift = new_mid - self.mid
        self.mid = new_mid
        
        # 限价单到达（泊松过程）
        lambda_arrival = 5  # 每秒到达5单
        n_arrivals = np.random.poisson(lambda_arrival * dt)
        for _ in range(n_arrivals):
            side = 'buy' if np.random.random() < 0.5 else 'sell'
            price_offset = np.random.exponential(0.5)  # 价格偏移
            size = np.random.exponential(2)
            
            if side == 'buy':
                price = self.mid - self.spread/2 - price_offset * 0.01
                self.bids[price] = self.bids.get(price, 0) + size
            else:
                price = self.mid + self.spread/2 + price_offset * 0.01
                self.asks[price] = self.asks.get(price, 0) + size
        
        # 限价单撤销（每个价格以概率撤销）
        for book in [self.bids, self.asks]:
            for price in list(book.keys()):
                if np.random.random() < 0.1 * dt:  # 10%/秒的撤销率
                    del book[price]
        
        return self.get_snapshot()

我的经验：刚开始做仿真时，我总想把订单簿做得越精细越好。后来发现，对于做市商策略测试，关键是要模拟出「流动性分布」和「订单到达的随机性」，而不是追求每个tick都跟真实市场一模一样。够用就好。

26.2 对手方行为建模：你的敌人是谁？

做市商赚的是谁的钱？对手方的钱。所以你得知道对手方长什么样。

26.2.1 对手方类型

我一般把对手方分成三类：

类型	行为特征	对做市商的影响
知情交易者	有信息优势，只在有利可图时交易	造成逆向选择，做市商需要扩大价差
流动性交易者	随机买卖，无信息优势	提供利润来源，做市商赚取价差
算法交易者	高频、小单、策略化	增加市场噪音，可能被「钓鱼」

26.2.2 知情交易者建模

这个最难，也最关键。我参考了Kyle模型的思想：知情交易者知道资产的「真实价值」，当市场价格偏离真实价值时，他们会出手。

class InformedTrader:
    def __init__(self, true_value, sigma_info=0.01):
        self.true_value = true_value  # 真实价值（只有他知道）
        self.sigma_info = sigma_info  # 信息精度
    
    def decide(self, orderbook):
        """根据订单簿决定是否交易"""
        best_bid = max(orderbook.bids.keys())
        best_ask = min(orderbook.asks.keys())
        mid = (best_bid + best_ask) / 2
        
        # 如果市场价格偏离真实价值超过阈值
        threshold = self.sigma_info * 2
        
        if self.true_value - best_ask > threshold:
            # 真实价值远高于卖价，买入
            return ('buy', best_ask, orderbook.asks[best_ask])
        elif best_bid - self.true_value > threshold:
            # 真实价值远低于买价，卖出
            return ('sell', best_bid, orderbook.bids[best_bid])
        else:
            return None  # 不交易

26.2.3 流动性交易者建模

这个简单，用泊松过程加随机方向就行。但要注意，真实市场中流动性交易者的订单大小往往有「肥尾」特征——就是偶尔会出现大单。

class LiquidityTrader:
    def __init__(self, intensity=2.0):
        self.intensity = intensity  # 每秒交易次数
        self.size_dist = lambda: np.random.pareto(1.5) * 0.5  # 肥尾分布
    
    def generate_orders(self, dt):
        """生成一个时间步内的流动性订单"""
        n_orders = np.random.poisson(self.intensity * dt)
        orders = []
        for _ in range(n_orders):
            side = 'buy' if np.random.random() < 0.5 else 'sell'
            size = self.size_dist()
            orders.append((side, size))
        return orders

注意：我曾经在模拟中只用了正态分布来生成订单大小，结果策略表现特别好。一上实盘，被几个大单直接打穿。后来才意识到，真实市场的订单大小分布是肥尾的。所以模拟时一定要用Pareto分布或者混合分布。

26.3 压力测试场景设计：把策略往死里整

策略在正常市场里赚钱不算本事。能在极端行情下活下来，才是真功夫。

26.3.1 经典压力场景

我一般会测试以下场景：

闪电崩盘：价格在几秒内暴跌10%，然后快速反弹
流动性枯竭：订单簿深度骤降，价差扩大10倍
波动率爆发：波动率从10%瞬间跳到100%
对手方狙击：大量知情交易者同时出现，专门吃你的单
网络延迟：你的订单比别人晚到100ms

26.3.2 闪电崩盘模拟

def flash_crash_scenario(simulator, crash_depth=0.1, recovery_time=5):
    """
    模拟闪电崩盘
    crash_depth: 跌幅比例
    recovery_time: 恢复时间(秒)
    """
    original_mid = simulator.mid
    
    # 第一阶段：暴跌
    for t in range(10):  # 10个tick内完成暴跌
        new_mid = original_mid * (1 - crash_depth * (t+1)/10)
        simulator.step(new_mid, dt=0.1)
        yield simulator.get_snapshot()
    
    # 第二阶段：底部震荡
    for t in range(int(recovery_time * 10)):
        noise = np.random.normal(0, 0.005)
        new_mid = original_mid * (1 - crash_depth) * (1 + noise)
        simulator.step(new_mid, dt=0.1)
        yield simulator.get_snapshot()
    
    # 第三阶段：反弹
    for t in range(10):
        new_mid = original_mid * (1 - crash_depth * (1 - (t+1)/10))
        simulator.step(new_mid, dt=0.1)
        yield simulator.get_snapshot()

26.3.3 流动性枯竭模拟

def liquidity_drought(simulator, duration=30):
    """
    流动性枯竭：订单到达率降到正常值的1/10
    """
    original_intensity = simulator.liquidity_intensity
    simulator.liquidity_intensity = original_intensity * 0.1
    
    for _ in range(int(duration / 0.1)):
        # 同时扩大价差
        simulator.spread *= 1.01  # 每次扩大1%
        simulator.step(simulator.mid, dt=0.1)
        yield simulator.get_snapshot()
    
    # 恢复
    simulator.liquidity_intensity = original_intensity

26.4 完整模拟框架

把这些组件拼起来，就是一个完整的模拟环境了。我习惯用事件驱动的方式组织：

class MarketSimulator:
    def __init__(self, config):
        self.orderbook = OrderBookSimulator(
            config['initial_price'],
            config['initial_spread']
        )
        self.informed = InformedTrader(config['true_value'])
        self.liquidity = LiquidityTrader(config['liquidity_intensity'])
        self.history = []
    
    def run(self, n_steps, dt=0.1, stress_scenario=None):
        """运行模拟"""
        for step in range(n_steps):
            # 1. 更新价格（如果有压力场景）
            if stress_scenario:
                new_mid = next(stress_scenario, None)
                if new_mid is None:
                    break
            else:
                # 正常随机游走
                new_mid = self.orderbook.mid * (1 + np.random.normal(0, 0.001))
            
            # 2. 更新订单簿
            snapshot = self.orderbook.step(new_mid, dt)
            
            # 3. 生成对手方订单
            informed_order = self.informed.decide(self.orderbook)
            liquidity_orders = self.liquidity.generate_orders(dt)
            
            # 4. 执行交易（这里可以接入你的做市策略）
            all_orders = []
            if informed_order:
                all_orders.append(informed_order)
            all_orders.extend(liquidity_orders)
            
            # 5. 记录状态
            self.history.append({
                'step': step,
                'mid': new_mid,
                'spread': self.orderbook.spread,
                'n_orders': len(all_orders),
                'snapshot': snapshot
            })
        
        return self.history

核心要点：

仿真市场要能同时模拟「正常波动」和「极端事件」
对手方建模要区分知情交易者和流动性交易者
压力测试不是走过场，要真的把策略逼到极限
模拟环境要可复现（设置随机种子），方便对比不同策略

嗯，这套模拟环境我用了快三年。每次开发新策略，我都会先在这个环境里跑至少1000次模拟，覆盖各种极端场景。虽然不能保证实盘100%没问题，但至少能筛掉80%的坑。

记住一句话：模拟环境越残酷，实盘越安全。

无相订单流研究社微信Lucian808555

📚 课程目录