价格是结果，订单簿是原因

2010 年 5 月 6 日下午 2 点 32 分，道琼斯工业指数在 20 分钟内暴跌 600 点，然后又在一个小时内收复了大部分失地。事后调查报告显示，这一切的起点是一个共同基金的“大单出货”——一笔挂出卖出的订单直接击穿了多层流动性，引发了连锁式的止损反应。

但真正有意思的不是那个起点，而是在这 20 分钟里发生的事：做市商的报价去哪了？

答案是：撤了。在价格开始下跌的第一秒，做市商的报价系统就检测到了异常流量，立刻撤回了自己的 bid 和 ask。这意味着市场上的流动性提供者几乎在同一时刻选择了沉默。没有任何阴谋论，没有人为操纵，只是一套基于库存风险和撤单成本的理性决策算法。

理解这一点，是理解整个美股微观结构的核心。

本文要解决的就是这个问题：作为普通参与者，我们有没有可能从公开可观测的数据里，反推出做市商的行为模式？答案是有限的——我们永远拿不到他们的内部数据和库存日志——但通过对 tick 数据和成交数据的细致分析，我们可以推断出他们行为的轮廓，甚至利用这些推断构建交易信号。

这不是一篇教你“追踪庄家”的玄学文章。我们用数据说话，用代码验证，用市场微观结构理论做支撑。

---

一、市场微观结构基础：为什么做市商的行为是可推断的

在进入技术细节之前，需要先把理论框架立住。以下几件事是理解后文所有分析的前提。

1.1 做市商的核心逻辑

美股市场（NYSE/Nasdaq）的流动性主要由三类参与者提供：传统做市商（Designated Market Makers）、电子通信网络（ECN）和高频交易商（HFT）。无论名称如何，它们的行为模式高度相似：持续在 bid 和 ask 上挂出限价单，等待成交，然后从买卖价差中赚取利润。

做市商的利润来源有两个：

价差收益（Spread Capture）。假设你挂出 bid @ $100.00，ask @ $100.01。每一个完成 round-trip（先买后卖或先卖后买）的交易者，你赚 0.01 美元/股。在高频场景下，这个微小的数字乘以巨大的成交量，就是稳定的收入来源。

库存收益（Inventory收益）。但这里有个陷阱：做市商不是零风险的。每一笔交易都会让它的库存偏向一侧——如果持续有人买，它手里就堆满了多头仓位。如果价格随后下跌，这些多头仓位就会产生亏损。因此，做市商必须管理库存风险：当库存偏离中性仓位太多时，它会主动调整报价，把 bid 压低、把 ask 抬高，吸引反向交易者来“平仓”。

这两个逻辑叠加在一起，就产生了一个可观测的行为模式：做市商的报价变化是对库存状态的实时映射。库存越偏，它越倾向于单边撤单或调价。

1.2 信息不对称与知情交易者

Glosten 和 Milgrom（1985）在其经典模型中指出，做市商在每一笔成交时都面临信息不对称：对手方可能是一个“知情交易者”（知道股票的真实价值），也可能只是一个随机交易者（liquidity trader）。

做市商的理性策略是：根据观察到成交的方向和频率，持续更新自己的“先验概率”，并相应调整报价。如果一笔 buy 成交了，做市商会把这个信号解读为“价格可能上涨”，于是把 ask 往上抬一点。如果连续出现多笔 buy，做市商会加速调高报价，直到价差扩大到足以弥补“被知情交易者薅羊毛”的期望损失。

这意味着：成交的方向和频率本身就携带了信息。我们不需要看做市商自己的仓位，只要观察谁在买、谁在卖，就能反推他们的状态调整逻辑。

1.3 什么数据是我们能拿到的

明确了理论边界之后，接下来要问：哪些数据是我们能实际拿到的？

数据类型	内容	可推断的做市商行为
Tick/逐笔成交	每笔交易的成交价、成交量、方向（buy/sell）	订单流方向、供需失衡
Level 2/订单簿快照	bid/ask 各档价格和挂单量	流动性深度变化
Quote（NBBO）	全国最佳买卖报价	价差变化、报价方向
K 线/分钟数据	聚合后的 OHLCV	大趋势，无法做微观推断

本文的分析将主要基于逐笔成交数据（trades）和 Level 2 订单簿快照（depth）。如果我们用 TickDB 的数据源，美股支持 depth 频道（1 档快照），港股和数字货币支持 depth 频道的多档深度数据——这意味着港股和数字货币的分析可以更精细，美股的分析则需要在 1 档数据上做更多推断。

