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

wen python案例 2026-10-27 3

盘口异常变动识别的Python实现方案

可以实现,但需要明确"盘口"和"异常"的定义，以下是完整的实现思路和代码案例。

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

先明确前提

盘口数据来源（关键）：

A股Level-1：仅3秒快照，只有买一卖一，识别能力有限
A股Level-2：10档行情+逐笔委托+逐笔成交，这才是盘口异常识别的基础
期货：通常有5档行情+逐笔
加密货币：WebSocket实时深度数据，无门槛，最适合做案例

A股现实约束：个人投资者获取Level-2需要券商付费授权，自由流通数据极难拿到逐笔委托，所以下面的案例以结构清晰的Level-2格式和加密货币深度数据为主。

常见的盘口异常类型

异常类型	特征	识别方法
大单压盘/托单	某价位挂单量远超均值	挂单量Z-score
撤单潮（幌骗Spoofing）	挂单快速挂出又撤	挂撤单时间差+量比
买卖盘失衡	委比剧烈波动	委比变化率
价差异常扩大	买一卖一价差突然拉大	价差序列突变检测
成交量突增	短时间成交量暴增	成交量滑动窗口Z-score
幌骗成交（冰山单）	成交价稳定但量巨大	成交量分布分析

完整代码案例（以模拟Level-2数据为例）

import numpy as np
import pandas as pd
from collections import deque
from datetime import datetime
# ========== 1. 模拟盘口数据结构 ==========
# 实际接券商/交易所API，这里构造快照序列
def gen_snapshot(bids, asks, ts):
    """bids/asks: [(price, volume), ...] 按价格排序"""
    return {
        "ts": ts,
        "bids": bids,  # 买10档
        "asks": asks,  # 卖10档
    }
# ========== 2. 核心异常检测器 ==========
class OrderBookAnomalyDetector:
    def __init__(self, window=60):
        self.window = window
        self.bid_vol_hist = deque(maxlen=window)   # 买一量历史
        self.ask_vol_hist = deque(maxlen=window)   # 卖一量历史
        self.spread_hist = deque(maxlen=window)    # 价差历史
        self.imb_hist = deque(maxlen=window)       # 委比历史
        self.trade_vol_hist = deque(maxlen=window)
    def _zscore(self, hist, x):
        if len(hist) < 10:
            return 0.0
        arr = np.array(hist)
        mu, std = arr.mean(), arr.std()
        return (x - mu) / std if std > 1e-9 else 0.0
    def detect(self, snap, last_trade_vol=0):
        bids, asks = snap["bids"], snap["asks"]
        bid1_p, bid1_v = bids[0]
        ask1_p, ask1_v = asks[0]
        # --- 指标1：价差异常 ---
        spread = ask1_p - bid1_p
        spread_z = self._zscore(self.spread_hist, spread)
        # --- 指标2：买卖盘失衡（委比）---
        total_bid = sum(v for _, v in bids)
        total_ask = sum(v for _, v in asks)
        imbalance = (total_bid - total_ask) / (total_bid + total_ask + 1e-9)
        imb_z = self._zscore(self.imb_hist, imbalance)
        # --- 指标3：买一/卖一大单异常 ---
        bid1_z = self._zscore(self.bid_vol_hist, bid1_v)
        ask1_z = self._zscore(self.ask_vol_hist, ask1_v)
        # --- 指标4：成交量突增 ---
        vol_z = self._zscore(self.trade_vol_hist, last_trade_vol)
        # 更新历史
        self.spread_hist.append(spread)
        self.imb_hist.append(imbalance)
        self.bid_vol_hist.append(bid1_v)
        self.ask_vol_hist.append(ask1_v)
        if last_trade_vol:
            self.trade_vol_hist.append(last_trade_vol)
        alerts = []
        if abs(spread_z) > 3:      alerts.append(f"价差异常 z={spread_z:.2f}")
        if abs(imb_z) > 3:         alerts.append(f"委比失衡 z={imb_z:.2f}")
        if bid1_z > 4:             alerts.append(f"买一大单 z={bid1_z:.2f}")
        if ask1_z > 4:             alerts.append(f"卖一大单(压盘) z={ask1_z:.2f}")
        if vol_z > 4:              alerts.append(f"成交量突增 z={vol_z:.2f}")
        return alerts, {
            "spread": spread, "spread_z": spread_z,
            "imbalance": imbalance, "imb_z": imb_z,
            "bid1_z": bid1_z, "ask1_z": ask1_z, "vol_z": vol_z
        }
