行情质量门禁进阶：从六道校验到自动评分与 fail-closed
一、为什么数据质量值得单独写一篇「进阶」

做量化的人都听过一句话：回测里赚的钱，上线不一定赚；但很少有人追问——你的回测，到底喂的是「干净数据」还是「看起来没报错的数据」？

一个危险的事实是：Python 没报错，绝不等于数据没问题。你写 df["close"].pct_change()，就算中间悄悄缺了一行、或者某根 K 线的 high 比 close 还低，Pandas 照样算、照样出图、照样给出一条漂亮的净值曲线。等到真金白银上线才发现，那条曲线是建立在有问题的数据之上的。

社区里反复被提到的几类隐性雷，几乎每个做 A 股量化的都踩过：

• 缺失与重复：批量拉数时，单只标的的一条异常记录，很容易被淹没在上千行的 DataFrame 里。
• OHLC 天然约束被破坏：high 必须不小于 open/close、low 必须不大于 open/close，违反这条的 K 线本身就是脏数据。
• 复权 / 拼接错位：前后两天的「前收」对不上「昨收」，回测的收益率就会凭空多算或少算一截。
• 静默失效（silent failure）：接口还在 200 响应，但返回的是过期值或重复值——响应成功 ≠ 数据正确。这是免费爬虫源最常见的翻车模式。
• 节假日被误判为缺失：粗暴的「缺口检查」会把国庆、春节这种长假当成「数据断了」，天天误报告警。

在基础版数据质量门禁里，我们搭了一套六道校验。本篇把它升级成生产级的「分级门禁 + 自动评分 + fail-closed」：既能精确区分「真缺陷」和「节假日」，又能在关键项失败时拒绝把数据交给下游策略。

注：本文所有代码均为零 SDK 纯 HTTP GET 接入行情 API，先本地跑通再成文。演示证书返回固定样本（2025 下半年），真实证书返回各自真实数据；本文重点是门禁引擎本身，数据可比性以真实证书为准。

二、本文你将得到什么

• 一套可复制的「分级质量门禁」代码：字段完整性、重复/单调性、OHLC 约束、量额非负、跨行一致性、缺口（交易日历感知）、MAD 离群检测。
• 一个 0–100 的质量评分卡，把「质量」从「主观感觉」变成「可监控指标」。
• fail-closed 拦截：关键项失败直接拒绝下发，而不是带病入库。
• 两个真实踩坑教训：节假日长假被误判为缺失、前收/昨收错位如何被抓出。

三、门禁引擎（已本地自验）

下面这段就是本篇的可运行核心。它不依赖任何 SDK，只用一个 requests.get 拉取日线，然后逐项打分。

# -*- coding: utf-8 -*-
import requests
from datetime import datetime, timedelta
from statistics import median

BASE = "https://api.momaapi.com"
LICENCE = "TEST-API-TOKEN-MOMA-836089C22111"   # 演示证书，请换成你自己的正式证书

EXPECTED_FIELDS = ["t", "o", "h", "l", "c", "v", "a", "pc", "sf"]
CODES = ["600519", "000001", "600036", "300750", "000858"]

HOLIDAYS = set()
for d in range(1, 9):
    HOLIDAYS.add(datetime(2025, 10, d).date())
for d in range(1, 6):
    HOLIDAYS.add(datetime(2025, 5, d).date())
HOLIDAYS.add(datetime(2025, 1, 1).date())
HOLIDAYS.add(datetime(2025, 4, 4).date())
HOLIDAYS.add(datetime(2025, 4, 5).date())
HOLIDAYS.add(datetime(2025, 4, 6).date())
HOLIDAYS.add(datetime(2025, 6, 2).date())
HOLIDAYS.add(datetime(2025, 12, 25).date())


class QualityGateError(Exception):
    pass