---

二、Tick 数据的方向推断：从逐笔成交读懂谁在主导

2.1 问题的核心：谁主动买了？

在美股市场，每一笔成交都有两个方向：buy 或 sell。问题是，成交价格并不能直接告诉我们方向。$100.00 的成交，可能是一个买家用 $100.00 的价格买入了股票（主动买入），也可能是一个持有股票的人以 $100.00 的价格卖出了股票（主动卖出）。

这就是 Tick 数据方向推断（tick direction inference）的核心问题。

2.2 主流方向推断方法

方法一： uptick / downtick 规则

最经典的方法来自 Lee 和 Ready（1993），也被称为 Lee-Ready 算法：

• uptick：当前成交价 > 上一笔成交价 → 这笔成交标注为 buy
• downtick：当前成交价 < 上一笔成交价 → 这笔成交标注为 sell
• zero uptick/downtick：当前成交价 = 上一笔成交价 → 沿用更早一笔的方向

def infer_tick_direction(prices: list[float]) -> list[str]:
    """
    Lee-Ready 方向推断算法
    prices: 按时间顺序排列的成交价格列表
    返回: ['buy', 'sell', ...] 的方向列表
    """
    if len(prices) < 2:
        return []

    directions = [None]  # 第一笔没有前序，标记为 None
    
    for i in range(1, len(prices)):
        if prices[i] > prices[i - 1]:
            directions.append("buy")
        elif prices[i] < prices[i - 1]:
            directions.append("sell")
        else:
            # zero uptick/downtick：回溯找上一个非零变化
            direction = directions[-1]
            for j in range(i - 2, -1, -1):
                if prices[j + 1] != prices[j]:
                    direction = "buy" if prices[j + 1] > prices[j] else "sell"
                    break
            directions.append(direction)

    return directions

这个方法在 90% 的情况下是准确的，但它有一个致命的缺陷：在价格横盘或剧烈波动时，方向的判断会滞后于实际成交。

方法二：tick rule（修正版）

tick rule 是 Lee-Ready 的简化版本，也是学术研究中的默认方法：

$$
d_i = \begin{cases} +1 & \text{if } p_i > p_{i-1} \text{ (uptick)} \ -1 & \text{if } p_i < p_{i-1} \text{ (downtick)} \ d_{i-1} & \text{if } p_i = p_{i-1} \text{ (zero tick)} \end{cases}
$$

其中 $d_i > 0$ 对应 buy，$d_i < 0$ 对应 sell。这个方法在计算上更高效，适合大规模回测。

方法三：结合订单簿数据的方向推断

当有 Level 2 数据时，可以做更精确的判断：比较成交价与 bid/ask 的关系。

def infer_with_quote(price: float, bid: float, ask: float) -> str:
    """
    结合报价的方向推断
    如果成交价 = ask：买方主动吃掉卖单 → buy
    如果成交价 = bid：卖方主动卖掉持仓 → sell
    如果成交价在 bid 和 ask 之间：用 uptick/downtick 规则兜底
    """
    if abs(price - ask) < 1e-9:
        return "buy"    # 成交在 ask，buy 触发了这笔交易
    elif abs(price - bid) < 1e-9:
        return "sell"   # 成交在 bid，sell 触发了这笔交易
    else:
        # 成交价在 bid-ask 之间，无法直接判断，使用 tick rule
        return "unknown"  # 实际实现中需要传入前一笔价格用 uptick 规则

这个方法比纯 uptick 规则准确得多，因为它直接比较了成交价与当时的流动性报价。这也是为什么 Level 2 数据在微观结构分析中如此重要。

2.3 方向推断的局限性

需要明确指出的是，以上所有方法都是近似推断，而非真实方向。原因如下：

成交价不是唯一的判断标准。一笔 $100.00 的成交，可能对应 100 股以 $100.00 触发在 ask 上的买单（buy），也可能对应 5000 股以 $100.00 在 bid 上被动成交的大单（sell 触发）。仅凭价格无法区分“100股的小单”和“5000股的大单”。

时间戳精度问题。美国市场的成交数据（consolidated tape）通常有毫秒级时间戳，但在极端行情下，时间戳可能存在乱序或延迟，这会直接影响 uptick/downtick 的判断。

