📚 课程目录第3章买卖价差统计与动态监测

3. 买卖价差统计与动态监测

任务目标

基于实时订单簿的买一/卖一报价，计算价差与相对价差（基点），并监测价差在滚动窗口内的均值、标准差及异常扩张事件，输出结构化监测结果。

核心实现

import time
import math
from collections import deque
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class SpreadMonitor:
    window_sec: int = 60
    spread_bp_threshold: float = 5.0  # 相对价差异常阈值(基点)
    
    # 内部状态
    _quotes: deque = field(default_factory=deque)  # (timestamp, bid, ask)
    
    def _relative_bp(self, bid: float, ask: float) -> Optional[float]:
        """计算相对价差（基点），无效报价返回None"""
        mid = (bid + ask) / 2.0
        if mid <= 0 or bid <= 0 or ask <= 0 or ask < bid:
            return None
        return (ask - bid) / mid * 10000.0
    
    def update(self, bid: float, ask: float) -> dict:
        """输入最新买一/卖一价，返回当前价差状态"""
        now = time.time()
        bp = self._relative_bp(bid, ask)
        
        # 清理过期数据
        cutoff = now - self.window_sec
        while self._quotes and self._quotes[0][0] < cutoff:
            self._quotes.popleft()
        
        # 记录有效报价
        if bp is not None:
            self._quotes.append((now, bid, ask))
        
        # 计算窗口统计
        if not self._quotes:
            return {"valid": False, "reason": "no_valid_quotes"}
        
        spreads_bp = [self._relative_bp(b, a) for _, b, a in self._quotes]
        spreads_bp = [s for s in spreads_bp if s is not None]
        
        if not spreads_bp:
            return {"valid": False, "reason": "no_valid_spreads"}
        
        latest_bp = spreads_bp[-1]
        avg_bp = sum(spreads_bp) / len(spreads_bp)
        var_bp = sum((s - avg_bp) ** 2 for s in spreads_bp) / len(spreads_bp)
        std_bp = math.sqrt(var_bp)
        
        # 异常判定：最新价差超过均值+阈值且超过均值一定比例
        is_abnormal = (latest_bp > avg_bp + self.spread_bp_threshold and
                       latest_bp > avg_bp * 1.2)
        
        return {
            "valid": True,
            "bid": bid,
            "ask": ask,
            "absolute_spread": round(ask - bid, 6),
            "relative_spread_bp": round(latest_bp, 2),
            "window_avg_bp": round(avg_bp, 2),
            "window_std_bp": round(std_bp, 2),
            "sample_count": len(spreads_bp),
            "abnormal_spread": is_abnormal,
            "timestamp": now
        }

# 最小调用示例
if __name__ == "__main__":
    monitor = SpreadMonitor(window_sec=30, spread_bp_threshold=3.0)
    
    # 模拟实时行情流（实际接入时替换为交易所WebSocket/API推送）
    test_feed = [
        (100.00, 100.10),  # 正常价差 ~10bp
        (100.01, 100.11),
        (100.00, 100.09),
        (99.95, 100.15),   # 价差扩张 ~20bp
        (100.00, 100.10),
        (100.02, 100.12),
    ]
    
    for bid, ask in test_feed:
        result = monitor.update(bid, ask)
        if result["valid"]:
            print(f"bid={bid:.2f} ask={ask:.2f} | "
                  f"价差={result['absolute_spread']:.4f} | "
                  f"相对价差={result['relative_spread_bp']:.1f}bp | "
                  f"窗口均值={result['window_avg_bp']:.1f}bp | "
                  f"标准差={result['window_std_bp']:.2f}bp | "
                  f"异常={'是' if result['abnormal_spread'] else '否'}")
        else:
            print(f"跳过无效报价: bid={bid}, ask={ask}")
        time.sleep(0.1)  # 模拟推送间隔

运行与验证

运行方式：直接执行 python spread_monitor.py（需Python 3.7+，无第三方依赖）。

验证路径：

观察输出中第4条记录（99.95/100.15）应触发异常=是，因其相对价差约20bp，明显高于前3条约10bp的均值；
确认窗口统计随新数据滚动更新——第5、6条记录加入后，窗口均值回升，异常标记恢复为否；
将window_sec调小（如5秒）并增加time.sleep(0.2)，可验证过期数据被正确清理，sample_count不会无限增长。

判定通过条件：程序无异常退出；输出中价差扩张样本被标记为异常，且窗口统计值随输入序列变化方向与手工计算一致。

关键说明

本实现聚焦价差计算与滚动监测核心逻辑，实际接入时只需将test_feed替换为交易所的实时行情回调即可；
异常判定采用“阈值+相对比例”双重条件，避免在价差本身很高时误报；阈值参数可按交易品种波动率调整；
相对价差以基点（bp）表示，1bp = 0.01%，适用于不同价格水平的品种间比较。