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

wen python案例 2026-11-03 3

用Python识别盘口异常变动：完整案例

盘口异常识别在量化交易、风控和反欺诈中都有实际需求，下面给你一套可落地的Python方案，包含多种异常类型识别和完整代码。

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

盘口异常的主要类型

异常类型	特征	常见场景
挂单量异常	单一价位挂单远超历史水平	虚假挂单/诱导
撤单异常	短时间内大量撤单	幌骗（Spoofing）
价差异常	买卖价差异常扩大/收窄	流动性骤变
深度失衡	买卖盘力量严重不对称	大单压盘/托盘
瞬时波动	盘口数据突变	闪崩/拉盘
挂单墙	突兀的巨量挂单	阻力/支撑操纵

完整实现代码

基础数据结构与模拟数据

import pandas as pd
import numpy as np
from collections import deque
import matplotlib.pyplot as plt
np.random.seed(42)
def generate_orderbook_data(n=1000, anomaly_points=[]):
    """
    生成模拟盘口数据（真实场景用交易所API/std接口）
    字段：timestamp, bid1_price, bid1_vol, ask1_price, ask1_vol, ...
    """
    data = []
    base_price = 100.0
    for i in range(n):
        # 正常波动
        mid = base_price + np.random.normal(0, 0.02)
        spread = 0.02 + abs(np.random.normal(0, 0.005))
        # 注入异常
        if i in anomaly_points:
            spread *= 5      # 价差突变
            vol_mult = 20    # 挂单量异常
        else:
            vol_mult = 1
        bid1 = mid - spread / 2
        ask1 = mid + spread / 2
        bid_vol = max(int(np.random.normal(500, 100)), 1) * vol_mult
        ask_vol = max(int(np.random.normal(500, 100)), 1) * vol_mult
        data.append({
            'timestamp': pd.Timestamp('2024-01-01') + pd.Timedelta(seconds=i),
            'bid1_price': bid1,
            'bid1_vol': bid_vol,
            'ask1_price': ask1,
            'ask1_vol': ask_vol,
        })
    return pd.DataFrame(data)

异常检测引擎

class OrderbookAnomalyDetector:
    def __init__(self, window=50, z_threshold=3.0):
        self.window = window
        self.z_threshold = z_threshold
        self.history = {
            'spread': deque(maxlen=window),
            'imbalance': deque(maxlen=window),
            'bid_vol': deque(maxlen=window),
            'ask_vol': deque(maxlen=window),
        }
    def detect(self, row):
        """对单条盘口快照做多维度异常检测"""
        anomalies = []
        # 1. 价差
        spread = row['ask1_price'] - row['bid1_price']
        # 2. 盘口失衡度 = (买量-卖量)/(买量+卖量)，范围[-1,1]
        imbalance = ((row['bid1_vol'] - row['ask1_vol']) /
                     (row['bid1_vol'] + row['ask1_vol'] + 1e-9))
        # 3. 挂单量（对数处理）
        log_bid = np.log1p(row['bid1_vol'])
        log_ask = np.log1p(row['ask1_vol'])
        # Z-score异常检测
        def check_z(name, value):
            hist = list(self.history[name])
            if len(hist) < 20:  # 冷启动
                return None
            mu, sigma = np.mean(hist), np.std(hist) + 1e-9
            z = (value - mu) / sigma
            return z
        checks = {
            'spread_pct': check_z('spread', spread),
            'imbalance_z': check_z('imbalance', abs(imbalance)),
            'bid_vol_z': check_z('bid_vol', log_bid),
            'ask_vol_z': check_z('ask_vol', log_ask),
        }
        for name, z in checks.items():
            if z is not None and abs(z) > self.z_threshold:
                anomalies.append({
                    'type': name,
                    'z_score': round(z, 2),
                    'value': value if (value:=row.get(name.replace('_z','_pct'))) else None
                })
        # 4. 断崖式撤单：当前量比上一时刻骤降
        if self.history['bid_vol']:
            prev_bid = self.history['bid_vol'][-1]
            if prev_bid > 0 and row['bid1_vol'] / prev_bid < 0.2:
                anomalies.append({'type': 'bid_vol_drop', 
                                  'ratio': round(row['bid1_vol']/prev_bid, 3)})
            prev_ask = self.history['ask_vol'][-1]
            if prev_ask > 0 and row['ask1_vol'] / prev_ask < 0.2:
                anomalies.append({'type': 'ask_vol_drop',
                                  'ratio': round(row['ask1_vol']/prev_ask, 3)})
        # 更新历史
        self.history['spread'].append(spread)
        self.history['imbalance'].append(abs(imbalance))
        self.history['bid_vol'].append(log_bid)
        self.history['ask_vol'].append(log_ask)
        return anomalies