替代方案的局限性。有些数据供应商会直接标注成交方向（FINRA 的 TRF 数据），这当然是最准确的，但这类数据通常需要商业授权。在没有标注数据的情况下，Lee-Ready 仍是业界最广泛使用的近似方法。

---

三、从成交数据反推做市商库存状态

3.1 核心指标：订单流失衡（Order Flow Imbalance, OFI）

有了方向标注的逐笔数据，我们就可以计算 OFI（Order Flow Imbalance），这是反推做市商行为最直接的代理变量。

OFI 的定义是：单位时间内主动买入量与主动卖出量之差。

$$
\text{OFI}_t = V_t^{\text{buy}} - V_t^{\text{sell}}
$$

当 OFI > 0，表示当前时间窗口内买入压力占优，买方更积极，价格有向上驱动；OFI < 0 则相反。

from dataclasses import dataclass
from collections import deque
import time

@dataclass
class TickData:
    timestamp: float  # Unix 时间戳，秒级
    price: float
    volume: int
    direction: str   # 'buy' or 'sell'


class OrderFlowAnalyzer:
    """
    订单流分析器
    实时计算滚动窗口内的 OFI 和累计压力
    """
    
    def __init__(self, window_seconds: float = 60.0, tick_rule: bool = True):
        self.window_seconds = window_seconds
        self.tick_rule = tick_rule
        self.tick_buffer: deque[TickData] = deque()
        self.current_ask = 0.0
        self.current_bid = 0.0
        self.last_price = 0.0
        
        # OFI 累积序列（用于后续因子计算）
        self.ofi_history: list[float] = []
        
        # 预警阈值（示例）
        self.ofi_threshold_bullish = 5000   # 每分钟 OFI 超过此值触发预警
        self.ofi_threshold_bearish = -5000

    def update_quote(self, bid: float, ask: float):
        """接收 Level 2 报价更新，维护当前 bid/ask"""
        self.current_bid = bid
        self.current_ask = ask

    def process_trade(self, price: float, volume: int, timestamp: float) -> dict:
        """
        处理一笔成交，返回更新后的 OFI 状态
        """
        # 推断方向（如果没有外部标注）
        if self.tick_rule:
            direction = self._tick_rule_direction(price)
        else:
            direction = self._quote_based_direction(price)
        
        tick = TickData(timestamp=timestamp, price=price, volume=volume, direction=direction)
        self.tick_buffer.append(tick)
        self.last_price = price
        
        # 移除超出窗口的数据
        cutoff = timestamp - self.window_seconds
        while self.tick_buffer and self.tick_buffer[0].timestamp < cutoff:
            self.tick_buffer.popleft()
        
        # 计算当前窗口 OFI
        ofi = self._calculate_ofi()
        self.ofi_history.append(ofi)
        
        # 生成预警信号
        signal = self._generate_signal(ofi)
        
        return {
            "ofi": ofi,
            "direction": direction,
            "tick_count": len(self.tick_buffer),
            "signal": signal,
            "bid": self.current_bid,
            "ask": self.current_ask,
            "spread": self.current_ask - self.current_bid,
        }

    def _tick_rule_direction(self, price: float) -> str:
        """Tick rule 方向推断"""
        if self.last_price == 0:
            return "unknown"
        if price > self.last_price:
            return "buy"
        elif price < self.last_price:
            return "sell"
        else:
            # zero tick：需要记录 last_direction，此处简化处理
            return "neutral"

    def _quote_based_direction(self, price: float) -> str:
        """基于报价的方向推断（更准确）"""
        if self.current_ask > 0 and abs(price - self.current_ask) < 1e-6:
            return "buy"
        if self.current_bid > 0 and abs(price - self.current_bid) < 1e-6:
            return "sell"
        return self._tick_rule_direction(price)

    def _calculate_ofi(self) -> float:
        """计算当前滚动窗口内的 OFI"""
        buy_volume = sum(t.volume for t in self.tick_buffer if t.direction == "buy")
        sell_volume = sum(t.volume for t in self.tick_buffer if t.direction == "sell")
        return buy_volume - sell_volume

    def _generate_signal(self, ofi: float) -> str:
        """基于 OFI 生成简单信号"""
        if ofi > self.ofi_threshold_bullish:
            return "STRONG_BUY_PRESSURE"
        elif ofi < self.ofi_threshold_bearish:
            return "STRONG_SELL_PRESSURE"
        elif abs(ofi) < 1000:
            return "BALANCED"
        elif ofi > 0:
            return "MILD_BUY_PRESSURE"
        else:
            return "MILD_SELL_PRESSURE"

