05-数据质量与校验规则
akshare 聚合抓取的数据存在缺失、异常、漂移等风险。数据写入前后需经多层校验，确保本地库数据可信。本规则定义校验维度、阈值与处理方式。
校验维度
数据质量按五个维度把控：
- 完整性：该有的数据是否有（行级、列级）
- 准确性：数值是否在合理范围
- 一致性：表与表之间、字段与字段之间是否矛盾
- 时效性：数据是否最新，是否有非预期延迟
- 交叉验证：同一数据点从多个独立来源获取，比对一致性
完整性校验
行级完整性
检查每个标的的数据是否覆盖预期时间范围：
-- 校验指数日线覆盖范围
SELECT
d.index_code,
d.index_name,
MIN(t.trade_date) AS first_date,
MAX(t.trade_date) AS last_date,
COUNT(*) AS row_count,
-- 交易日数估算（排除周末，粗略）
(julianday(MAX(t.trade_date)) - julianday(MIN(t.trade_date))) * 250 / 365 AS expected_rows,
CASE
WHEN COUNT(*) < (julianday(MAX(t.trade_date)) - julianday(MIN(t.trade_date))) * 200 / 365
THEN 'SUSPICIOUS'
ELSE 'OK'
END AS status
FROM dim_index d
LEFT JOIN index_daily t ON d.index_code = t.index_code
GROUP BY d.index_code
HAVING status = 'SUSPICIOUS';
阈值说明：
- 250/365 是全年交易日占比，200/365 是宽松下限（容忍 20% 缺失）
- 新上市标的可能不足一年，按实际交易日数校验
- 校验失败的标的写入
update_failure_log，标记 integrity_row
列级完整性
关键字段不允许缺失：
-- 检查指数日线关键字段缺失
SELECT index_code, COUNT(*) AS missing_close
FROM index_daily
WHERE close IS NULL
GROUP BY index_code
HAVING missing_close > 0;
交易日缺失检测
对比交易日历，找出未更新的日期：
-- 假设有 trade_calendar 表（可从 akshare 的 tool_trade_date_hist_sina 获取）
SELECT c.trade_date
FROM trade_calendar c
LEFT JOIN index_daily t
ON t.trade_date = c.trade_date
AND t.index_code = '000300'
WHERE c.is_open = 1
AND t.trade_date IS NULL
AND c.trade_date >= date('now', '-30 days');
交易日历表结构：
CREATE TABLE IF NOT EXISTS trade_calendar (
trade_date TEXT PRIMARY KEY,
is_open INTEGER NOT NULL, -- 1=交易日, 0=非交易日
exchange TEXT NOT NULL -- SSE/SZSE
);
准确性校验
数值范围校验
关键数值字段设定合理范围，超出范围标记异常：
-- 高低价逻辑校验
SELECT index_code, trade_date, open, close, high, low
FROM index_daily
WHERE high < GREATEST(open, close)
OR low > LEAST(open, close)
OR high < low;
跨表一致性
同一天同一标的的数据在不同表中应一致：
-- stock_daily 的 pe_ttm 与 stock_valuation 的 pe_ttm 应一致
SELECT s.stock_code, s.trade_date,
s.pe_ttm AS daily_pe,
v.pe_ttm AS val_pe,
ABS(s.pe_ttm - v.pe_ttm) AS diff
FROM stock_daily s
JOIN stock_valuation v
ON s.stock_code = v.stock_code
AND s.trade_date = v.trade_date
WHERE ABS(s.pe_ttm - v.pe_ttm) > 0.1;
允许小误差（浮点精度），差异超过 0.1 标记 cross_table_inconsistency。
时效性校验
数据新鲜度
检查最新数据日期是否在预期内：
SELECT 'index_daily' AS table_name, MAX(trade_date) AS latest_date,
julianday('now') - julianday(MAX(trade_date)) AS days_behind
FROM index_daily
UNION ALL
SELECT 'fund_daily', MAX(trade_date),
julianday('now') - julianday(MAX(trade_date))
FROM fund_daily
UNION ALL
SELECT 'stock_daily', MAX(trade_date),
julianday('now') - julianday(MAX(trade_date))
FROM stock_daily;
阈值：
- 日级数据：
days_behind > 3（跨周末）标记 stale_data
- 月级数据：
days_behind > 35 标记
- 季级数据：
days_behind > 95 标记
更新时间戳
每张表记录最后更新时间：
SELECT key, value AS last_updated
FROM db_meta
WHERE key LIKE '%_last_updated';
update_all.py 完成每张表更新后写入 db_meta：
def mark_table_updated(conn, table_name: str):
conn.execute("""
INSERT OR REPLACE INTO db_meta (key, value, updated_at)
VALUES (?, ?, datetime('now'))
""", (f"{table_name}_last_updated", datetime.now().isoformat()))
多数据源交叉验证
单一数据源存在系统性偏差或静默错误的风险（如上游接口数据源切换、复权方式不一致、口径调整等）。对关键数据点，从两个独立上游来源同时获取并比对，是发现这类问题的最有效手段。
验证范围
并非所有数据都需要交叉验证，按数据重要性分级：
验证策略
三种触发方式结合，平衡覆盖度与抓取成本：
- 全量交叉验证：首次建库或
--mode full 时，对所有历史数据做交叉验证。耗时较长但确保基线可信。
- 抽样交叉验证：日常
--mode daily 更新时，对新数据按比例抽样（如每 20 个交易日取 1 天，或最近 5 个交易日全量）。成本低，能发现持续性偏差。
- 关键时点强制验证：月末、季末、单日跌幅超 3% 的日期，强制交叉验证。这些时点数据准确性对分析结论影响最大。
容差设定
不同字段的容差按数据性质区分：
容差内的差异标记为 match，超容差标记为 mismatch，单源缺失标记为 missing。
验证结果存储
交叉验证结果写入 cross_validation_log 表（schema 见 02-数据架构与表结构设计），便于追踪历史趋势：
SELECT target_code, trade_date, field_name,
value_primary, value_secondary, diff_pct, status
FROM cross_validation_log
WHERE status = 'mismatch'
ORDER BY trade_date DESC
LIMIT 50;
数据质量标记联动
交叉验证结果影响主表的 data_quality 字段：
实现脚本
交叉验证逻辑在 fetcher_validation.py，由 update_all.py 在数据获取后调用：
# 简化示意：指数收盘价交叉验证
def validate_index_close(conn, run_id, index_code, symbol_em, trade_dates):
"""从东财和新浪分别获取收盘价，比对一致性。"""
df_em = fetch_with_retry(lambda: ak.stock_zh_index_daily_em(symbol=symbol_em))()
df_sina = fetch_with_retry(lambda: ak.stock_zh_index_daily(symbol=symbol_em))()
for date in trade_dates:
close_em = df_em.loc[df_em['date'] == date, 'close'].values
close_sina = df_sina.loc[df_sina['date'] == date, 'close'].values
if len(close_em) == 0 or len(close_sina) == 0:
status = 'missing'
elif abs(close_em[0] - close_sina[0]) <= 0.01:
status = 'match'
else:
status = 'mismatch'
# 标记主表数据为可疑
conn.execute(
"UPDATE index_daily SET data_quality='suspicious' "
"WHERE index_code=? AND trade_date=?",
(index_code, date)
)
# 写入验证日志
conn.execute(
"INSERT INTO cross_validation_log (...) VALUES (...)",
(run_id, index_code, date, 'close', 'em', 'sina',
close_em[0] if len(close_em) else None,
close_sina[0] if len(close_sina) else None, ...)
)
完整实现见 02-数据脚本/fetcher_validation.py。
校验执行
校验脚本
校验逻辑封装在 update_all.py 的 validate 模式：
python update_all.py --mode validate
输出校验报告到 logs/validation_YYYYMMDD.log，包含：
- 各表的行数、日期范围
- 异常记录数与详情
- 跨表不一致记录
- 时效性状态
校验结果表
校验结果写入 validation_result 表，便于查询历史趋势：
CREATE TABLE IF NOT EXISTS validation_result (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_id TEXT NOT NULL,
table_name TEXT NOT NULL,
check_type TEXT NOT NULL, -- integrity/accuracy/consistency/timeliness
check_name TEXT NOT NULL, -- 具体检查项
status TEXT NOT NULL, -- pass/warning/fail
detail TEXT, -- 详情（异常数量、样本）
checked_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_validation_run
ON validation_result(run_id);
阈值与告警
fail 级别需要立即处理，warning 级别记录但可继续。
异常处理流程
发现异常后的处理顺序：
- 记录：写入
validation_result 和 update_failure_log
- 隔离：异常数据不删除，但在查询接口中默认过滤（通过
is_valid 标记或视图）
- 排查：人工核查异常样本，判断是 akshare 接口问题还是真实数据
- 修复：
- 接口问题：切换备用接口或升级 akshare
- 真实异常：保留数据但标注
- 数据错误：用正确数据覆盖（需记录修改原因）
- 复盘：同类异常加入校验规则，避免下次遗漏
数据标记字段
为支持异常隔离，关键表增加 data_quality 字段：
-- 以 index_daily 为例，其他表同理
ALTER TABLE index_daily ADD COLUMN data_quality TEXT DEFAULT 'ok';
-- 取值：ok / suspicious / error / manual_fixed
查询接口默认只返回 data_quality IN ('ok', 'manual_fixed') 的数据，异常数据需显式查询：
-- 默认查询（只看正常数据）
SELECT * FROM index_daily
WHERE index_code = '000300'
AND data_quality IN ('ok', 'manual_fixed')
ORDER BY trade_date;
-- 排查时查看异常数据
SELECT * FROM index_daily
WHERE data_quality NOT IN ('ok', 'manual_fixed');
这样设计避免异常数据污染分析结论，同时保留数据用于排查。
定期全量校验
日常 validate 模式只校验近期数据（近 30 天），每月一次全量校验：
python update_all.py --mode validate --full
全量校验覆盖所有历史数据，耗时较长（可能数十分钟），安排在周末执行。全量校验的重点：
- 历史数据的完整性（是否有早期缺失未发现）
- 跨年数据的一致性（除权除息后复权数据是否正确）
- 长期未更新的标的（可能已退市）