运行检测

# 生成数据，在第300、600、800条注入异常
df = generate_orderbook_data(1000, anomaly_points=[300, 600, 800])
detector = OrderbookAnomalyDetector(window=50, z_threshold=3.0)
results = []
for idx, row in df.iterrows():
    anoms = detector.detect(row)
    if anoms:
        results.append({'idx': idx, 'timestamp': row['timestamp'], 'anomalies': anoms})
# 输出
print(f"共检测到 {len(results)} 条异常记录\n")
for r in results[:10]:
    print(f"#{r['idx']} @ {r['timestamp'].strftime('%H:%M:%S')}")
    for a in r['anomalies']:
        print(f"    - {a['type']}: {a}")

预期输出示例：

共检测到 8 条异常记录
#300 @ 00:05:00
    - spread_z: {'type': 'spread_z', 'z_score': 15.3, ...}
    - bid_vol_z: {'type': 'bid_vol_z', 'z_score': 8.7, ...}
    - ask_vol_z: {'type': 'ask_vol_z', 'z_score': 9.1, ...}
#600 ...

进阶：识别"幌骗"（Spoofing）

幌骗的典型特征是大单挂出后快速撤单，价格反向波动，用事件流方法识别：

class SpoofingDetector:
    """基于事件流识别幌骗"""
    def __init__(self, big_order_mult=10, fast_cancel_ms=1000):
        self.big_order_mult = big_order_mult
        self.fast_cancel_ms = fast_cancel_ms
        self.big_orders = {}  # order_id -> (price, vol, ts, side)
        self.avg_vol = 500
    def on_order_event(self, event):
        """
        event: {type: 'add'|'cancel', order_id, price, vol, side, ts}
        """
        t = event['type']
        oid = event['order_id']
        flags = []
        if t == 'add' and event['vol'] > self.big_order_mult * self.avg_vol:
            self.big_orders[oid] = (event['price'], event['vol'],
                                    event['ts'], event['side'])
        elif t == 'cancel' and oid in self.big_orders:
            price, vol, ts0, side = self.big_orders.pop(oid)
            dt = (event['ts'] - ts0).total_seconds() * 1000
            if dt < self.fast_cancel_ms:
                flags.append({
                    'type': 'potential_spoofing',
                    'order_id': oid,
                    'vol': vol,
                    'side': side,
                    'lifetime_ms': dt
                })
        return flags

工程化建议

数据源：A股用 Level-2 快照（3秒一笔），期货/加密用逐笔委托，用 websocket 接入（如 ccxt、期货CTP接口、vnpy）。
性能：deque + 环状窗口做流式计算；生产环境用 numba/polars 或用 Redis Stream 做冗余。
误报控制：
- 用 EWMA 替代简单均值，适应不同标的波动率
- 加"持续时长"过滤（连续 N 帧异常才算）
- 分时段建模（开盘/收盘波动天然大）
多因子组合：单一 z-score 易误报，用规则引擎组合（如"量异常 + 撤单快 + 价格反向"三条件）。
回放验证：用历史数据回测，用 同标的正常交易日基线 对比阈值。