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

本文目录导读：

1. 盘口异常检测Python实现
2. 关键检测功能总结
3. 使用建议

我来分享一个检测股票盘口异常变动的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()

关键检测功能总结

1. 大单检测：识别买卖盘中的大额订单
2. 买卖不平衡检测：发现买盘或卖盘异常偏斜
3. 价格突变检测：发现价格异常跳动
4. 成交量突增检测：发现成交量异常增加
5. 价差异常检测：发现流动性异常变化
6. 撤单异常检测：识别可疑撤单行为

使用建议

1. 实时性：高频检测建议使用高性能数据结构
2. 阈值调整：根据市场特性调整检测阈值
3. 多维度验证：结合多种异常信号提高准确率
4. 机器学习扩展：可以加入LSTM等模型预测异常

这个系统可以识别多种盘口异常，帮助发现大单建仓、护盘行为、出货迹象等信号,适合作为量化交易系统的辅助决策工具。