3.2 用 OFI 反推做市商行为的逻辑链

现在，我们可以把 OFI 和做市商的行为模式串起来：

场景 A：OFI 持续为正，且价差在扩大

OFI > 0 → 买方持续主导 → 知情交易者可能在场
做市商感知到：对手方更可能是知情者 → 提高 ask、降低 bid
价差扩大 → 做市商在增加"信息不对称溢价"
推断：价格短期有向上动量，但高 OFI + 大价差是反转信号的前兆

场景 B：OFI 接近零，但价格仍在上涨

OFI ≈ 0 → 成交数量均衡，但价格上涨
可能原因：被动买盘（limit buy orders）增多，做市商被动成交
推断：趋势可能是噪音驱动的，可持续性存疑

场景 C：OFI 突然由正转负，伴随价差急剧扩大

OFI 从 +5000 瞬间变为 -3000
价差从 $0.01 扩大到 $0.05
推断：做市商集体撤单或大幅调价
这是最强的短期反转信号之一——流动性真空即将形成

这就是“闪电崩盘”前夕的微观结构前兆。当 OFI 的急剧逆转叠加价差扩大出现时，意味着做市商已经开始对当前方向失去信心。

3.3 库存调整速率指标

除了 OFI，还有一类指标可以反推做市商的库存压力——库存调整速率（Inventory Adjustment Rate）。

思路是：观察价差中点（mid-price）的移动方向和 OFI 的关系。如果 OFI > 0 但 mid-price 下跌（而不是上涨），说明有强卖方在压制价格，而买方的成交未能驱动价格上涨——这通常意味着对手方是更有信息优势的一方。

$$
\text{Price-Decrease Ratio} = \frac{\text{count}(OFI_t > 0 \cap \Delta P_t < 0)}{\text{count}(OFI_t > 0)}
$$

这个比值越高，说明在 OFI 为正时价格反而下跌的情况越多，知情交易者反向操作的可能性越大。

---

四、VPIN：检测知情交易者混入的预警指标

4.1 什么是 VPIN

VPIN（Volume-Synchronized Probability of Informed Trading）由 Easley、Lopez de Prado 和 O'Hara 在 2012 年提出，是一种专门用于高频场景的知情交易概率估计。

它的核心思想是：把成交量分成固定大小的"桶"（volume buckets），在每个桶内计算 buy/sell 比例，然后用这个比例来估计当前市场中有多少比例的交易者是知情交易者。

为什么不用时间窗口而用成交量桶？原因是：在信息事件发生前后，交易频率可能发生剧烈变化。用固定时间窗口会导致在高交易量时期一个窗口内有很多交易，在低交易量时期一个窗口内几乎没有交易。成交量桶保证了每个统计单元有相同的信息量。

4.2 VPIN 的计算实现

import numpy as np
from collections import deque

