

你是不是也有过这种经历：

- 回测结果看着很美，实盘一跑就崩，回头查数据发现**某段行情根本不在库里**。
- 某天因子值突然全是 NaN，查了半天才发现是**上游接口字段改名**。
- 多只股票一起跑，跑完才发现其中**有两只从头到尾就没拉到数据**。

数据问题不会被代码报错暴露，它只会安静地让你的策略失效。与其事后救火，不如给数据流水线配一张**质量评分卡**——每天跑完自动打分，哪里掉链子一眼看见。

## 本文你将得到什么

1. **5 个维度的数据质量评分模型**：完整性、及时性、一致性、有效性、覆盖度。
2. **一份可直接复制运行的 Python 评分卡代码**，输入是你的 K 线 DataFrame，输出是评分报告 + 告警。
3. **阈值告警机制**：评分低于阈值自动打 `[ALERT]`，方便接入飞书/钉钉/企微机器人。
4. **把评分卡挂到流水线里的最佳实践**：放在 `monitor` 阶段，失败即停，不让脏数据进模型。

## 为什么数据质量必须用"评分卡"而不是靠肉眼

很多开发者的数据检查是临时性的：

```python
df.isna().sum()  # 看一下有没有空值
df.tail()        # 看一下最新日期
```

这在 3 只股票、80 天数据时还够用；一旦扩展到全市场、多数据源、长期运行，就会面临三个问题：

- **维度太多**：空值只是冰山一角，还要关心价格逻辑、多表对齐、最新日期、覆盖度。
- **没有基线**：今天 "NaN 占比 2%" 是好是坏？没有历史分数对比，无法判断。
- **无法告警**：等肉眼发现问题，往往已经污染了下游特征和回测结果。

评分卡解决的是"把主观判断变成客观分数"。

## 5 维评分模型设计

| 维度 | 含义 | 典型问题 |
|------|------|----------|
| 完整性 | 核心字段非空占比 | 接口超时导致某列全 NaN |
| 及时性 | 最新数据距离今天多久 | 采集任务失败导致缺最近几天 |
| 一致性 | 价格逻辑是否自洽 | high < low、close 不在 [low, high] 内 |
| 有效性 | 收益率是否在合理区间 | 极端异常值、未复权导致的跳变 |
| 覆盖度 | 目标股票是否全部拉到 | 某只代码退市、接口漏返回 |

每个维度 0–100 分，最后算综合分。你可以按业务重要性给不同维度加权，但入门阶段先等权即可。

## 代码实战：数据质量评分卡

下面这段代码假设你已经会用 `mairui` 的 `stock_history` 拿到日线。如果还没接触过，可以先看本系列前几篇的清洗/特征/流水线内容。

