2. 买卖一档价差异常的阈值扫描

任务拆解：从分钟快照到价差异常片段

本例只处理一个指标：买一价与卖一价的价差。定义如下（参照约定：价差 = 卖一价 − 买一价，单位与报价一致，通常为元）：

• 正常盘口：卖一价 > 买一价，价差 > 0，且通常等于最小报价单位（如 0.01 元）或其整数倍。
• 价差放大：价差显著大于该股票自身的常态水平，说明盘口瞬时变薄。
• 价差倒挂：卖一价 < 买一价，价差 < 0，属于撮合异常或数据异常，需要单独标记。

固定阈值（例如“价差 > 0.05 元即异常”）的问题在于：高价股与低价股的绝对价差天然不同，同一只股票在开盘、午间、尾盘的量级也不同。因此本例用分位数 + 滚动统计构造动态阈值，并与固定阈值做检出对比。

数据准备与指标计算

输入为单只股票的分钟级盘口快照，至少包含时间戳、买一价、卖一价。下面用可运行的 Python 完成全流程，示例数据为构造的教学数据，用于演示计算与判读逻辑，不代表真实行情。

import numpy as np
import pandas as pd

# 构造教学示例：240 分钟快照，前 200 分钟价差在 0.01~0.02 波动，
# 第 120~124 分钟出现价差放大，第 180~182 分钟出现倒挂
rng = np.random.default_rng(7)
n = 240
mid = 10.00 + np.cumsum(rng.normal(0, 0.004, n))   # 模拟中间价缓慢漂移
spread = rng.choice([0.01, 0.01, 0.01, 0.02], size=n)
spread[120:125] = [0.06, 0.08, 0.07, 0.09, 0.05]   # 价差放大片段
spread[180:183] = [-0.01, -0.02, -0.01]            # 倒挂片段

bid1 = np.round(mid - spread / 2, 2)
ask1 = np.round(mid + spread / 2, 2)
df = pd.DataFrame({
    "ts": pd.date_range("2024-06-03 09:31", periods=n, freq="1min"),
    "bid1": bid1,
    "ask1": ask1,
})
df["spread"] = (df["ask1"] - df["bid1"]).round(4)
df["inverted"] = df["spread"] < 0
print(df.head(3).to_string(index=False))

关键点：价差必须用同一时刻的买一价与卖一价相减，不能跨快照拼接；四舍五入到报价精度后再比较，避免浮点误差把 0.01 判成 0.009999。

动态阈值：分位数定基准，滚动统计定自适应

两步走：先用历史分位数给出“常态上限”，再用滚动中位数与 MAD 做局部自适应，二者取较严者作为当分钟阈值。

# 1) 全局分位数基准（只用前 60 分钟做“冷启动”估计，避免未来信息）
warm = df["spread"].iloc[:60]
q_hi = warm.quantile(0.95)          # 常态价差 95 分位
print("冷启动 95 分位:", q_hi)

# 2) 滚动稳健统计：窗口 30 分钟，中位数 + 3*1.4826*MAD
win = 30
roll_med = df["spread"].rolling(win, min_periods=10).median()
mad = (df["spread"] - roll_med).abs().rolling(win, min_periods=10).median()
sigma_robust = 1.4826 * mad
dyn_thr = np.maximum(q_hi, roll_med + 3 * sigma_robust)

df["dyn_thr"] = dyn_thr
df["flag_dyn"] = df["spread"] > df["dyn_thr"]
df["flag_fix"] = df["spread"] > 0.05          # 固定阈值对照
df["flag_inv"] = df["inverted"]

print(df.loc[118:126, ["ts","spread","dyn_thr","flag_dyn","flag_fix"]].to_string(index=False))

说明：1.4826 是把 MAD 折算成正态标准差的常用系数；np.maximum 保证动态阈值不低于冷启动分位基准，防止盘口长期平静时阈值被压得过低而误报。倒挂单独用 spread < 0 判定，不参与放大阈值比较。

固定阈值与动态阈值的检出对比

def segments(mask):
    """把布尔序列压缩成连续片段 [(start_idx, end_idx), ...]"""
    out, start = [], None
    for i, v in enumerate(mask):
        if v and start is None:
            start = i
        elif not v and start is not None:
            out.append((start, i - 1)); start = None
    if start is not None:
        out.append((start, len(mask) - 1))
    return out

for name, col in [("固定阈值", "flag_fix"), ("动态阈值", "flag_dyn"), ("倒挂", "flag_inv")]:
    segs = segments(df[col].values)
    print(f"{name}: 命中 {int(df[col].sum())} 分钟, 片段 {segs}")

预期现象（基于上述构造数据可直接推出）：固定阈值 0.05 元只能抓到价差放大片段中较大的几根，且对倒挂完全无感；动态阈值能覆盖放大片段的起始与收尾，倒挂则由独立规则全部命中。若把固定阈值调低到 0.02，则会大量误报正常波动——这正是需要动态阈值的原因。

结果判读与参数取舍

实操建议：冷启动窗口不少于 30 个快照；滚动窗口 20~40 分钟、min_periods 设为窗口一半左右；放大与倒挂分开输出，倒挂片段应回查原始快照确认是否为字段错误。最终输出字段建议为 ts, spread, dyn_thr, flag_dyn, flag_fix, flag_inv，便于逐分钟复核与片段合并。