class VPINCalculator:
    """
    Volume-Synchronized Probability of Informed Trading (VPIN)
    参考: Easley, López de Prado, O'Hara (2012)
    
    VPIN 值越高，表示知情交易者比例越高
    在正常市场条件下，VPIN 通常 < 0.7
    VPIN 急剧上升通常是价格大幅波动的先兆
    """
    
    def __init__(self, bucket_size: int = 1000):
        """
        Args:
            bucket_size: 每个 volume bucket 对应的成交量股数
                        对于大盘股可设更高（如 50000），小盘股设更低
        """
        self.bucket_size = bucket_size
        self.volume_buckets: deque[float] = deque(maxlen=50)
        self.pending_volume = 0.0  # 尚未填满一个 bucket 的累积成交量
        self.pending_buy_pct = 0.0  # pending volume 中的 buy 比例
        
        self.vpin_history: list[float] = []
        self.alert_threshold = 0.8  # VPIN 预警阈值

    def update(self, volume: int, direction: str):
        """
        更新 VPIN 计算
        """
        buy_volume = volume if direction == "buy" else 0.0
        
        # 累积 pending volume
        self.pending_volume += volume
        self.pending_buy_pct = (
            (self.pending_buy_pct * (self.pending_volume - volume) + buy_volume)
            / max(self.pending_volume, 1e-9)
        )
        
        # 如果 pending volume 超过 bucket size，生成新 bucket
        while self.pending_volume >= self.bucket_size:
            # 填满一个 bucket，记录其 buy percentage
            fill_ratio = (self.pending_volume - (self.pending_volume - self.bucket_size)) / self.bucket_size
            bucket_buy_pct = self.pending_buy_pct
            self.volume_buckets.append(bucket_buy_pct)
            
            # 剩余 volume 继续累积
            self.pending_volume -= self.bucket_size
        
        # 计算当前 VPIN
        if len(self.volume_buckets) >= 10:
            vpin = self._compute_vpin()
            self.vpin_history.append(vpin)
        else:
            vpin = None
        
        return vpin

    def _compute_vpin(self, n: int = 50) -> float:
        """
        VPIN = |V_buy - V_sell| / V_total (在最近 n 个 buckets 上)
        等价于：|2 * avg(buy_pct) - 1| 近似
        """
        recent = list(self.volume_buckets)[-n:]
        if not recent:
            return 0.0
        
        vpin = sum(abs(p - (1 - p)) for p in recent) / len(recent)
        # vpin ∈ [0, 1]，值越大知情交易概率越高
        return vpin

    def check_alert(self) -> dict:
        """
        检查是否触发 VPIN 预警
        返回预警状态和建议
        """
        if len(self.vpin_history) < 20:
            return {"alert": False, "reason": "数据不足"}
        
        current_vpin = self.vpin_history[-1]
        avg_vpin = np.mean(self.vpin_history[-20:])
        std_vpin = np.std(self.vpin_history[-20:])
        
        # z-score 预警
        z_score = (current_vpin - avg_vpin) / max(std_vpin, 1e-9)
        
        if current_vpin > self.alert_threshold:
            return {
                "alert": True,
                "level": "HIGH",
                "current_vpin": current_vpin,
                "z_score": z_score,
                "reason": f"VPIN={current_vpin:.3f} 超过阈值 {self.alert_threshold}",
                "suggestion": "知情交易者活动异常，建议减少仓位或等待波动率回归"
            }
        elif z_score > 2.0:
            return {
                "alert": True,
                "level": "MEDIUM",
                "current_vpin": current_vpin,
                "z_score": z_score,
                "reason": f"VPIN 的 z-score={z_score:.1f}，急剧上升",
                "suggestion": "检测到异常交易活动，建议关注后续价差变化"
            }
        else:
            return {
                "alert": False,
                "current_vpin": current_vpin,
                "z_score": z_score
            }

4.3 VPIN 的实际应用价值

VPIN 最著名的应用场景是高频闪崩预警。Easley 等人在 2012 年的论文中指出，VPIN 在 2010 年 5 月 6 日的闪电崩盘前 30-60 分钟就开始急剧上升，显示出知情交易者（或者说有信息优势的交易者）的异常活动。

但需要注意：VPIN 不是一个预测指标，而是一个异常检测指标。它告诉你"现在可能有问题"，但不能告诉你"价格会向哪个方向走"。结合 OFI 的方向信息，VPIN 的预警才能被赋予方向性。

---

五、实战框架：用 Tick 数据构建做市商行为监控面板

5.1 系统架构

一个完整的做市商行为监控系统需要以下组件：

数据源 (WebSocket/REST)
    │
    ▼
数据接收层
│  - 鉴权 (API Key from env)
│  - 心跳保活 (ping/pong)
│  - 限频处理 (code 3001 + Retry-After)
│  - 重连机制 (指数退避 + 抖动)
    │
    ▼
数据处理层
│  - tick 方向推断 (Lee-Ready / Quote-based)
│  - OFI 实时计算 (滚动窗口)
│  - VPIN 异步更新 (volume bucket)
│  - 价差监控 (spread monitoring)
    │
    ▼
信号生成层
│  - OFI 方向信号
│  - VPIN 异常检测
│  - 库存压力评分
│  - 流动性真空预警
    │
    ▼
告警层
│  - 飞书/Slack Webhook 通知
│  - 价格阈值触发
│  - OFI/VPIN 异常触发

5.2 生产级数据订阅代码

以下是使用 TickDB WebSocket 订阅 depth 数据（可用于美股 1 档订单簿快照分析）的生产级代码示例。注意：美股不支持 trades 接口，此处演示 depth 频道的订阅逻辑，并展示如何将其与 OFI 计算结合：

