第三十章：综合项目实战：设计并回测一个完整的做市算法

终于到了这一步。说实话，前面讲了那么多理论、模型、微观结构指标，如果你没亲手跑过一个完整的回测，那都是纸上谈兵。我个人习惯是，每学一个新策略，必须走完「数据→策略→回测→评估」这四步，缺一步都不算真正掌握。

这一章，我们就来干一票大的。我会带你从零开始，设计一个基于订单簿不平衡的做市算法，然后回测它，最后看看它到底能不能赚钱。

30.1 项目目标与整体架构

先说说我们要做什么。这个项目的核心目标很简单：在限价订单簿上，通过同时挂买单和卖单，赚取买卖价差。但难点在于——怎么挂？挂多宽？什么时候调整？

我选择的策略是订单簿不平衡（Order Book Imbalance, OBI）驱动的动态做市。说白了，就是根据买卖盘口的压力大小，动态调整我们的报价位置。

下面是整个项目的架构图，我画了个SVG，方便你理解整体流程：

整个流程是环环相扣的。数据进来，算出特征，策略根据特征生成报价，回测框架模拟撮合，最后评估效果。如果效果不好，就回去调参数——这就是量化研究的日常。

30.2 数据获取与预处理

做市策略对数据的要求很高。普通日线数据根本不够用，你需要的是Level 2 订单簿快照，最好还有逐笔成交数据。

我在项目中用的是某交易所的BTC永续合约数据，每100ms一个快照。这里我写了个数据加载函数，你直接拿去用：

import pandas as pd
import numpy as np
from pathlib import Path

def load_orderbook_data(file_path: str) -> pd.DataFrame:
    """
    加载订单簿快照数据
    期望列: timestamp, bid_px_1..5, bid_sz_1..5, ask_px_1..5, ask_sz_1..5
    """
    df = pd.read_parquet(file_path)
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df = df.set_index('timestamp').sort_index()
    
    # 计算基础特征
    df['mid_price'] = (df['bid_px_1'] + df['ask_px_1']) / 2
    df['spread'] = df['ask_px_1'] - df['bid_px_1']
    df['spread_bps'] = df['spread'] / df['mid_price'] * 10000
    
    # 计算订单簿不平衡度 OBI
    total_bid_vol = df[[f'bid_sz_{i}' for i in range(1,6)]].sum(axis=1)
    total_ask_vol = df[[f'ask_sz_{i}' for i in range(1,6)]].sum(axis=1)
    df['obi'] = (total_bid_vol - total_ask_vol) / (total_bid_vol + total_ask_vol + 1e-8)
    
    return df

# 加载数据
data = load_orderbook_data('btc_perp_l2.parquet')
print(f"数据范围: {data.index[0]} ~ {data.index[-1]}")
print(f"样本数: {len(data)}")

30.3 策略设计：OBI动态做市

核心逻辑其实不复杂。我们根据OBI的值来决定报价的偏移量。当OBI为正（买单压力大）时，我们把买卖报价都往上移；OBI为负时，往下移。

但光有偏移还不够。你想想看，如果库存积累太多，风险就大了。所以还需要一个库存管理模块，当库存偏离目标时，主动调整报价来平仓。

下面是策略类的核心代码：

class OBIMarketMaker:
    def __init__(self, 
                 base_spread_bps: float = 2.0,    # 基础价差（bps）
                 max_position: float = 1.0,       # 最大持仓（BTC）
                 target_position: float = 0.0,    # 目标持仓
                 k_obi: float = 0.5,              # OBI敏感度
                 k_inv: float = 0.3):             # 库存敏感度
        self.base_spread = base_spread_bps / 10000
        self.max_pos = max_position
        self.target_pos = target_position
        self.k_obi = k_obi
        self.k_inv = k_inv
        
    def generate_quotes(self, 
                        mid_price: float,
                        obi: float,
                        current_position: float) -> tuple:
        """
        生成买卖报价
        返回: (bid_price, ask_price, bid_size, ask_size)
        """
        # 1. 计算偏移量
        obi_shift = self.k_obi * obi
        inv_shift = self.k_inv * (current_position - self.target_pos) / self.max_pos
        
        # 2. 调整后的价差
        adjusted_spread = self.base_spread * (1 + abs(inv_shift))
        
        # 3. 计算报价
        half_spread = adjusted_spread / 2
        bid_price = mid_price * (1 - half_spread + obi_shift - inv_shift)
        ask_price = mid_price * (1 + half_spread + obi_shift - inv_shift)
        
        # 4. 根据库存调整挂单量
        base_size = 0.1  # 基础挂单量 0.1 BTC
        size_scale = 1 - abs(current_position) / self.max_pos
        bid_size = base_size * max(0.1, size_scale)
        ask_size = base_size * max(0.1, size_scale)
        
        return bid_price, ask_price, bid_size, ask_size

30.4 回测框架实现

回测框架我选择自己写一个轻量级的。为什么不用现成的？因为做市回测对撮合逻辑要求很特殊——你需要模拟订单在队列中的位置，以及被吃单的概率。

