📦 Quant Data Quality — 量化数据质量检查

版本：1.0.0
适用项目：量化策略项目

🎯 检查目标

确保数据完整性、准确性、一致性、时效性

📋 数据质量检查清单

1. 数据完整性

1.1 价格数据完整性

检查项：

• [ ] 数据量是否充足？（股票数 × 交易日数）
• [ ] 是否有缺失日期？
• [ ] 是否有缺失股票？

检查方法：

import pandas as pd
# 加载价格数据
price = pd.read_parquet('data/integrated/price_integrated.parquet')
# 检查数据量
print(f"记录数: {len(price)}")
print(f"股票数: {price['code'].nunique()}")
print(f"日期范围: {price['date'].min()} ~ {price['date'].max()}")
# 检查缺失日期
dates = pd.to_datetime(price['date'].unique())
all_dates = pd.date_range(start=dates.min(), end=dates.max(), freq='B')  # 工作日
missing_dates = set(all_dates) - set(dates)
print(f"缺失日期: {len(missing_dates)}")

1.2 因子数据完整性

检查项：

• [ ] 因子数量是否合理？
• [ ] 是否有大量nan值？
• [ ] 是否有极端值？

检查方法：

import pandas as pd
import numpy as np
# 加载因子数据
factors = pd.read_parquet('data/factors_v2/alphagbm_rolling_factors_active.parquet')
# 检查nan值
nan_count = factors.isna().sum().sum()
total_count = factors.size
nan_ratio = nan_count / total_count
print(f"nan值数量: {nan_count:,}")
print(f"nan值比例: {nan_ratio:.2%}")
# 检查极端值
for col in factors.select_dtypes(include=[np.number]).columns:
    q1 = factors[col].quantile(0.01)
    q99 = factors[col].quantile(0.99)
    extreme_count = ((factors[col] < q1) | (factors[col] > q99)).sum()
    print(f"{col}: 极端值数量 {extreme_count}")

1.3 信号数据完整性

检查项：

• [ ] 信号数量是否合理？
• [ ] 信号日期是否最新？
• [ ] 信号股票是否在股票池内？

2. 数据准确性

2.1 价格数据准确性

检查项：

• [ ] 是否有零价格？
• [ ] 是否有负价格？
• [ ] 是否有异常收益率（>20%）？

检查方法：

# 检查零价格
zero_price = price[price['close'] == 0]
print(f"零价格记录: {len(zero_price)}")
# 检查负价格
neg_price = price[price['close'] < 0]
print(f"负价格记录: {len(neg_price)}")
# 检查异常收益率
price['return'] = price.groupby('code')['close'].pct_change()
abnormal_return = price[abs(price['return']) > 0.2]
print(f"异常收益率记录: {len(abnormal_return)}")

2.2 因子数据准确性

检查项：

• [ ] 是否有inf值？
• [ ] 是否有极端值（>1e10）？
• [ ] 因子分布是否合理？

检查方法：

import numpy as np
# 检查inf值
inf_count = np.isinf(factors.select_dtypes(include=[np.number])).sum().sum()
print(f"inf值数量: {inf_count}")
# 检查极端值
extreme_count = (abs(factors.select_dtypes(include=[np.number])) > 1e10).sum().sum()
print(f"极端值数量: {extreme_count}")

3. 数据一致性

3.1 价格-成交量一致性

检查项：

• [ ] 零成交量比例是否合理？
• [ ] 价格-成交量时间是否对齐？

检查方法：

# 检查零成交量
zero_volume = price[price['volume'] == 0]
print(f"零成交量记录: {len(zero_volume)}")
print(f"零成交量比例: {len(zero_volume) / len(price):.2%}")

3.2 跨源数据一致性

检查项：

• [ ] 不同数据源的价格是否一致？
• [ ] 不同数据源的日期范围是否一致？

4. 数据时效性

4.1 价格数据时效性

检查项：

• [ ] 最新数据日期？
• [ ] 滞后天数？

检查方法：

from datetime import datetime
latest_date = pd.to_datetime(price['date'].max())
today = datetime.now()
lag = (today - latest_date).days
print(f"最新日期: {latest_date}")
print(f"滞后天数: {lag}")

4.2 因子数据时效性

检查项：

• [ ] 因子更新频率？
• [ ] 因子滞后天数？

🔍 常见数据问题

问题1：价格数据缺失

表现：

• 某些日期没有数据
• 某些股票没有数据

原因：

• 数据源问题
• 爬虫失败
• 停牌

解决方案：

• ✅ 使用多数据源
• ✅ 定期检查数据完整性
• ✅ 建立数据更新告警

问题2：因子数据大量nan

表现：

• 因子列有很多nan值

原因：

• 计算窗口不足
• 数据源缺失
• 计算逻辑错误

解决方案：

• ✅ 检查计算窗口
• ✅ 填充缺失值
• ✅ 修复计算逻辑

问题3：因子极端值

表现：

• 因子值异常大或异常小

原因：

• 除零错误
• 计算逻辑错误
• 数据质量问题

解决方案：

• ✅ Winsorize处理
• ✅ 检查计算逻辑
• ✅ 过滤极端值

问题4：信号过期

表现：

• 信号日期滞后很久

原因：

• 信号生成任务未运行
• 数据更新任务未运行

解决方案：

• ✅ 检查cron任务
• ✅ 建立更新告警
• ✅ 手动触发更新

📊 数据质量报告模板

# 数据质量报告
检查日期：YYYY-MM-DD
一、数据概览
数据类型
记录数
股票数
日期范围
滞后天数
价格数据
1.16M
525
2015-2026
1
因子数据
991K
525
2020-2026
17
信号数据
10
10
2026-03-27
0
二、数据质量评分
维度
评分
说明
完整性
100/100
数据完整
准确性
100/100
无异常值
一致性
100/100
跨源一致
时效性
70/100
因子滞后17天
总评分：92.5/100
三、问题清单
问题
严重程度
建议
因子数据滞后
中
下周一更新
检查人：OpenClaw Assistant
日期：YYYY-MM-DD

🚫 禁止事项

• ❌ 使用未经检查的数据
• ❌ 忽略数据质量问题
• ❌ 没有数据质量报告

技能版本：1.0.0