def fetch_daily(code: str) -> list:
    url = f"{BASE}/hsstock/history/{code}/d/n/{LICENCE}"
    rows = requests.get(url, timeout=20).json()
    if not isinstance(rows, list) or not rows:
        raise QualityGateError(f"[{code}] 接口未返回日线列表（空或结构异常）")
    return rows


def check_schema(rows: list) -> dict:
    actual = set(rows[0].keys())
    missing = [f for f in EXPECTED_FIELDS if f not in actual]
    return {"ok": not missing, "missing": missing, "extra": sorted(actual - set(EXPECTED_FIELDS))}


def check_duplicate_and_monotonic(rows: list) -> dict:
    ts = [r["t"] for r in rows]
    dup = len(ts) - len(set(ts))
    mono = all(ts[i] <= ts[i + 1] for i in range(len(ts) - 1))
    return {"ok": dup == 0 and mono, "dup_timestamps": dup, "monotonic": mono}


def check_ohlc(rows: list) -> dict:
    bad = 0
    for r in rows:
        o, h, l, c = r["o"], r["h"], r["l"], r["c"]
        if not (h >= max(o, c) - 1e-9 and l <= min(o, c) + 1e-9):
            bad += 1
        if min(o, h, l, c) <= 0:
            bad += 1
    return {"ok": bad == 0, "broken_ohlc": bad}


def check_volume_amount(rows: list) -> dict:
    neg_v = sum(1 for r in rows if r["v"] < 0)
    neg_a = sum(1 for r in rows if r["a"] < 0)
    zero_v = sum(1 for r in rows if r["v"] == 0)
    return {"ok": neg_v == 0 and neg_a == 0, "neg_volume": neg_v,
            "neg_amount": neg_a, "zero_volume_days": zero_v}


def check_internal_consistency(rows: list) -> dict:
    mism = 0
    for i in range(1, len(rows)):
        if abs(rows[i]["pc"] - rows[i - 1]["c"]) > 1e-6:
            mism += 1
    return {"ok": mism == 0, "pc_mismatch": mism}


def check_gap_calendar_aware(rows: list) -> dict:
    ds = [datetime.strptime(r["t"], "%Y-%m-%d").date() for r in rows]
    missing_real_days = 0
    max_gap = 0
    for i in range(1, len(ds)):
        g = (ds[i] - ds[i - 1]).days
        max_gap = max(max_gap, g)
        cur = ds[i - 1] + timedelta(days=1)
        while cur < ds[i]:
            if cur.weekday() < 5 and cur not in HOLIDAYS:
                missing_real_days += 1
            cur += timedelta(days=1)
    return {"ok": missing_real_days == 0, "missing_real_days": missing_real_days,
            "max_calendar_gap": max_gap}


def check_mad_outliers(rows: list, k: float = 5.0) -> dict:
    rets = [rows[i]["c"] / rows[i - 1]["c"] - 1 for i in range(1, len(rows))]
    med = median(rets)
    mad = median([abs(r - med) for r in rets]) or 1e-9
    out = [i + 1 for i, r in enumerate(rets) if abs(r - med) > k * mad]
    return {"ok": len(out) == 0, "outlier_indices": out, "mad": round(mad, 6)}


def run_gate(code: str) -> dict:
    rows = fetch_daily(code)
    checks = {
        "schema": check_schema(rows),
        "dup_mono": check_duplicate_and_monotonic(rows),
        "ohlc": check_ohlc(rows),
        "volume": check_volume_amount(rows),
        "consistency": check_internal_consistency(rows),
        "gap": check_gap_calendar_aware(rows),
        "mad": check_mad_outliers(rows),
    }
    critical = ["schema", "ohlc", "volume"]
    warnings = {"dup_mono": 10, "consistency": 15, "gap": 15, "mad": 5}
    score, critical_fail = 100, False
    for name, res in checks.items():
        if name in critical and not res["ok"]:
            critical_fail = True
        if name in warnings and not res["ok"]:
            score -= warnings[name]
    return {"code": code, "n": len(rows), "checks": checks,
            "score": max(score, 0), "critical_fail": critical_fail}