import os
import json
import time
import random
import threading
import logging
from dataclasses import dataclass, field
from collections import deque
from typing import Callable, Optional
import websocket  # pip install websocket-client

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)


@dataclass
class DepthQuote:
    """订单簿快照"""
    symbol: str
    timestamp: float
    bid_price: float
    bid_volume: int
    ask_price: float
    ask_volume: int

    @property
    def mid_price(self) -> float:
        return (self.bid_price + self.ask_price) / 2

    @property
    def spread(self) -> float:
        return self.ask_price - self.bid_price


class TickDBWebSocketClient:
    """
    TickDB WebSocket 客户端
    生产级实现：心跳保活、指数退避重连、限频处理
    """
    
    def __init__(self, symbols: list[str]):
        self.symbols = symbols
        self.api_key = os.environ.get("TICKDB_API_KEY")
        if not self.api_key:
            raise ValueError("请设置环境变量 TICKDB_API_KEY")
        
        self.ws: Optional[websocket.WebSocketApp] = None
        self.running = False
        self.reconnect_delay = 1.0
        self.max_reconnect_delay = 60.0
        
        # 回调
        self.on_depth: Optional[Callable[[DepthQuote], None]] = None
        
        # 内部状态
        self._last_heartbeat = 0.0
        self._hb_interval = 30.0  # 心跳间隔秒数
        self._lock = threading.Lock()

    def connect(self):
        """建立 WebSocket 连接"""
        base_url = "wss://api.tickdb.ai/v1/market/ws"
        params = f"?api_key={self.api_key}"
        url = f"{base_url}{params}"
        
        logger.info(f"连接 TickDB WebSocket: {url.split('?')[0]}")
        
        self.ws = websocket.WebSocketApp(
            url,
            on_open=self._on_open,
            on_message=self._on_message,
            on_error=self._on_error,
            on_close=self._on_close,
        )
        
        self.running = True
        self.ws.run_forever(
            ping_interval=20,       # 协议层 ping
            ping_timeout=10,
        )

    def _on_open(self, ws):
        logger.info("WebSocket 连接已建立，开始订阅 depth 频道")
        for symbol in self.symbols:
            self._send_subscribe(symbol)
        self._last_heartbeat = time.time()
        # 重置重连延迟
        self.reconnect_delay = 1.0

    def _send_subscribe(self, symbol: str):
        """发送订阅命令"""
        msg = json.dumps({
            "cmd": "subscribe",
            "params": {
                "symbol": symbol,
                "channels": ["depth"]
            }
        })
        if self.ws:
            self.ws.send(msg)
        logger.info(f"已订阅 {symbol} depth 频道")

    def _on_message(self, ws, message: str):
        """处理接收到的消息"""
        try:
            data = json.loads(message)
            
            # 心跳响应
            if data.get("type") == "pong":
                self._last_heartbeat = time.time()
                return
            
            # 限频响应
            if data.get("code") == 3001:
                retry_after = int(data.get("headers", {}).get("Retry-After", 5))
                logger.warning(f"触发限频，等待 {retry_after}s")
                time.sleep(retry_after)
                return
            
            # depth 频道数据
            if data.get("type") == "depth" or "bid" in data.get("data", {}):
                self._handle_depth(data)
                
        except json.JSONDecodeError:
            logger.warning(f"无法解析消息: {message[:100]}")
        except Exception as e:
            logger.error(f"处理消息时出错: {e}")

    def _handle_depth(self, data: dict):
        """解析 depth 数据，触发回调"""
        try:
            d = data["data"]
            quote = DepthQuote(
                symbol=data.get("symbol", ""),
                timestamp=time.time(),
                bid_price=float(d.get("bid", [0, 0])[0]),
                bid_volume=int(d.get("bid", [0, 0])[1]),
                ask_price=float(d.get("ask", [0, 0])[0]),
                ask_volume=int(d.get("ask", [0, 0])[1]),
            )
            
            if self.on_depth:
                self.on_depth(quote)
                
        except (KeyError, ValueError) as e:
            logger.warning(f"depth 数据解析失败: {e}")

    def _on_error(self, ws, error):
        logger.error(f"WebSocket 错误: {error}")
        self._schedule_reconnect()

    def _on_close(self, ws, close_status_code, close_msg):
        logger.warning(f"WebSocket 关闭: {close_status_code} - {close_msg}")
        self.running = False

    def _schedule_reconnect(self):
        """指数退避重连 + 抖动"""
        self.running = False
        
        # 抖动：避免惊群效应
        jitter = random.uniform(0, self.reconnect_delay * 0.1)
        wait_time = self.reconnect_delay + jitter
        
        logger.info(f"{wait_time:.1f}s 后尝试重连 (当前延迟: {self.reconnect_delay:.1f}s)")
        time.sleep(wait_time)
        
        # 指数退避
        self.reconnect_delay = min(
            self.reconnect_delay * 2,
            self.max_reconnect_delay
        )
        
        if not self.running:
            self.connect()

    def start(self):
        """启动客户端（在独立线程中）"""
        t = threading.Thread(target=self.connect, daemon=True)
        t.start()
        logger.info("TickDB WebSocket 客户端已启动")