```python
import json
import sys
from datetime import datetime
from pathlib import Path

import pandas as pd
from mairui import Client

# 演示证书，请换成你自己的正式证书
client = Client("LICENCE-66D8-9F96-0C7F0FBCD073")

CODES = ["600519", "000001", "000002", "600036", "000858"]
REPORT_PATH = Path(__file__).parent / "quality_report.json"


def fetch_daily(code: str, lt: int = 60) -> pd.DataFrame:
    raw = client.stock_history(code, lt=lt, dividend="n")
    df = pd.DataFrame(raw)
    if df.empty:
        return df
    df["t"] = pd.to_datetime(df["t"])
    df["code"] = code
    for col in ["o", "c", "h", "l", "v"]:
        df[col] = pd.to_numeric(df[col], errors="coerce")
    return df[["code", "t", "o", "c", "h", "l", "v"]]


def score_completeness(df: pd.DataFrame) -> dict:
    core_cols = ["o", "c", "h", "l", "v"]
    total = len(df) * len(core_cols)
    missing = df[core_cols].isna().sum().sum()
    score = round((1 - missing / total) * 100, 2) if total else 0.0
    return {"score": score, "missing_cells": int(missing), "total_cells": int(total)}


def score_timeliness(df: pd.DataFrame) -> dict:
    if df.empty:
        return {"score": 0.0, "latest_date": None, "days_behind": None}
    latest = df["t"].max()
    today = datetime.now()
    latest = min(latest, today)
    days_behind = (today - latest).days
    score = max(0.0, 100 - days_behind * 20)
    return {"score": round(score, 2), "latest_date": latest.strftime("%Y-%m-%d"), "days_behind": days_behind}


def score_consistency(df: pd.DataFrame) -> dict:
    checks = {
        "high_ge_low": (df["h"] >= df["l"]).sum(),
        "close_in_range": df["c"].between(df["l"], df["h"]).sum(),
        "open_in_range": df["o"].between(df["l"], df["h"]).sum(),
        "volume_positive": (df["v"] > 0).sum(),
    }
    total = len(df)
    passed = sum(checks.values())
    score = round(passed / (total * len(checks)) * 100, 2) if total else 0.0
    return {"score": score, "checks": {k: int(v) for k, v in checks.items()}, "total_rows": int(total)}


def score_validity(df: pd.DataFrame) -> dict:
    df = df.copy()
    df["ret"] = df.groupby("code")["c"].pct_change() * 100
    valid = df["ret"].between(-20, 20).sum()
    total = df["ret"].notna().sum()
    score = round(valid / total * 100, 2) if total else 0.0
    outliers = df[(df["ret"].notna()) & (~df["ret"].between(-20, 20))]
    return {
        "score": score,
        "valid_rows": int(valid),
        "total_rows": int(total),
        "outlier_count": len(outliers),
        "outlier_examples": outliers[["code", "t", "ret"]].head(3).to_dict("records"),
    }


def score_coverage(codes: list, df: pd.DataFrame) -> dict:
    fetched = df["code"].unique().tolist()
    missing = [c for c in codes if c not in fetched]
    score = round((len(codes) - len(missing)) / len(codes) * 100, 2) if codes else 0.0
    return {"score": score, "expected": len(codes), "fetched": len(fetched), "missing": missing}


def evaluate_alerts(scores: dict, thresholds: dict) -> list:
    alerts = []
    for dim, thr in thresholds.items():
        s = scores.get(dim, {}).get("score", 0)
        if s < thr:
            alerts.append(f"[ALERT] {dim} 评分 {s} 低于阈值 {thr}")
    return alerts


def main():
    frames = []
    for code in CODES:
        df = fetch_daily(code, lt=60)
        if not df.empty:
            frames.append(df)
            print(f"[pull] {code}: {len(df)} 行, latest={df['t'].max().strftime('%Y-%m-%d')}")

    df_all = pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()

    scores = {
        "completeness": score_completeness(df_all),
        "timeliness": score_timeliness(df_all),
        "consistency": score_consistency(df_all),
        "validity": score_validity(df_all),
        "coverage": score_coverage(CODES, df_all),
    }

    thresholds = {
        "completeness": 99.0,
        "timeliness": 80.0,
        "consistency": 99.0,
        "validity": 95.0,
        "coverage": 100.0,
    }

    alerts = evaluate_alerts(scores, thresholds)
    overall = round(sum(s["score"] for s in scores.values()) / len(scores), 2)

    report = {
        "generated_at": datetime.now().isoformat(),
        "codes": CODES,
        "total_rows": int(len(df_all)),
        "overall_score": overall,
        "scores": scores,
        "thresholds": thresholds,
        "alerts": alerts,
        "status": "PASS" if not alerts else "WARN",
    }

    REPORT_PATH.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"\n[report] overall={overall}, status={report['status']}")
    for dim, s in scores.items():
        print(f"  {dim:<15}: {s['score']}")
    if alerts:
        print("\n[alerts]")
        for a in alerts:
            print(f"  {a}")


if __name__ == "__main__":
    main()
```

