本文目录导读：
我来分享一个检测股票盘口异常变动的Python案例，涵盖买卖盘突变、大单涌入等场景。
盘口异常检测Python实现
基础数据结构与异常定义
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
import json
class OrderBookSnapshot:
"""盘口快照数据结构"""
def __init__(self, timestamp, bid_prices, bid_volumes, ask_prices, ask_volumes):
self.timestamp = timestamp # 时间戳
self.bid_prices = bid_prices # 买盘价格列表（从高到低）
self.bid_volumes = bid_volumes # 买盘对应量
self.ask_prices = ask_prices # 卖盘价格列表（从低到高）
self.ask_volumes = ask_volumes # 卖盘对应量
def get_bid_ask_imbalance(self):
"""计算买卖盘不平衡度"""
total_bid = sum(self.bid_volumes)
total_ask = sum(self.ask_volumes)
if total_ask == 0:
return float('inf')
return total_bid / total_ask
class AbnormalDetector:
"""盘口异常检测器"""
def __init__(self, thresholds: Dict = None):
# 默认阈值设置
self.thresholds = {
'volume_spike': 5.0, # 成交量突增倍数
'price_change': 0.02, # 价格变动百分比
'imbalance_ratio': 2.0, # 买卖盘不平衡比例
'large_order_size': 100000, # 大单金额阈值
'cancel_ratio': 0.3, # 撤单比例
'spread_change': 0.01 # 价差变化
}
if thresholds:
self.thresholds.update(thresholds)
# 历史数据存储
self.history = []
核心异常检测算法
class AbnormalDetector:
# ... 续上
def detect_abnormal(self, current: OrderBookSnapshot,
previous: OrderBookSnapshot = None) -> List[Dict]:
"""
检测盘口异常
返回异常事件列表
"""
events = []
# 1. 检测大单异常
large_events = self.detect_large_orders(current)
events.extend(large_events)
# 2. 检测买卖盘不平衡
imbalance_events = self.detect_imbalance(current)
events.extend(imbalance_events)
# 3. 检测价格突变
if previous:
price_events = self.detect_price_spike(current, previous)
events.extend(price_events)
# 4. 检测成交量突增
volume_events = self.detect_volume_spike(current, previous)
events.extend(volume_events)
# 5. 检测价差异常
spread_events = self.detect_spread_anomaly(current, previous)
events.extend(spread_events)
return events
def detect_large_orders(self, snapshot: OrderBookSnapshot) -> List[Dict]:
"""检测大单异常"""
events = []
# 检查买盘大单
for price, volume in zip(snapshot.bid_prices, snapshot.bid_volumes):
if price * volume >= self.thresholds['large_order_size']:
events.append({
'type': 'large_bid',
'price': price,
'volume': volume,
'amount': price * volume,
'timestamp': snapshot.timestamp,
'severity': 'high'
})
# 检查卖盘大单
for price, volume in zip(snapshot.ask_prices, snapshot.ask_volumes):
if price * volume >= self.thresholds['large_order_size']:
events.append({
'type': 'large_ask',
'price': price,
'volume': volume,
'amount': price * volume,
'timestamp': snapshot.timestamp,
'severity': 'high'
})
return events
def detect_imbalance(self, snapshot: OrderBookSnapshot) -> List[Dict]:
"""检测买卖盘严重不平衡"""
events = []
imbalance = snapshot.get_bid_ask_imbalance()
# 检测买盘过强
if imbalance > self.thresholds['imbalance_ratio']:
events.append({
'type': 'bid_overweight',
'imbalance_ratio': imbalance,
'timestamp': snapshot.timestamp,
'severity': 'warning',
'message': f'买盘明显强于卖盘，比率达到{imbalance:.2f}'
})
# 检测卖盘过强
elif imbalance < 1 / self.thresholds['imbalance_ratio']:
events.append({
'type': 'ask_overweight',
'imbalance_ratio': imbalance,
'timestamp': snapshot.timestamp,
'severity': 'warning',
'message': f'卖盘明显强于买盘，比率仅为{imbalance:.2f}'
})
return events
def detect_price_spike(self, current: OrderBookSnapshot,
previous: OrderBookSnapshot) -> List[Dict]:
"""检测价格突变"""
events = []
# 计算买卖中价
curr_mid = (current.bid_prices[0] + current.ask_prices[0]) / 2
prev_mid = (previous.bid_prices[0] + previous.ask_prices[0]) / 2
# 计算价格变动率
if prev_mid == 0:
return events
price_change = abs(curr_mid - prev_mid) / prev_mid
if price_change > self.thresholds['price_change']:
direction = '上涨' if curr_mid > prev_mid else '下跌'
events.append({
'type': 'price_spike',
'price_change': price_change,
'direction': direction,
'current_price': curr_mid,
'previous_price': prev_mid,
'timestamp': current.timestamp,
'severity': 'critical'
})
return events
高级异常检测功能
class AdvancedAbnormalDetector(AbnormalDetector):
"""高级异常检测器"""
def detect_volume_spike(self, current: OrderBookSnapshot,
previous: OrderBookSnapshot) -> List[Dict]:
"""检测成交量突增"""
events = []
# 比较总成交量
curr_volume = sum(current.bid_volumes) + sum(current.ask_volumes)
prev_volume = sum(previous.bid_volumes) + sum(previous.ask_volumes)
if prev_volume == 0:
return events
volume_ratio = curr_volume / prev_volume
if volume_ratio > self.thresholds['volume_spike']:
events.append({
'type': 'volume_spike',
'volume_ratio': volume_ratio,
'current_volume': curr_volume,
'previous_volume': prev_volume,
'timestamp': current.timestamp,
'severity': 'warning'
})
# 检测单侧突增（可能有大单买入/卖出）
curr_bid_volume = sum(current.bid_volumes)
curr_ask_volume = sum(current.ask_volumes)
prev_bid_volume = sum(previous.bid_volumes)
prev_ask_volume = sum(previous.ask_volumes)
if prev_bid_volume > 0:
bid_ratio = curr_bid_volume / prev_bid_volume
if bid_ratio > self.thresholds['volume_spike']:
events.append({
'type': 'bid_volume_spike',
'volume_ratio': bid_ratio,
'timestamp': current.timestamp,
'severity': 'warning',
'message': '买盘成交量突增'
})
if prev_ask_volume > 0:
ask_ratio = curr_ask_volume / prev_ask_volume
if ask_ratio > self.thresholds['volume_spike']:
events.append({
'type': 'ask_volume_spike',
'volume_ratio': ask_ratio,
'timestamp': current.timestamp,
'severity': 'warning',
'message': '卖盘成交量突增'
})
return events
def detect_spread_anomaly(self, current: OrderBookSnapshot,
previous: OrderBookSnapshot) -> List[Dict]:
"""检测买卖价差异常"""
events = []
curr_spread = current.ask_prices[0] - current.bid_prices[0]
prev_spread = previous.ask_prices[0] - previous.bid_prices[0]
if prev_spread == 0:
return events
spread_change = abs(curr_spread - prev_spread) / prev_spread
# 价差突然扩大（可能流动性变差）
if curr_spread > prev_spread and spread_change > self.thresholds['spread_change']:
events.append({
'type': 'spread_widening',
'spread_change': spread_change,
'current_spread': curr_spread,
'previous_spread': prev_spread,
'timestamp': current.timestamp,
'severity': 'warning'
})
# 价差突然缩小（可能有大单报价）
elif curr_spread < prev_spread and spread_change > self.thresholds['spread_change']:
events.append({
'type': 'spread_narrowing',
'spread_change': spread_change,
'timestamp': current.timestamp,
'severity': 'info',
'message': '价差缩小，可能有大单进入'
})
return events
def detect_cancel_suspicious(self, order_book_history: List[Dict]) -> List[Dict]:
"""检测撤单异常（需要连续快照数据）"""
events = []
if len(order_book_history) < 3:
return events
# 分析撤单模式
for i in range(2, len(order_book_history)):
before = order_book_history[i-2]
during = order_book_history[i-1]
after = order_book_history[i]
# 检查大单快速消失（疑似撤单）
for side in ['bid', 'ask']:
before_volumes = before[side + '_volumes']
after_volumes = after[side + '_volumes']
# 模拟检测大单消失
if len(before_volumes) > 0 and len(after_volumes) > 0:
large_before = [v for v in before_volumes if v > 10000]
large_after = [v for v in after_volumes if v > 10000]
# 大单数量突然减少
if len(large_before) > 0 and len(large_after) == 0:
events.append({
'type': 'cancel_suspicious',
'side': side,
'timestamp': after['timestamp'],
'severity': 'high',
'message': f'{side}盘大单突然消失'
})
return events
实时监控与预警系统
class RealTimeMonitor:
"""实时盘口监控系统"""
def __init__(self, detector: AdvancedAbnormalDetector):
self.detector = detector
self.current_snapshot = None
self.previous_snapshot = None
self.event_log = []
self.alert_callbacks = []
def process_snapshot(self, snapshot: OrderBookSnapshot):
"""处理新的盘口快照"""
self.previous_snapshot = self.current_snapshot
self.current_snapshot = snapshot
# 运行异常检测
if self.previous_snapshot:
events = self.detector.detect_abnormal(
self.current_snapshot,
self.previous_snapshot
)
# 记录事件并触发告警
for event in events:
self.event_log.append(event)
self.trigger_alert(event)
def trigger_alert(self, event: Dict):
"""触发告警回调"""
severity = event.get('severity', 'info')
# 打印告警信息
print(f"\n{'='*50}")
print(f"⚠️ 盘口异常告警 (严重度: {severity.upper()})")
print(f"时间: {event.get('timestamp', 'N/A')}")
print(f"类型: {event.get('type', 'N/A')}")
print(f"详情: {event.get('message', event)}")
print(f"{'='*50}\n")
# 调用自定义回调函数
for callback in self.alert_callbacks:
try:
callback(event)
except Exception as e:
print(f"告警回调执行失败: {e}")
def add_alert_callback(self, callback):
"""添加自定义告警回调"""
self.alert_callbacks.append(callback)
def get_stats(self) -> Dict:
"""获取监控统计"""
stats = {
'total_events': len(self.event_log),
'critical_events': sum(1 for e in self.event_log if e['severity'] == 'critical'),
'high_events': sum(1 for e in self.event_log if e['severity'] == 'high'),
'warning_events': sum(1 for e in self.event_log if e['severity'] == 'warning'),
'event_types': {}
}
# 统计事件类型
for event in self.event_log:
etype = event['type']
stats['event_types'][etype] = stats['event_types'].get(etype, 0) + 1
return stats
def export_event_log(self, filename: str):
"""导出事件日志"""
with open(filename, 'w', encoding='utf-8') as f:
json.dump(self.event_log, f, ensure_ascii=False, indent=2)
print(f"事件日志已导出到 {filename}")
模拟测试与演示
def simulate_market_data():
"""模拟生成市场数据流"""
import random
base_price = 100.0
timestamps = []
snapshots = []
# 生成5分钟的数据
for i in range(300):
timestamp = datetime.now() + timedelta(seconds=i)
timestamps.append(timestamp)
# 模拟正常价格波动
price = base_price + random.uniform(-1, 1)
# 模拟异常情况
if i == 50:
# 大单买入
bid_volumes = [random.randint(10000, 20000) for _ in range(5)]
ask_volumes = [random.randint(100, 500) for _ in range(5)]
elif i == 150:
# 大量撤单（卖单减少）
bid_volumes = [random.randint(500, 1000) for _ in range(5)]
ask_volumes = [random.randint(100, 200) for _ in range(5)]
elif i == 200:
# 价格突变
price = base_price + 3.0
bid_volumes = [random.randint(500, 1000) for _ in range(5)]
ask_volumes = [random.randint(500, 1000) for _ in range(5)]
else:
# 正常数据
bid_volumes = [random.randint(500, 1000) for _ in range(5)]
ask_volumes = [random.randint(500, 1000) for _ in range(5)]
# 创建价格档位
bid_prices = [price - i*0.01 for i in range(1, 6)]
ask_prices = [price + i*0.01 for i in range(1, 6)]
snapshot = OrderBookSnapshot(
timestamp,
bid_prices,
bid_volumes,
ask_prices,
ask_volumes
)
snapshots.append(snapshot)
return snapshots
def main():
"""主函数 - 运行盘口异常检测"""
print("开始盘口异常检测系统...\n")
# 初始化检测器
thresholds = {
'volume_spike': 3.0, # 比正常阈值更敏感
'price_change': 0.01, # 更敏感的价格变动
'imbalance_ratio': 1.5, # 更敏感的不平衡检测
'large_order_size': 50000, # 大单金额阈值
'cancel_ratio': 0.2, # 撤单比例
'spread_change': 0.005 # 价差变化
}
detector = AdvancedAbnormalDetector(thresholds)
monitor = RealTimeMonitor(detector)
# 添加自定义告警回调
def custom_alert_callback(event):
if event['severity'] == 'critical':
print("🔴 紧急：检测到严重异常，建议立即关注！")
elif event['severity'] == 'high':
print("🟠 高危：发现大额异常交易")
monitor.add_alert_callback(custom_alert_callback)
# 模拟数据流
snapshot_data = simulate_market_data()
# 处理快照
for i, snapshot in enumerate(snapshot_data):
monitor.process_snapshot(snapshot)
# 输出统计结果
print("\n" + "="*50)
print("📊 盘口异常检测统计")
print("="*50)
stats = monitor.get_stats()
print(f"总事件数: {stats['total_events']}")
print(f"紧急事件: {stats['critical_events']}")
print(f"高危事件: {stats['high_events']}")
print(f"警告事件: {stats['warning_events']}")
print("\n事件类型分布:")
for etype, count in stats['event_types'].items():
print(f" - {etype}: {count}次")
# 导出事件日志
monitor.export_event_log('event_log.json')
return monitor
if __name__ == "__main__":
monitor = main()
关键检测功能总结
- 大单检测：识别买卖盘中的大额订单
- 买卖不平衡检测：发现买盘或卖盘异常偏斜
- 价格突变检测：发现价格异常跳动
- 成交量突增检测：发现成交量异常增加
- 价差异常检测：发现流动性异常变化
- 撤单异常检测：识别可疑撤单行为
使用建议
- 实时性：高频检测建议使用高性能数据结构
- 阈值调整：根据市场特性调整检测阈值
- 多维度验证：结合多种异常信号提高准确率
- 机器学习扩展：可以加入LSTM等模型预测异常
这个系统可以识别多种盘口异常，帮助发现大单建仓、护盘行为、出货迹象等信号,适合作为量化交易系统的辅助决策工具。