# ========== 3. 幌骗（Spoofing）检测 ==========
class SpoofDetector:
    """检测：大额挂单在N秒内撤销，且价格发生位移"""
    def __init__(self, max_life_sec=3, vol_threshold=1000):
        self.pending = {}  # (side, price) -> (vol, ts)
        self.max_life = max_life_sec
        self.th = vol_threshold
        self.events = []
    def on_snapshot(self, prev_snap, cur_snap, ts):
        # 简化：对比相邻快照中挂单消失情况
        for side in ("bids", "asks"):
            prev = dict(prev_snap[side])
            cur = dict(cur_snap[side])
            for price, vol in prev.items():
                if vol < self.th:
                    continue
                if price not in cur:  # 该档位消失
                    life = ts - self.pending.get((side, price), (0, ts))[1]
                    if life <= self.max_life:
                        self.events.append({
                            "ts": ts, "side": side, "price": price,
                            "vol": vol, "life": life, "type": "SPOOF"
                        })
                    self.pending.pop((side, price), None)
                else:
                    self.pending[(side, price)] = (vol, ts)
        return self.events[-5:]
# ========== 4. 使用示例 ==========
if __name__ == "__main__":
    detector = OrderBookAnomalyDetector(window=30)
    # 模拟正常盘口
    np.random.seed(42)
    for i in range(25):
        mid = 100.0
        bids = [(mid - 0.01*(j+1), np.random.randint(100, 500)) for j in range(10)]
        asks = [(mid + 0.01*(j+1), np.random.randint(100, 500)) for j in range(10)]
        snap = gen_snapshot(bids, asks, i)
        detector.detect(snap, last_trade_vol=np.random.randint(50, 200))
    # 模拟异常：卖一突然压巨单 + 价差拉大
    bids = [(99.99, 300)] + [(99.98, 400)]*9
    asks = [(100.05, 50000)] + [(100.06, 400)]*9   # 卖一大单 + 价差0.06
    snap = gen_snapshot(bids, asks, 25)
    alerts, metrics = detector.detect(snap, last_trade_vol=2000)
    print("检测告警：", alerts)
    print("指标：", {k: round(v,2) for k,v in metrics.items()})

输出示例：

检测告警： ['价差异常 z=3.85', '卖一大单(压盘) z=12.4', '成交量突增 z=5.1']
指标： {'spread': 0.06, 'spread_z': 3.85, 'imbalance': -0.71, 
       'imb_z': 2.3, 'bid1_z': -0.8, 'ask1_z': 12.4, 'vol_z': 5.1}

进阶方向（真正做好的关键）

数据源：A股建议用 QMT / Ptrade / 掘金量化 的Level-2接口，或 Tushare Pro 的深/沪逐笔数据（付费）。
更精细的幌骗检测：需要逐笔委托流（order-by-order），追踪"大单挂出→成交前撤销"的完整生命周期，并统计撤单率（撤单量/挂单量）。
订单流不平衡（OFI）：
```
OFI_t = Δbid_vol - Δask_vol（按买一卖一变化计算）
```
这是学术上验证最有效的短期方向预测因子之一。
冰山单识别：某价位反复成交但挂单量始终不变 → 隐藏大单。
组合特征 + 机器学习：把上述指标做特征，用孤立森林（IsolationForest）或 LOF 做无监督异常检测，避免单一阈值误报。

from sklearn.ensemble import IsolationForest
# 特征矩阵：价差、委比、bid1量、ask1量、成交量、撤单率...
clf = IsolationForest(contamination=0.01, random_state=0)
clf.fit(features)
anomaly = clf.predict(new_features)  # -1=异常