python案例能否识别盘口异常变动？

wen python案例 2026-09-13 1

Python 识别盘口异常变动案例

可以,Python 非常适合做盘口（Level-2 行情）异常监控，下面给你一套可运行的思路 + 代码框架。

python案例能否识别盘口异常变动？

什么是"盘口异常"

常见的异常类型：

类型	特征
大单挂撤	买卖档位出现远超均值的挂单，然后快速撤掉
挂单失衡	买一~买五 vs 卖一~卖五量级严重不平衡
频繁撤单	单位时间内撤单次数异常
吃单加速	主动性成交突然放大（大单扫货/砸盘）
冰山单	某价位持续成交但挂单量几乎不变
价差异常	spread 突然放大/收窄

数据源准备

常见渠道：

A股：QMT、掘金、聚宽、Tushare Pro、Level-2 行情（付费）
期货：CTP、vnpy
加密：Binance/OKX WebSocket（免费，最适合练手）

下面示例用 JSON 快照格式，通用：

snapshot = {
    "symbol": "BTCUSDT",
    "ts": 1730000000.123,
    "bids": [[100.0, 5.2], [99.9, 3.1], ...],   # [价, 量]
    "asks": [[100.1, 4.8], [100.2, 2.0], ...],
}

完整案例代码

import time
import statistics
from collections import deque, defaultdict
class OrderBookMonitor:
    def __init__(self, window=60, depth=5):
        self.window = window          # 滚动窗口（秒）
        self.depth = depth
        self.history = deque()        # 保存快照历史
        self.bid_vol_hist = deque(maxlen=100)
        self.ask_vol_hist = deque(maxlen=100)
        self.cancel_count = defaultdict(int)
        self.last_book = None
    # ---------- 工具 ----------
    @staticmethod
    def _sum_vol(levels, n):
        return sum(v for _, v in levels[:n])
    def _imbalance(self, bids, asks):
        bv = self._sum_vol(bids, self.depth)
        av = self._sum_vol(asks, self.depth)
        if bv + av == 0:
            return 0
        return (bv - av) / (bv + av)   # [-1, 1]
    # ---------- 核心检测 ----------
    def on_snapshot(self, snap):
        bids, asks = snap["bids"], snap["asks"]
        now = snap["ts"]
        # 1. 买卖失衡
        imb = self._imbalance(bids, asks)
        if abs(imb) > 0.6:
            self._alert(now, "IMBALANCE", f"imbalance={imb:.2f}")
        # 2. 大单挂单（相对历史均值）
        b1 = bids[0][1] if bids else 0
        a1 = asks[0][1] if asks else 0
        if len(self.bid_vol_hist) > 20:
            b_mean = statistics.mean(self.bid_vol_hist)
            a_mean = statistics.mean(self.ask_vol_hist)
            if b_mean > 0 and b1 > b_mean * 5:
                self._alert(now, "BIG_BID", f"b1={b1:.2f} mean={b_mean:.2f}")
            if a_mean > 0 and a1 > a_mean * 5:
                self._alert(now, "BIG_ASK", f"a1={a1:.2f} mean={a_mean:.2f}")
        # 3. 撤单检测（对比上一快照）
        if self.last_book:
            prev_bids = dict(self.last_book["bids"][:self.depth])
            for p, v in bids[:self.depth]:
                prev_v = prev_bids.get(p, 0)
                if prev_v > 0 and v < prev_v * 0.3:   # 挂单被撤 70%+
                    self._alert(now, "CANCEL_BID",
                                f"price={p} {prev_v}->{v}")
        # 4. 价差异常
        if bids and asks:
            spread = asks[0][0] - bids[0][0]
            mid = (asks[0][0] + bids[0][0]) / 2
            rel = spread / mid if mid else 0
            if rel > 0.005:   # 0.5% 价差（阈值按品种调）
                self._alert(now, "WIDE_SPREAD", f"spread={rel:.4%}")
        # 5. 更新历史
        self.bid_vol_hist.append(b1)
        self.ask_vol_hist.append(a1)
        self.last_book = snap
    def _alert(self, ts, kind, msg):
        print(f"[{ts:.3f}] 🚨 {kind}: {msg}")
# ---------- 模拟数据 ----------
if __name__ == "__main__":
    mon = OrderBookMonitor()
    # 正常盘口
    for i in range(30):
        mon.on_snapshot({
            "ts": time.time(),
            "bids": [[100.0 - j*0.01, 2.0] for j in range(5)],
            "asks": [[100.01 + j*0.01, 2.0] for j in range(5)],
        })
    # 突然出现大买单 + 失衡
    mon.on_snapshot({
        "ts": time.time(),
        "bids": [[100.0, 50.0]] + [[99.99 - j*0.01, 2.0] for j in range(4)],
        "asks": [[100.01 + j*0.01, 2.0] for j in range(5)],
    })
    # 立刻撤掉
    mon.on_snapshot({
        "ts": time.time(),
        "bids": [[100.0, 0.5]] + [[99.99 - j*0.01, 2.0] for j in range(4)],
        "asks": [[100.01 + j*0.01, 2.0] for j in range(5)],
    })

输出示例：

[1730000000.123] 🚨 BIG_BID: b1=50.00 mean=2.00
[1730000000.123] 🚨 IMBALANCE: imbalance=0.67
[1730000010.456] 🚨 CANCEL_BID: price=100.0 50.0->0.5

进阶方向

接真实数据（示例：Binance WebSocket）

import asyncio, json, websockets
async def run():
    url = "wss://stream.binance.com:9443/ws/btcusdt@depth20@100ms"
    mon = OrderBookMonitor()
    async with websockets.connect(url) as ws:
        while True:
            data = json.loads(await ws.recv())
            snap = {
                "ts": time.time() * 1000,
                "bids": [[float(p), float(q)] for p, q in data["bids"]],
                "asks": [[float(p), float(q)] for p, q in data["asks"]],
            }
            mon.on_snapshot(snap)
asyncio.run(run())

用统计模型替代硬阈值

Z-Score 检测：z = (x - μ) / σ，|z| > 3 视为异常
EWMA 平滑：对盘口量做指数加权平均，检测偏离
Isolation Forest / LOF：多维度（失衡、撤单率、价差）联合打分
滑窗分位数：>P99 或 <P1 报警

import numpy as np
def zscore_alert(series, x, thresh=3):
    if len(series) < 30: return False
    mu, sd = np.mean(series), np.std(series)
    if sd == 0: return False
    return abs((x - mu) / sd) > thresh