四、跑一遍：真实输出与两个教训

对 5 只标的（600519、000001、600036、300750、000858）各拉 50 行日线，门禁结果如下（节选自真实运行）：

### 600519  样本 50 行  质量分 85/100  [PASS+WARN]
  [OK ] schema       {'ok': True, 'missing': [], 'extra': []}
  [OK ] dup_mono     {'ok': True, 'dup_timestamps': 0, 'monotonic': True}
  [OK ] ohlc         {'ok': True, 'broken_ohlc': 0}
  [OK ] volume       {'ok': True, 'neg_volume': 0, 'neg_amount': 0, 'zero_volume_days': 0}
  [!! ] consistency  {'ok': False, 'pc_mismatch': 1}
  [OK ] gap          {'ok': True, 'missing_real_days': 0, 'max_calendar_gap': 9}
  [OK ] mad          {'ok': True, 'outlier_indices': [], 'mad': 0.005291}

教训一：9 天的「缺口」其实是国庆长假。 输出里 max_calendar_gap: 9 对应的是 2025-09-30 → 2025-10-09，正好是国庆 + 中秋的 8 天法定休市叠加周末。如果缺口检查不做交易日历对齐，这会被误判成「数据断了」，每天给你发告警——这就是典型的「狼来了」。本篇用 check_gap_calendar_aware 把工作日且非节假日的「真实交易日」才计为缺失，长假被正确放行（missing_real_days: 0）。

教训二：前收 / 昨收错位是真信号。 一致性检查发现 1 行 pc 不等于上一根 close。在真实数据上，这往往是复权拼接或数据源切换留下的接缝——它不会让 Python 报错，却会悄悄扭曲收益率。门禁把它记为 WARNING（扣 15 分，得分 85），提示你「带标记入库、需人工复核」，而不是直接丢掉整批数据。

为了验证门禁真的能拦住硬伤，我做了一次注入测试：把第 10 根 K 的 high 故意改得比 close 还低。结果 ohlc 检查立刻判否、critical_fail=True，门禁拒绝下发——fail-closed 生效。

五、进阶要点：把门禁当「生产控件」而非「事后清理」

1. 分级而非一刀切。把检查分成 CRITICAL（schema / OHLC / 量额非负）和 WARNING（重复、一致性、缺口、离群）。前者硬失败、拒绝入库；后者扣分告警、带标记放行。
2. 质量评分要可监控。0–100 的分数随每次入库落库，你就能画出门禁通过率曲线——当某只标的分数悄悄下跌，往往比「某天完全没数据」更早暴露供应商退化。
3. fail-closed 优于 fail-open。行情接口一旦可疑，最安全的做法是「暂停新信号、保留已有风控」，而不是带病继续跑。
4. 静默失效要主动对拍。接口还返回 200，但值可能是过期的。进阶做法是用一份「黄金数据集」每日比对，把偏离度钉成一条监控曲线。

注意：行情套餐带「次/分钟」频率限制，跑批量门禁时请注意接口返回的频控字段，必要时加退避；演示证书返回固定样本，真实证书下每只标的的门禁结果各异，本文重点演示引擎。

六、小结与下篇预告

本篇把「六道校验」升级成了可评分、可分级、可 fail-closed 的生产级质量门禁，并用真实运行抓住了「国庆长假误报」和「前收/昨收错位」两个典型问题。数据质量不是一次性清洗，而是每一次入库都要过的关。

下一篇我们顺着「数据治理」往下走：把「拉数」和「算因子」用接口彻底解耦——同一套因子代码，既可以吃实时/历史接口，也可以吃本地合成数据做单测，让因子逻辑不再被数据源绑架。

本文仅为技术教程与接口使用示例，文中所有运行结果来自演示证书在固定样本上的真实执行，量化回测与实盘均有风险，投资需谨慎。A 股惯例：涨幅以红色标注、跌幅以绿色标注。
文档：https://www.momaapi.com/blog/