有值但已經錯了：行情數據靜默失效的六類檢測與代價

# 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 以上判為尖峰
    }