14.4 限价订单簿动力学

名词解释：限价订单簿（Limit Order Book, LOB）

限价订单簿是现代电子化交易所的核心数据结构，按价格-时间优先级维护所有未成交的限价订单。LOB的动力学（Dynamics）研究订单的到达、取消、成交过程如何共同决定价格的演化，是高频交易和做市策略的理论基础。

一、LOB 的统计特性

1.1 订单簿状态的高维描述

限价订单簿在每个时刻 $t$ 的状态可以由以下变量完全描述：

• 买盘：$\mathbf{b}(t)=[b_1(t),b_2(t),\dots ,b_{K_b}(t)]$，其中 $b_i(t)$ 表示买盘第 $i$ 个价位上的挂单量
• 卖盘：$\mathbf{a}(t)=[a_1(t),a_2(t),\dots ,a_{K_a}(t)]$，其中 $a_i(t)$ 表示卖盘第 $i$ 个价位上的挂单量
• 最佳买卖价：$B(t)$（最高买价），$A(t)$（最低卖价）
• 中间价：$M(t)=(A(t)+B(t))/2$
• 价差：$S(t)=A(t)-B(t)$

$$LOB(t)=\{\mathbf{b}(t),\mathbf{a}(t),B(t),A(t)\}$$

1.2 关键统计特征

LOB 表现出一些普遍的统计规律：

1. 订单簿形状：平均订单簿随距离增加呈指数或幂律衰减
2. 价差分布：波动率越高、交易量越大，价差越小
3. 订单流不平衡：买卖订单到达率存在自相关和交叉相关
4. 成交量聚簇：大成交量之后往往跟随大成交量
5. 价格变动的离散性：由于最小变动价位（tick size），价格变动是离散的

PYTHON104 行 · 3.7 KB

📄此处有展示代码104 行 · 3.7 KB展开 ▼

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from scipy import stats

def analyze_lob_statistics(bid_prices: np.ndarray,
                            bid_sizes: np.ndarray,
                            ask_prices: np.ndarray,
                            ask_sizes: np.ndarray) -> dict:
    """
    分析限价订单簿的关键统计特性。

    参数:
        bid_prices: (T, K) 数组，T个时刻、K个档位的买价
        bid_sizes: (T, K) 数组，对应挂单量
        ask_prices: (T, K) 数组，卖价
        ask_sizes: (T, K) 数组，卖盘挂单量
    返回:
        统计特性字典
    """
    T, K = bid_prices.shape

    # 1. 价差统计
    spread = ask_prices[:, 0] - bid_prices[:, 0]
    mid_price = (ask_prices[:, 0] + bid_prices[:, 0]) / 2
    relative_spread = spread / mid_price * 10000  # bps

    # 2. 订单簿不平衡
    total_bid = np.sum(bid_sizes, axis=1)
    total_ask = np.sum(ask_sizes, axis=1)
    book_imbalance = (total_bid - total_ask) / (total_bid + total_ask + 1e-10)

    # 3. 深度加权平均价格
    dwap_bid = np.sum(bid_prices * bid_sizes, axis=1) / (total_bid + 1e-10)
    dwap_ask = np.sum(ask_prices * ask_sizes, axis=1) / (total_ask + 1e-10)

    # 4. 价格变动的离散性
    price_changes = np.diff(mid_price)
    tick_size = np.median(np.abs(np.diff(np.unique(np.round(mid_price, 2)))))

    stats_summary = {
        'spread_mean': np.mean(spread),
        'spread_median': np.median(spread),
        'relative_spread_bps_mean': np.mean(relative_spread),
        'book_imbalance_mean': np.mean(book_imbalance),
        'book_imbalance_autocorr': np.corrcoef(
            book_imbalance[:-1], book_imbalance[1:]
        )[0, 1],
        'price_changes_std': np.std(price_changes),
        'tick_size_estimate': tick_size
    }

    # 可视化
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))

    # 价差分布
    axes[0, 0].hist(spread, bins=50, color='steelblue', edgecolor='white',
                    alpha=0.8, density=True)
    axes[0, 0].axvline(x=np.mean(spread), color='red', linestyle='--',
                       label=f'均值 = {np.mean(spread):.4f}')
    axes[0, 0].set_xlabel('买卖价差')
    axes[0, 0].set_ylabel('频率')
    axes[0, 0].set_title('价差分布')
    axes[0, 0].legend()
    axes[0, 0].grid(True, alpha=0.3)

    # 订单簿平均形状
    avg_bid_depth = np.mean(bid_sizes, axis=0)
    avg_ask_depth = np.mean(ask_sizes, axis=0)

    levels = np.arange(1, K + 1)
    axes[0, 1].bar(-levels, avg_bid_depth[::-1], color='green', alpha=0.6,
                   label='买盘')
    axes[0, 1].bar(levels, avg_ask_depth, color='red', alpha=0.6,
                   label='卖盘')
    axes[0, 1].set_xlabel('档位（负=买, 正=卖）')
    axes[0, 1].set_ylabel('平均挂单量')
    axes[0, 1].set_title('订单簿平均形状')
    axes[0, 1].legend()
    axes[0, 1].grid(True, alpha=0.3, axis='y')

    # 订单簿不平衡的时序
    axes[1, 0].plot(book_imbalance, 'b-', linewidth=0.8)
    axes[1, 0].axhline(y=0, color='gray', linestyle='--', alpha=0.5)
    axes[1, 0].set_xlabel('时间')
    axes[1, 0].set_ylabel('订单簿不平衡')
    axes[1, 0].set_title('订单簿不平衡时序')
    axes[1, 0].grid(True, alpha=0.3)

    # 不平衡自相关
    lags = range(1, 51)
    autocorrs = [np.corrcoef(book_imbalance[:-lag], book_imbalance[lag:])[0, 1]
                 for lag in lags]
    axes[1, 1].bar(lags, autocorrs, width=0.8, color='steelblue')
    axes[1, 1].axhline(y=0, color='gray', linestyle='--', alpha=0.5)
    axes[1, 1].set_xlabel('滞后阶数')
    axes[1, 1].set_ylabel('自相关系数')
    axes[1, 1].set_title('订单簿不平衡自相关函数')
    axes[1, 1].grid(True, alpha=0.3)

    plt.tight_layout()
    plt.show()

    return stats_summary