# ─────────────────────────────────────────────────────────
# 应用层：做市商行为监控器
# ─────────────────────────────────────────────────────────

class MarketMakerMonitor:
    """
    做市商行为监控器
    订阅 depth 数据，实时计算 OFI-proxy 和流动性指标
    """
    
    def __init__(self, symbols: list[str]):
        self.symbols = symbols
        self.client = TickDBWebSocketClient(symbols)
        
        # 滚动窗口状态（每个 symbol）
        self.mid_history: dict[str, deque[float]] = {
            s: deque(maxlen=1000) for s in symbols
        }
        self.spread_history: dict[str, deque[float]] = {
            s: deque(maxlen=1000) for s in symbols
        }
        self.last_mid: dict[str, float] = {}
        
        # 告警回调
        self.client.on_depth = self._process_quote

    def _process_quote(self, quote: DepthQuote):
        """
        处理 depth 快照，推断 OFI 方向
        注意：depth 不直接包含成交方向，
        我们用 mid-price 的变化作为 OFI 的代理变量
        """
        symbol = quote.symbol
        
        # mid-price 变化作为 OFI 代理
        current_mid = quote.mid_price
        if symbol in self.last_mid:
            prev_mid = self.last_mid[symbol]
            
            if current_mid > prev_mid:
                implied_ofi = +1   # mid 上升 → 买入压力驱动
            elif current_mid < prev_mid:
                implied_ofi = -1   # mid 下降 → 卖出压力驱动
            else:
                implied_ofi = 0    # mid 不变
            
            # 记录历史
            self.mid_history[symbol].append(current_mid)
            self.spread_history[symbol].append(quote.spread)
            
            # 检测异常
            self._check_anomaly(symbol, quote, implied_ofi)
        
        self.last_mid[symbol] = current_mid

    def _check_anomaly(self, symbol: str, quote: DepthQuote, implied_ofi: int):
        """检测做市商行为异常"""
        spreads = list(self.spread_history[symbol])
        mids = list(self.mid_history[symbol])
        
        if len(spreads) < 20:
            return
        
        # 计算当前价差相对历史的分位数
        import numpy as np
        current_spread = quote.spread
        spread_pct = np.percentile(spreads, 95)
        
        # 预警条件：价差急剧扩大 + mid 持续单边移动
        if current_spread > spread_pct:
            recent_mid_changes = [mids[i] - mids[i-1] for i in range(-10, 0)]
            direction_consistency = sum(recent_mid_changes) / len(recent_mid_changes)
            
            if abs(direction_consistency) > 0.001:
                logger.warning(
                    f"[{symbol}] ⚠️ 做市商行为异常预警:\n"
                    f"  当前价差: ${current_spread:.4f} (历史95分位: ${spread_pct:.4f})\n"
                    f"  mid-price 趋势: {'↑' if direction_consistency > 0 else '↓'} {abs(direction_consistency):.4f}\n"
                    f"  推断: 做市商正在{'扩大卖方报价' if direction_consistency > 0 else '扩大买方报价'}\n"
                    f"  建议: 等待流动性回归或确认方向后再操作"
                )

    def run(self):
        """启动监控"""
        logger.info("启动做市商行为监控...")
        self.client.start()
        
        try:
            while True:
                time.sleep(60)
                # 每分钟输出一次状态摘要
                for symbol in self.symbols:
                    mids = list(self.mid_history[symbol])
                    spreads = list(self.spread_history[symbol])
                    if mids and spreads:
                        import numpy as np
                        logger.info(
                            f"[{symbol}] 状态: mid=${np.mean(mids[-10:]):.4f}, "
                            f"平均价差=${np.mean(spreads[-10:]):.4f}, "
                            f"样本数={len(mids)}"
                        )
        except KeyboardInterrupt:
            logger.info("监控停止")


