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

Python 识别盘口异常变动案例

可以,Python 非常适合做盘口异常监控，下面给你一个完整的实战案例。

常见的盘口异常类型

完整代码示例（模拟数据）

import numpy as np
import pandas as pd
from collections import deque
from datetime import datetime
class OrderBookMonitor:
    def __init__(self, window=20, 
                 volume_ratio_thresh=5.0,
                 imbalance_thresh=3.0,
                 cancel_ratio_thresh=0.7):
        self.window = window                          # 滑动窗口
        self.volume_ratio_thresh = volume_ratio_thresh # 挂单量倍数阈值
        self.imbalance_thresh = imbalance_thresh       # 失衡比阈值
        self.cancel_ratio_thresh = cancel_ratio_thresh # 撤单比例阈值
        self.history = deque(maxlen=window)
        self.last_snapshot = None
    def update(self, snapshot: dict):
        """
        snapshot 结构:
        {
          'timestamp': ...,
          'bids': [(price, vol), ...5档, 由高到低],
          'asks': [(price, vol), ...5档, 由低到高]
        }
        """
        alerts = []
        bids = snapshot['bids']
        asks = snapshot['asks']
        # ---------- 1. 挂单量异常 ----------
        if len(self.history) >= self.window:
            avg_vol = self._avg_volume()
            top_bid_vol = bids[0][1]
            top_ask_vol = asks[0][1]
            if top_bid_vol > avg_vol * self.volume_ratio_thresh:
                alerts.append(('大额买单挂单', 
                               f'买一量 {top_bid_vol} 是均值 {avg_vol:.0f} 的 '
                               f'{top_bid_vol/avg_vol:.1f} 倍'))
            if top_ask_vol > avg_vol * self.volume_ratio_thresh:
                alerts.append(('大额卖单挂单',
                               f'卖一量 {top_ask_vol} 是均值的 '
                               f'{top_ask_vol/avg_vol:.1f} 倍'))
        # ---------- 2. 盘口失衡 ----------
        bid_sum = sum(v for _, v in bids[:5])
        ask_sum = sum(v for _, v in asks[:5])
        if ask_sum > 0:
            imbalance = bid_sum / ask_sum
            if imbalance > self.imbalance_thresh:
                alerts.append(('买盘失衡', f'买卖比 {imbalance:.2f}'))
            elif imbalance < 1 / self.imbalance_thresh:
                alerts.append(('卖盘失衡', f'买卖比 {imbalance:.2f}'))
        # ---------- 3. 价差异常 ----------
        spread = asks[0][0] - bids[0][0]
        if len(self.history) >= self.window:
            avg_spread = self._avg_spread()
            if spread > avg_spread * 3:
                alerts.append(('价差扩大', f'当前价差 {spread} vs 均值 {avg_spread:.3f}'))
        # ---------- 4. 撤单检测 ----------
        if self.last_snapshot:
            cancel_alerts = self._detect_cancel(self.last_snapshot, snapshot)
            alerts.extend(cancel_alerts)
        # ---------- 5. 扫单检测 ----------
        if self.last_snapshot:
            sweep = self._detect_sweep(self.last_snapshot, snapshot)
            if sweep:
                alerts.append(sweep)
        self.history.append(snapshot)
        self.last_snapshot = snapshot
        return alerts
    # -------- 辅助方法 --------
    def _avg_volume(self):
        vols = []
        for s in self.history:
            vols.append(s['bids'][0][1])
            vols.append(s['asks'][0][1])
        return np.mean(vols)
    def _avg_spread(self):
        return np.mean([s['asks'][0][0] - s['bids'][0][0] for s in self.history])
    def _detect_cancel(self, prev, curr):
        alerts = []
        # 买一大幅撤单
        if prev['bids'][0][1] > 0:
            cancel_ratio = (prev['bids'][0][1] - curr['bids'][0][1]) / prev['bids'][0][1]
            if cancel_ratio > self.cancel_ratio_thresh and curr['bids'][0][0] == prev['bids'][0][0]:
                alerts.append(('买一撤单', f'撤单比例 {cancel_ratio:.0%}'))
        # 卖一大幅撤单
        if prev['asks'][0][1] > 0:
            cancel_ratio = (prev['asks'][0][1] - curr['asks'][0][1]) / prev['asks'][0][1]
            if cancel_ratio > self.cancel_ratio_thresh and curr['asks'][0][0] == prev['asks'][0][0]:
                alerts.append(('卖一撤单', f'撤单比例 {cancel_ratio:.0%}'))
        return alerts
    def _detect_sweep(self, prev, curr):
        """如果卖一价格被上移多档，说明有人扫货"""
        prev_ask_px = prev['asks'][0][0]
        curr_ask_px = curr['asks'][0][0]
        curr_bid_px = curr['bids'][0][0]
        # 主动买入：最新买价 ≥ 之前卖一价
        if curr_bid_px >= prev_ask_px:
            ticks_moved = sum(1 for p, _ in prev['asks'] if p <= curr_bid_px)
            return ('向上扫单', f'吃穿 {ticks_moved} 档，价格从 {prev_ask_px} → {curr_bid_px}')
        return None