▶ 点击展开可浏览运行结果

📘 本段代码定义了 1 个函数/类：函数 `analyze_lob_statistics`（分析限价订单簿的关键统计特性）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。

二、队列位置模型（Queue Position）

2.1 限价单排队机制

在电子化市场中，限价单按照价格优先-时间优先的原则排队。队列位置（Queue Position）决定了限价单成交的概率和等待时间。

• 队列位置：在同一价位上的相对排列位置，位置越靠前，优先成交
• 队列长度：某价位上的总挂单量
• 队列动态：新增订单排在队尾，取消订单从任意位置删除（但通常在队尾取消较多）

2.2 成交概率模型

一个位于队列位置 $q$、在价位上有总排队量 $Q$ 的限价单的成交概率为：

$$P(\text{fill}\mid q,Q)=F(\frac{q}{Q};\theta )$$

其中 $F$ 是经验累积分布函数，$\theta$ 是市场状态的参数。

PYTHON69 行 · 2.2 KB

📄此处有展示代码69 行 · 2.2 KB展开 ▼

def queue_position_model(lob_snapshots: list,
                          target_price: float,
                          max_horizon: int = 100) -> dict:
    """
    基于历史LOB快照估计队列位置的成交概率。

    参数:
        lob_snapshots: LOB快照列表，每个包含 bid_sizes, ask_sizes 等
        target_price: 目标挂单价位
        max_horizon: 最大预测时间步
    返回:
        包含成交概率曲线的字典
    """
    # 简化模拟：假设Poisson过程的订单消耗

    n_simulations = 1000
    fill_times = []

    for sim in range(n_simulations):
        queue_size = np.random.randint(100, 10000)  # 初始队列量
        position = np.random.randint(1, max(10, queue_size // 10))

        # 模拟订单消耗过程
        time = 0
        filled = False

        while time < max_horizon and not filled:
            # 市价单到达率（简化模型）
            market_order_arrival = np.random.poisson(0.3)
            if market_order_arrival > 0:
                executed_volume = np.random.exponential(500)
                queue_size -= executed_volume
                position -= executed_volume

                if position <= 0:
                    filled = True
                    fill_times.append(time)

            # 取消率
            if np.random.rand() < 0.02:
                cancelled = np.random.exponential(200)
                if position > queue_size * 0.5:  # 在队尾附近
                    queue_size -= cancelled

            time += 1

    # 成交概率曲线
    horizon_range = range(1, max_horizon + 1)
    fill_probs = [np.mean(np.array(fill_times) <= h) for h in horizon_range]

    # 拟合指数衰减曲线
    from scipy.optimize import curve_fit

    def exp_decay(t, a, b):
        return a * (1 - np.exp(-b * t))

    try:
        popt, _ = curve_fit(exp_decay, list(horizon_range), fill_probs,
                            p0=[1.0, 0.1])
    except:
        popt = [1.0, 0.1]

    return {
        'fill_times': fill_times,
        'fill_probs': fill_probs,
        'mean_fill_time': np.mean(fill_times) if fill_times else max_horizon,
        'median_fill_time': np.median(fill_times) if fill_times else max_horizon,
        'fit_params': popt
    }

▶ 点击展开可浏览运行结果

📘 本段代码定义了 1 个函数/类：函数 `queue_position_model`（基于历史LOB快照估计队列位置的成交概率）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。

2.3 最优挂单价位选择

做市商在选择挂单价位时，需要在成交概率和价差收益之间权衡：

• 靠近中间价（第1档）：成交概率高，但价差收益小，且面临逆向选择风险大
• 远离中间价（第3-5档）：成交概率低，但价差收益大，逆向选择风险小

最优挂单价位是使得期望收益最大化的价位：

$$k^{\ast }=\arg \underset{k}{max}[P(\text{fill}\mid k)\cdot (\text{spread at level }k)-C(\text{inventory})]$$

三、Hawkes 过程建模订单流自激性

3.1 Hawkes 过程基础

Hawkes 过程是一种自激性（Self-Exciting）的点过程，非常适合建模金融市场中事件的聚簇效应。其条件强度函数为：

$$\lambda (t)=\mu +\sum _{t_i<t}\varphi (t-t_i)$$

其中 $\mu$ 是基线强度（外生事件到达率），$\varphi (s)=\alpha e^{-\beta s}$ 是激发核函数（表示过去事件对当前事件率的影响），$\alpha$ 是激发强度，$\beta$ 是衰减速率。

3.2 订单流的多元 Hawkes 模型

PYTHON147 行 · 4.4 KB

📄此处有展示代码147 行 · 4.4 KB展开 ▼

import numpy as np
from scipy.optimize import minimize
from typing import Tuple

class MultivariateHawkes:
    """
    多元 Hawkes 过程用于建模订单簿事件。
    事件类型：买单单、卖单单、买成交、卖成交、买取消、卖取消。
    """

    def __init__(self, n_types: int = 6, decay: float = 50.0):
        """
        参数:
            n_types: 事件类型数量
            decay: 指数衰减参数 β（固定）
        """
        self.n_types = n_types
        self.decay = decay

    def simulate(self,
                 mu: np.ndarray,
                 alpha: np.ndarray,
                 T: float,
                 seed: int = 42) -> Tuple[np.ndarray, np.ndarray]:
        """
        模拟多元 Hawkes 过程。

        参数:
            mu: (n_types,) 基线强度
            alpha: (n_types, n_types) 激发矩阵
            T: 模拟时长（秒）
        返回:
            (event_times, event_types)
        """
        np.random.seed(seed)

        # Ogata 的 thinning 算法
        events = []
        t = 0.0

        while t < T:
            # 计算当前总强度
            intensities = mu.copy()
            for ev_time, ev_type in events:
                decay_factor = np.exp(-self.decay * (t - ev_time))
                intensities += alpha[ev_type, :] * decay_factor

            total_intensity = np.sum(intensities)

            # 生成候选事件间隔（使用上界）
            M = total_intensity  # 简化：使用精确强度而非上界
            if M < 1e-10:
                t += 0.001
                continue

            dt = np.random.exponential(1.0 / M)
            t_new = t + dt

            if t_new > T:
                break

            # 接受/拒绝
            intensities_new = mu.copy()
            for ev_time, ev_type in events:
                decay_factor = np.exp(-self.decay * (t_new - ev_time))
                intensities_new += alpha[ev_type, :] * decay_factor

            total_intensity_new = np.sum(intensities_new)

            if np.random.rand() < total_intensity_new / M:
                # 接受事件
                probs = intensities_new / total_intensity_new
                ev_type = np.random.choice(self.n_types, p=probs)
                events.append((t_new, ev_type))

            t = t_new

        event_times = np.array([e[0] for e in events])
        event_types = np.array([e[1] for e in events])

        return event_times, event_types

    def log_likelihood(self,
                       event_times: np.ndarray,
                       event_types: np.ndarray,
                       mu: np.ndarray,
                       alpha: np.ndarray,
                       T: float) -> float:
        """
        计算多元 Hawkes 过程的对数似然函数。
        """
        n_events = len(event_times)
        ll = 0.0

        for i in range(n_events):
            ti = event_times[i]
            ki = event_types[i]

            # 强度
            lam_i = mu[ki]
            for j in range(i):
                tj = event_times[j]
                kj = event_types[j]
                lam_i += alpha[kj, ki] * np.exp(-self.decay * (ti - tj))

            ll += np.log(max(lam_i, 1e-10))

        # 补偿项
        compensator = T * np.sum(mu)
        for i in range(n_events):
            ki = event_types[i]
            ti = event_times[i]
            for k in range(self.n_types):
                compensator -= alpha[ki, k] / self.decay * \
                    (1 - np.exp(-self.decay * (T - ti)))

        ll -= compensator

        return ll

    def fit(self, event_times: np.ndarray,
            event_types: np.ndarray,
            T: float) -> Tuple[np.ndarray, np.ndarray]:
        """
        通过极大似然估计拟合 Hawkes 过程参数。
        """
        n_params = self.n_types + self.n_types ** 2

        def neg_ll(params):
            mu = np.exp(params[:self.n_types])  # 确保正值
            alpha = np.exp(params[self.n_types:]).reshape(
                self.n_types, self.n_types
            )
            return -self.log_likelihood(event_times, event_types, mu, alpha, T)

        # 初始值
        x0 = np.zeros(n_params)

        result = minimize(neg_ll, x0, method='L-BFGS-B',
                          options={'maxiter': 500})

        mu_hat = np.exp(result.x[:self.n_types])
        alpha_hat = np.exp(result.x[self.n_types:]).reshape(
            self.n_types, self.n_types
        )

        return mu_hat, alpha_hat

▶ 点击展开可浏览运行结果

📘 本段代码定义了 1 个函数/类：类 `MultivariateHawkes`（多元 Hawkes 过程用于建模订单簿事件）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。

四、LOB 模拟器实现

4.1 零智能（Zero-Intelligence）订单簿模型

PYTHON169 行 · 5.8 KB

📄此处有展示代码169 行 · 5.8 KB展开 ▼

class ZeroIntelligenceLOB:
    """
    零智能限价订单簿模拟器。

    订单的到达、取消由 Poisson 过程驱动，订单价格和数量从指定分布中抽取。
    价格发现通过市价单匹配限价单实现。
    """

    def __init__(self,
                 initial_mid_price: float = 100.0,
                 tick_size: float = 0.01,
                 lambda_lo: float = 50.0,    # 限价单到达率
                 lambda_mo: float = 30.0,    # 市价单到达率
                 lambda_cancel: float = 20.0, # 取消率
                 price_jump_std: float = 0.05):
        self.mid_price = initial_mid_price
        self.tick_size = tick_size
        self.lambda_lo = lambda_lo
        self.lambda_mo = lambda_mo
        self.lambda_cancel = lambda_cancel
        self.price_jump_std = price_jump_std

        # 买卖订单簿：{price: quantity}
        self.bids = {}
        self.asks = {}
        self.time = 0.0
        self.trade_log = []
        self.mid_price_history = []

    def _generate_limit_price(self, side: str) -> float:
        """生成限价单价格"""
        n_ticks = int(np.random.exponential(5) + 1)
        offset = n_ticks * self.tick_size
        if side == 'bid':
            price = self.mid_price - offset
        else:
            price = self.mid_price + offset
        return round(price / self.tick_size) * self.tick_size

    def _generate_size(self) -> int:
        """生成订单量"""
        return int(np.random.exponential(500) + 100)

    def step(self) -> dict:
        """执行一个模拟步长（1秒）"""
        events = []

        # 1. 限价单到达
        n_lo = np.random.poisson(self.lambda_lo)
        for _ in range(n_lo):
            side = np.random.choice(['bid', 'ask'])
            price = self._generate_limit_price(side)
            size = self._generate_size()

            if side == 'bid':
                self.bids[price] = self.bids.get(price, 0) + size
            else:
                self.asks[price] = self.asks.get(price, 0) + size

            events.append({
                'type': 'limit_order',
                'side': side,
                'price': price,
                'size': size,
                'time': self.time
            })

        # 2. 市价单到达
        n_mo = np.random.poisson(self.lambda_mo)
        for _ in range(n_mo):
            side = np.random.choice(['buy', 'sell'])
            size = self._generate_size()

            remaining = size
            trade_price = None

            if side == 'buy':
                sorted_asks = sorted(self.asks.keys())
                for ask_price in sorted_asks:
                    available = self.asks[ask_price]
                    matched = min(remaining, available)
                    self.asks[ask_price] -= matched
                    if self.asks[ask_price] <= 0:
                        del self.asks[ask_price]
                    remaining -= matched
                    trade_price = ask_price
                    if remaining <= 0:
                        break
            else:
                sorted_bids = sorted(self.bids.keys(), reverse=True)
                for bid_price in sorted_bids:
                    available = self.bids[bid_price]
                    matched = min(remaining, available)
                    self.bids[bid_price] -= matched
                    if self.bids[bid_price] <= 0:
                        del self.bids[bid_price]
                    remaining -= matched
                    trade_price = bid_price
                    if remaining <= 0:
                        break

            if trade_price is not None:
                self.mid_price = trade_price
                self.trade_log.append({
                    'time': self.time,
                    'price': trade_price,
                    'volume': size - remaining,
                    'side': side
                })

            events.append({
                'type': 'market_order',
                'side': side,
                'price': trade_price,
                'size': size - remaining,
                'time': self.time
            })

        # 3. 随机取消
        n_cancel = np.random.poisson(self.lambda_cancel)
        for _ in range(n_cancel):
            side = np.random.choice(['bid', 'ask'])
            book = self.bids if side == 'bid' else self.asks
            if book:
                price = np.random.choice(list(book.keys()))
                cancel_size = min(
                    np.random.exponential(200),
                    book[price]
                )
                book[price] -= cancel_size
                if book[price] <= 0:
                    del book[price]

                events.append({
                    'type': 'cancel',
                    'side': side,
                    'price': price,
                    'size': cancel_size,
                    'time': self.time
                })

        # 4. 随机噪音（外生价格变动）
        self.mid_price += np.random.randn() * self.price_jump_std

        self.time += 1.0
        self.mid_price_history.append(self.mid_price)

        return events

    def get_best_prices(self) -> dict:
        """获取最佳买卖价"""
        best_bid = max(self.bids.keys()) if self.bids else None
        best_ask = min(self.asks.keys()) if self.asks else None
        return {
            'best_bid': best_bid,
            'best_ask': best_ask,
            'spread': (best_ask - best_bid) if (best_bid and best_ask) else None
        }

    def run(self, n_steps: int = 1000, verbose: bool = False) -> pd.DataFrame:
        """运行模拟器 n_steps 秒"""
        for step in range(n_steps):
            self.step()
            if verbose and step % 100 == 0:
                best = self.get_best_prices()
                print(f"Step {step}: Mid={self.mid_price:.2f}, "
                      f"Spread={best['spread']}")

        return pd.DataFrame(self.trade_log)

▶ 点击展开可浏览运行结果

📘 本段代码定义了 1 个函数/类：类 `ZeroIntelligenceLOB`（零智能限价订单簿模拟器）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。

LOB 模拟器为做市策略的回测提供了可控的实验环境。结合 Hawkes 过程对订单流自激性的刻画，可以在仿真环境中评估高频做市策略在不同市场状态下的表现，为实盘部署提供参考。