跑完你会得到类似这样的输出：

```
[pull] 600519: 50 行, latest=2025-12-01
[pull] 000001: 50 行, latest=2025-12-01
...

[report] overall=80.0, status=WARN
  completeness   : 100.0
  timeliness     : 0.0
  consistency    : 100.0
  validity       : 100.0
  coverage       : 100.0

[alerts]
  [ALERT] timeliness 评分 0.0 低于阈值 80.0
```

如果你用的是演示证书，数据最新日期固定在某一天，`timeliness` 会很低，这是预期现象——换成正式证书后，它会真实反映你的数据延迟。

## 怎么把评分卡挂到流水线里

在上一篇的 6 阶段流水线中，评分卡就是第 6 阶段 `monitor` 的升级：

```
pull -> store -> clean -> merge -> feat -> monitor(评分卡)
```

推荐两个落地规则：

1. **评分卡不通过，下游不执行**：只要出现 `[ALERT]`，特征工程/回测/模型训练都停止，避免脏数据污染结果。
2. **分数要持久化**：把 `quality_report.json` 写到日志目录，长期追踪，方便发现"分数缓慢下降"这类慢性问题。

## 常见坑

- **及时性阈值别太严**：A股不是每天都有数据，周末/节假日没有 bar 是正常的。建议用"最新数据距离今天超过 3 个交易日"才算异常。
- **有效性要分板块**：科创板/创业板涨跌停是 20%，主板是 10%，全市场统一用 +/-20% 会漏掉主板的真实涨停，但不容易误报；反过来用 +/-10% 会误杀科创板。
- **覆盖度不能只看行数**：有些接口会返回空列表表示"成功但无数据"，此时要检查 `code` 是否真的在 DataFrame 里。

## 小结

数据质量评分卡的本质是**把"数据有没有问题"从玄学变成工程指标**。5 个维度覆盖了量化数据最常见的问题类型，100 行出头的代码就能每天自动巡检一次。

记住三件事：

1. **不要只看 NaN**：价格逻辑、最新日期、覆盖度同样重要。
2. **阈值要可配置**：不同策略、不同数据源的标准不一样。
3. **告警必须可行动**：每个 `[ALERT]` 都要能定位到具体阶段或具体股票。

**下篇预告**：评分卡让我们知道数据好不好，但研究阶段的 Notebook 怎么变成能稳定上线的工程？#10 给你一份**从研究到上线的最小闭环方案**。

---

## 免责声明

本文仅供技术学习交流，不构成任何投资建议。市场有风险，决策需谨慎。

---

## 了解更多

完整示例代码与文档：<https://github.com/MaiRuiApi>


数据问题不会被代码报错暴露，它只会安静地让你的策略失效。与其事后救火，不如给数据流水线配一张质量评分卡——每天跑完自动打分，哪里掉链子一眼看见。

【Python量化数据工程实战 #09】数据 Pipeline 跑了一个月才发现缺数？用质量评分卡 5 分钟定位问题

编程语言

大数据

运维

量化数据质量常被忽视却致命：回测优异实盘崩坏、因子NaN、股票漏拉，根源多是数据缺陷。本文提供5维评分卡模型（完整性、及时性、一致性、有效性、覆盖度），附可运行Python代码，自动评分并告警，支持阈值触发[ALERT]，可挂载流水线监控阶段，防脏数据入模型，助你实现客观、可量化的数据巡检。

代码解释

数据流

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【Python量化数据工程实战 #09】数据 Pipeline 跑了一个月才发现缺数？用质量评分卡 5 分钟定位问题-腾讯云开发者社区-腾讯云

【Python量化数据工程实战 #09】数据 Pipeline 跑了一个月才发现缺数？用质量评分卡 5 分钟定位问题

【Python量化数据工程实战 #09】数据 Pipeline 跑了一个月才发现缺数？用质量评分卡 5 分钟定位问题

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