# ==================== 使用示例 ====================
if __name__ == '__main__':
    monitor = OrderBookMonitor(window=20)
    # 模拟 30 个 tick，其中第 25 个 tick 出现"大额挂单+撤单"异常
    np.random.seed(42)
    for i in range(30):
        base = 10.00
        bids = [(round(base - 0.01*j, 2), int(np.random.normal(1000, 200))) 
                for j in range(5)]
        asks = [(round(base + 0.01*j, 2), int(np.random.normal(1000, 200))) 
                for j in range(5)]
        # 异常注入
        if i == 25:
            bids[0] = (base, 8000)   # 买一巨量挂单
        if i == 26:
            # 撤掉一半
            bids[0] = (base, 3000)
        snap = {'timestamp': datetime.now(), 'bids': bids, 'asks': asks}
        alerts = monitor.update(snap)
        for name, detail in alerts:
            print(f'[Tick {i}] ⚠️  {name}: {detail}')

实际场景接入真实数据

股票（Tick 数据）

# 通过 tushare / akshare / 掘金 / 聚宽获取
import akshare as ak
# 或用券商 CTP / 极速柜台 API 实时订阅

加密货币（深度数据，实时性最好）

import websocket, json
def on_message(ws, message):
    data = json.loads(message)
    # 币安 depth20@100ms
    bids = [(float(p), float(q)) for p, q in data['bids']]
    asks = [(float(p), float(q)) for p, q in data['asks']]
    alerts = monitor.update({'bids': bids, 'asks': asks})
    for a in alerts:
        print(a)
ws = websocket.WebSocketApp(
    "wss://stream.binance.com:9443/ws/btcusdt@depth20@100ms",
    on_message=on_message
)
ws.run_forever()

进阶思路（实战中常用）

1. OFI（Order Flow Imbalance）
   用挂单变化量建模，识别主力方向：

   OFI = ΔBid_vol - ΔAsk_vol（含撤单）

2. VPIN（成交量同步信息毒性）
   检测知情交易者，识别异常成交。

3. 机器学习方法

   • 特征：挂单量比、撤单率、价差、扫单深度、成交方向占比……
   • 模型：Isolation Forest / LOF / LSTM 做无监督异常检测

4. 分层缓存 + 异步处理
   盘口变化极快，用 asyncio + numpy 或 Cython 加速计算；用 Redis 存储快照。

5. 警告级别
   分轻微/中等/严重，配合可视化（Plotly/Matplotlib 实时画热力图）。

注意事项

• A 股 Level-1 只有 3 秒快照，识别精度有限；Level-2（逐笔委托+逐笔成交）才能真正看到撤单和挂单过程。
• 交易所直连 vs 第三方 API 延迟差几个数量级，做高频必须直连。
• 合规：识别"幌骗（Spoofing）"等行为可以自用研究，但不能据此操纵市场。

需要我针对某个具体市场（A 股/期货/加密）或某种具体异常（比如幌骗、扫单、冰山单识别）给一套更专用的代码吗？