有值但已经错了：行情数据静默失效的六类检测与代价

# quality.py — 行情数据质量校验：六条廉价断言，仅依赖标准库
from collections import deque
from datetime import datetime, timezone


class QualityGate:
    """median_window: 尖峰检测窗口长度；spike_k: MAD 倍数阈值；stale_after: 陈旧阈值（秒）。"""

    def __init__(self, median_window=200, spike_k=8.0, stale_after=60.0):
        self.prices = deque(maxlen=median_window)
        self.spike_k = spike_k
        self.stale_after = stale_after
        self.last_at = {}
        self.last_vol = {}

    def check(self, tick, now=None):
        now = now or datetime.now(timezone.utc)
        sym = tick["symbol"]
        out = []
        at = datetime.fromisoformat(tick["at"].replace("Z", "+00:00"))

        # 1 陈旧值：帧仍在到达，时间戳不再前进
        lag = (now - at).total_seconds()
        if lag > self.stale_after:
            out.append("stale %s lag=%.0fs" % (sym, lag))

        # 2 时间戳回退：同一标的后一条早于前一条
        prev = self.last_at.get(sym)
        if prev is not None and at < prev:
            out.append("time_regression %s" % sym)
        self.last_at[sym] = at if prev is None else max(at, prev)

        last = tick.get("last")

        # 3 价格尖峰：相对滚动中位数的 MAD 偏离
        if isinstance(last, (int, float)):
            if len(self.prices) >= 30:
                med = sorted(self.prices)[len(self.prices) // 2]
                mad = sorted(abs(x - med) for x in self.prices)[len(self.prices) // 2]
                if mad and abs(last - med) > self.spike_k * 1.4826 * mad:
                    out.append("spike %s %s" % (sym, last))
            self.prices.append(last)

        # 4 有价无量的伪更新：volume 未变而 last 变化
        vol = tick.get("volume")
        if isinstance(vol, (int, float)) and isinstance(last, (int, float)):
            pv = self.last_vol.get(sym)
            if pv is not None and pv[0] == vol and pv[1] != last:
                out.append("phantom_update %s" % sym)
            self.last_vol[sym] = (vol, last)

        # 5 盘口交叉 / 非正报价：不得进入中间价计算
        bid, ask = tick.get("bid"), tick.get("ask")
        if isinstance(bid, (int, float)) and isinstance(ask, (int, float)):
            if bid <= 0 or ask <= 0 or bid > ask:
                out.append("crossed_book %s bid=%s ask=%s" % (sym, bid, ask))

        return out


def check_bar(bar):
    """第 6 类：K 线内部自洽性，落库前逐根断言。"""
    o, h, l, c = (bar[k] for k in ("open", "high", "low", "close"))
    if h < l or h < max(o, c) or l > min(o, c):
        return ["inconsistent_bar %s" % bar.get("openTime")]
    return []

# thresholds.py — 从历史分布反推阈值，不要写死全局常量
import statistics as st


def suggest_thresholds(bars):
    """bars 取自 /api/v1/market/equity/kline 的 bars[] 数组。"""
    gaps = [bars[i]["closeTime"] - bars[i - 1]["closeTime"] for i in range(1, len(bars))]
    steps = [abs(bars[i]["close"] - bars[i - 1]["close"]) for i in range(1, len(bars))]
    med = st.median(steps)
    mad = st.median([abs(s - med) for s in steps]) or med * 0.01
    return {
        "stale_after_ms": int(st.median(gaps) * 5),     # 连续 5 个周期无更新即判陈旧
        "spike_abs": round(med + 8 * 1.4826 * mad, 6),  # 8 倍 MAD 以上判为尖峰
    }