# ─────────────────────────────────────────────────────────
# 入口
# ─────────────────────────────────────────────────────────

if __name__ == "__main__":
    # 监控多只标的
    MONITOR_SYMBOLS = [
        "AAPL.US",
        "TSLA.US", 
        "NVDA.US",
    ]
    
    # ⚠️  高频场景建议使用 aiohttp/asyncio 异步架构
    # ⚠️  此处为同步演示，适合低频监控场景
    
    monitor = MarketMakerMonitor(MONITOR_SYMBOLS)
    monitor.run()

⚠️ 工程提示：上述代码为同步单线程实现，适合分钟级别的监控场景。若需要实时订阅多个标的的高频数据（毫秒级响应），建议使用 asyncio + aiohttp 重构为异步架构，并将 OFI 计算移至独立的进程或线程池中。

---

六、进阶：从单标的分析到跨市场比较

6.1 不同市场的做市商行为差异

做市商的行为模式不是铁板一块，不同市场的微观结构差异会导致其表现形式截然不同：

市场	订单簿深度	价差特征	做市商行为可推断性
美股（大蓝筹）	1 档（NBBO）	极窄（$0.01）	低（数据粒度粗）
美股（小盘股）	1 档（NBBO）	较宽（$0.05-$0.20）	中（价差变化信息量大）
港股	10 档	较宽	高（多档数据可精确重建订单流）
数字货币	10-50 档	极宽（相对值）	高（数据最完整）

这意味着：对于美股大盘股（如 AAPL、MSFT），基于 depth 的 OFI 分析精度有限——1 档数据丢失了大量档位信息。真正的精细化分析需要 FINRA TRF 的逐笔成交数据（商业授权，约 $5,000/月起）。

6.2 库存管理的多周期视角

做市商的库存管理发生在多个时间尺度上：

• 毫秒级：被动撮合 + 即时仓位平衡
• 分钟级：库存偏移调整 + 报价倾斜
• 日内：仓位再平衡（当日清零策略）
• 隔夜：隔夜风险对冲

本文讨论的方法主要覆盖毫秒到分钟级的推断。如果你的策略周期是日间趋势跟踪，那么 OFI 的信号衰减和噪音比例会显著上升——这是需要注意的时间尺度匹配问题。

---

七、结语

回到开篇的闪电崩盘。

2010 年 5 月 6 日那个下午，做市商在价格下跌的第一秒就撤走了流动性。他们的算法做了两件事：一是检测到了异常的单边成交流（OFI 的急剧逆转），二是基于库存风险模型判断“继续报价的期望损失大于撤单的损失”。

这不是阴谋，这是数学。

作为市场参与者，我们当然没有做市商那么完整的内部数据。但通过对 tick 数据和订单簿的持续观测，我们可以无限逼近他们决策的输入变量。OFI、VPIN、价差变化率、库存压力评分——这些指标加在一起，构成了一个做市商行为的实时推断系统。

当然，这个系统永远是不完整的。我们看到的是结果，不是原因。但市场微观结构的迷人之处正在于此：知道结果之后，我们可以更好地理解原因；理解了原因之后，我们至少知道什么时候不该逆势操作。

这比追踪“庄家”要有价值得多。

---

下一步行动

如果你想深入理解市场微观结构：

• 阅读《Misreading the Rain》——Joel Hasbrouck 的市场微观结构教科书
• 阅读 Glosten & Milgrom (1985) 原文——理解做市商定价逻辑的起点

如果你想亲手复现本文的分析框架：

1. 访问 tickdb.ai 注册（免费，无需信用卡）
2. 在控制台生成 API Key
3. 设置环境变量 TICKDB_API_KEY，复制本文 WebSocket 代码即可订阅港股或数字货币的 depth 数据
4. 港股支持 10 档订单簿，比美股 1 档更适合做精细化 OFI 分析

如果你需要 10 年级别的美股历史 K 线数据做策略回测，联系 [email protected] 了解机构方案。

如果你习惯用 AI 辅助开发，在 AI 助手中搜索安装 tickdb-market-data SKILL。

---

风险提示：本文不构成任何投资建议。OFI、VPIN 等指标仅用于市场微观结构分析，不构成交易信号。做市商行为受多重因素影响，基于历史数据的推断存在固有局限性。市场有风险，投资需谨慎。