【Python量化数据工程实战 #09】数据 Pipeline 跑了一个月才发现缺数？用质量评分卡 5 分钟定位问题
你是不是也有过这种经历： - 回测结果看着很美，实盘一跑就崩，回头查数据发现**某段行情根本不在库里**。 - 某天因子值突然全是 NaN，查了半天才发现是**上游接口字段改名**。 - 多只股票一起跑，跑完才发现其中**有两只从头到尾就没拉到数据**。 数据问题不会被代码报错暴露，它只会安静地让你的策略失效。与其事后救火，不如给数据流水线配一张**质量评分卡**——每天跑完自动打分，哪里掉链子一眼看见。 ## 本文你将得到什么 1. **5 个维度的数据质量评分模型**：完整性、及时性、一致性、有效性、覆盖度。 2. **一份可直接复制运行的 Python 评分卡代码**，输入是你的 K 线 DataFrame，输出是评分报告 + 告警。 3. **阈值告警机制**：评分低于阈值自动打 `[ALERT]`，方便接入飞书/钉钉/企微机器人。 4. **把评分卡挂到流水线里的最佳实践**：放在 `monitor` 阶段，失败即停，不让脏数据进模型。 ## 为什么数据质量必须用"评分卡"而不是靠肉眼 很多开发者的数据检查是临时性的： ```python df.isna().sum() # 看一下有没有空值 df.tail() # 看一下最新日期 ``` 这在 3 只股票、80 天数据时还够用；一旦扩展到全市场、多数据源、长期运行，就会面临三个问题： - **维度太多**：空值只是冰山一角，还要关心价格逻辑、多表对齐、最新日期、覆盖度。 - **没有基线**：今天 "NaN 占比 2%" 是好是坏？没有历史分数对比，无法判断。 - **无法告警**：等肉眼发现问题，往往已经污染了下游特征和回测结果。 评分卡解决的是"把主观判断变成客观分数"。 ## 5 维评分模型设计 | 维度 | 含义 | 典型问题 | |------|------|----------| | 完整性 | 核心字段非空占比 | 接口超时导致某列全 NaN | | 及时性 | 最新数据距离今天多久 | 采集任务失败导致缺最近几天 | | 一致性 | 价格逻辑是否自洽 | high 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 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 给你一份**从研究到上线的最小闭环方案**。 --- ## 免责声明 本文仅供技术学习交流，不构成任何投资建议。市场有风险，决策需谨慎。 --- ## 了解更多 完整示例代码与文档：