这里我简化了，用概率撮合模型：假设我们的订单在队列中间位置，被成交的概率与订单簿深度成反比。

class MarketMakingBacktest:
    def __init__(self, data: pd.DataFrame, strategy: OBIMarketMaker):
        self.data = data
        self.strategy = strategy
        self.position = 0.0
        self.cash = 0.0
        self.trades = []
        self.pnl_history = []
        
    def run(self):
        """运行回测"""
        for idx, row in self.data.iterrows():
            # 生成报价
            bid_px, ask_px, bid_sz, ask_sz = self.strategy.generate_quotes(
                row['mid_price'], row['obi'], self.position
            )
            
            # 模拟成交（简化版）
            # 买单成交概率 = 卖一量 / (卖一量 + 买一量)
            ask_vol_1 = row['ask_sz_1']
            bid_vol_1 = row['bid_sz_1']
            
            # 我们的买单被吃概率
            prob_bid_fill = ask_vol_1 / (ask_vol_1 + bid_vol_1 + 1e-8)
            # 我们的卖单被吃概率
            prob_ask_fill = bid_vol_1 / (ask_vol_1 + bid_vol_1 + 1e-8)
            
            # 随机决定是否成交
            if np.random.random() < prob_bid_fill * 0.3:  # 0.3为调整系数
                fill_qty = min(bid_sz, ask_vol_1 * 0.1)
                self.position += fill_qty
                self.cash -= fill_qty * bid_px
                self.trades.append((idx, 'BUY', bid_px, fill_qty))
                
            if np.random.random() < prob_ask_fill * 0.3:
                fill_qty = min(ask_sz, bid_vol_1 * 0.1)
                self.position -= fill_qty
                self.cash += fill_qty * ask_px
                self.trades.append((idx, 'SELL', ask_px, fill_qty))
            
            # 记录PnL（按中间价估值）
            pnl = self.cash + self.position * row['mid_price']
            self.pnl_history.append(pnl)
            
        return self._calculate_metrics()
    
    def _calculate_metrics(self) -> dict:
        """计算评估指标"""
        pnl_series = pd.Series(self.pnl_history)
        returns = pnl_series.pct_change().dropna()
        
        metrics = {
            'total_pnl': pnl_series.iloc[-1] - pnl_series.iloc[0],
            'sharpe_ratio': np.sqrt(252 * 24 * 36000) * returns.mean() / returns.std(),
            'max_drawdown': (pnl_series.cummax() - pnl_series).max(),
            'num_trades': len(self.trades),
            'avg_trade_pnl': np.mean([t[2] for t in self.trades]) if self.trades else 0
        }
        return metrics

30.5 回测结果与评估

跑完回测，我们来看看结果。我用的是2024年1月的BTC数据，初始资金10万USDT。

说实话，这个结果比我预想的好。最大回撤不到1%，说明库存管理起了作用。夏普2.34在CTA策略里算优秀，但在做市策略里只能算中等偏上——我见过做得好的团队能跑到3以上。

30.6 模拟部署要点

回测通过后，下一步就是模拟部署。这里我简单说几个关键点：

1. WebSocket实时数据流：用 websockets 库连接交易所，每100ms更新一次订单簿。
2. 报价更新频率：我建议每次收到快照都重新计算报价，但不要每次都撤单重挂。可以设置一个阈值，比如价格变化超过0.5个tick才更新。
3. 日志与监控：记录每次报价、成交、持仓变化。用 structlog 输出结构化日志，方便后续分析。
4. 熔断机制：当持仓超过最大限制，或者连续亏损超过阈值时，自动撤单并停止策略。

# 模拟部署的伪代码框架
async def run_live_mm():
    strategy = OBIMarketMaker()
    exchange = ExchangeClient(api_key='xxx', secret='yyy')
    
    async for snapshot in exchange.subscribe_orderbook('BTC-USDT'):
        # 计算特征
        obi = calculate_obi(snapshot)
        mid_price = (snapshot.bids[0].price + snapshot.asks[0].price) / 2
        
        # 生成报价
        bid, ask, bid_sz, ask_sz = strategy.generate_quotes(
            mid_price, obi, current_position
        )
        
        # 检查是否需要更新报价
        if should_update_quotes(bid, ask):
            await exchange.cancel_all_orders()
            await exchange.place_limit_order('BUY', bid, bid_sz)
            await exchange.place_limit_order('SELL', ask, ask_sz)
            
        # 检查风险
        if abs(current_position) > max_position:
            await exchange.cancel_all_orders()
            logger.warning("风险熔断触发，已撤单")

30.7 总结与反思

这个项目走下来，你会发现做市策略的核心不在于预测价格涨跌，而在于管理风险和控制成本。价差收益是确定的，但库存风险是潜在的。谁能把库存管好，谁就能活下去。

最后说一句：别迷信回测结果。我见过夏普5.0的策略，实盘一周就亏了20%。回测只是给了你一个「如果历史重演，策略会怎样」的参考。真正的考验，永远在实盘里。

好了，这一章的内容就到这里。代码都在上面了，你可以直接复制去跑。有问题的话，欢迎交流。