📚 课程目录第3章数据源与数据清洗

3. 数据源与数据清洗

因子库的底层质量取决于原始数据的准确性与一致性。本章聚焦于国内量化研究中最常用的四大数据源，并系统讲解数据清洗的标准流程，包括去重、缺失值填充、异常值处理以及复权价格处理。掌握这些内容后，你能够构建出干净、可靠、可直接用于因子计算的本地数据集。

3.1 常见数据源对比

国内量化研究的主流数据源包括 Wind（万得）、Tushare、聚宽（JoinQuant）和 CSMAR（国泰安）。它们在数据覆盖、更新频率、获取方式和成本上各有差异。下表从关键维度进行对比：

数据源	覆盖范围	更新频率	获取方式	成本	适用场景
Wind	A股、港股、期货、债券、基金、宏观	实时/日频/周频	API（WindPy）、Excel插件、终端	高（机构年费）	机构级研究、实时策略、全资产覆盖
Tushare	A股、基金、期货、指数、宏观	日频/分钟频	HTTP API（Python SDK）	免费/积分制	个人研究、回测、因子探索
聚宽	A股、基金、期货、指数	日频/分钟频	本地API、平台内数据接口	免费/付费会员	策略回测、因子平台、社区协作
CSMAR	A股、港股、财务、公司治理、市场微观结构	日频/周频/月频	数据库直连、CSV导出	高（学术/机构订阅）	学术研究、精细化因子分析、财务数据深度清洗

选型建议： 个人研究者优先使用 Tushare 或聚宽免费接口；机构团队建议 Wind + CSMAR 组合，兼顾实时行情与深度财务数据。无论选择哪种数据源，都应在本地建立统一的数据存储格式（如 Parquet 或 HDF5），以便后续清洗与因子计算。

3.2 数据清洗流程

原始数据通常包含重复记录、缺失值、异常值以及价格未复权等问题。标准清洗流程分为四个步骤：去重 → 缺失值处理 → 异常值处理 → 复权价格处理。以下 SVG 图展示了完整的清洗流水线：

3.2.1 去重

重复数据主要来源于数据源接口重试、多源合并或手动追加。去重操作应基于 股票代码 + 交易日期 联合主键进行。以下 Python 代码展示了基于 pandas 的去重逻辑：

import pandas as pd

# 假设 df 包含 columns: ['stock_code', 'trade_date', 'open', 'high', 'low', 'close', 'volume']
# 去重：保留第一条记录，删除后续重复
df = df.drop_duplicates(subset=['stock_code', 'trade_date'], keep='first')

# 检查去重后索引是否唯一
assert df.set_index(['stock_code', 'trade_date']).index.is_unique, "去重后仍存在重复索引"

注意： 部分数据源（如 Wind 的某些接口）可能在除权除息日返回多条记录，此时应结合复权因子判断哪条是有效数据，而非简单去重。

3.2.2 缺失值处理

缺失值在因子数据中常见于停牌日、新股上市初期或财务数据未更新。处理策略取决于缺失原因和后续因子计算需求：

前向填充（ffill）： 适用于价格、成交量等连续型数据，用最近一个非缺失值填充。例如停牌日的价格沿用前一交易日数据。
后向填充（bfill）： 适用于数据开头缺失，用下一个非缺失值填充。
插值法： 适用于少量缺失且数据趋势明显，使用线性插值或时间加权插值。
删除： 当缺失比例过高（如超过 50%）或缺失集中在关键因子计算窗口时，直接删除该股票或该时间点。

# 前向填充停牌日价格
df['close'] = df.groupby('stock_code')['close'].ffill()

# 线性插值填充少量缺失
df['volume'] = df.groupby('stock_code')['volume'].apply(lambda x: x.interpolate(method='linear'))

# 删除缺失比例超过 40% 的股票
missing_ratio = df.groupby('stock_code')['close'].apply(lambda x: x.isnull().mean())
stocks_to_drop = missing_ratio[missing_ratio > 0.4].index
df = df[~df['stock_code'].isin(stocks_to_drop)]

3.2.3 异常值处理

异常值可能由数据录入错误、极端市场事件或数据源 bug 引起。常用检测方法包括：

方法	原理	适用场景	注意事项
MAD（中位数绝对偏差）	基于中位数的稳健统计量，不受极端值影响	收益率、换手率等分布不对称的数据	阈值通常设为 3~5 倍 MAD
Z-score	基于均值和标准差，假设数据近似正态	市值、成交额等近似正态分布的数据	阈值通常设为 3~4
分位数截断	直接截断上下 0.5%~1% 的极端值	因子值预处理，防止极端值影响分层	需结合业务理解，避免过度截断

import numpy as np

# MAD 方法检测异常值
def detect_outliers_mad(series, threshold=5):
    median = series.median()
    mad = np.median(np.abs(series - median))
    modified_z_scores = 0.6745 * (series - median) / mad
    return np.abs(modified_z_scores) > threshold

# 应用：将异常值替换为阈值上限/下限
df['return'] = df.groupby('trade_date')['return'].transform(
    lambda x: x.clip(lower=x.quantile(0.005), upper=x.quantile(0.995))
)

3.3 复权价格处理

复权价格是因子计算的基础。未复权价格在除权除息日会出现跳空，导致收益率计算失真。复权分为前复权和后复权两种方式：

前复权： 调整历史价格，使当前价格保持不变。适用于回测，因为回测起点价格被调整，能真实反映从历史持有到现在的收益。
后复权： 调整当前价格，使历史价格保持不变。适用于分析历史价格走势，但当前价格被调整，不便于实盘参考。

以下 SVG 图展示了复权前后价格曲线的对比效果：

复权价格的计算依赖于复权因子。复权因子通常由数据源提供（如 Wind 的 adj_factor 字段），也可根据除权除息信息自行计算。以下代码展示了如何使用复权因子进行前复权：

# 假设 df 包含 columns: ['stock_code', 'trade_date', 'close', 'adj_factor']
# adj_factor 为当日复权因子，通常数据源提供

# 计算前复权价格：复权价格 = 原始价格 * (最新复权因子 / 当日复权因子)
latest_adj = df.groupby('stock_code')['adj_factor'].transform('last')
df['close_adj'] = df['close'] * (latest_adj / df['adj_factor'])

# 对 open, high, low 同样处理
for col in ['open', 'high', 'low']:
    df[f'{col}_adj'] = df[col] * (latest_adj / df['adj_factor'])

# 验证：复权后收益率应连续
df['return_adj'] = df.groupby('stock_code')['close_adj'].pct_change()
print(df[['stock_code', 'trade_date', 'close', 'close_adj', 'return_adj']].head())

  关键检查点： 复权完成后，应检查复权收益率序列是否出现异常跳空。如果某日收益率绝对值超过 20%（A股涨跌停限制），则可能存在复权因子错误或数据源问题，需要回溯验证。

3.4 清洗后数据质量检查

完成上述清洗步骤后，建议执行以下质量检查清单：

完整性检查： 每个股票-日期组合是否唯一，无重复索引。
缺失率检查： 关键字段（如复权收盘价）缺失率是否低于 5%。
范围检查： 价格、成交量、收益率是否在合理业务范围内（如价格不为负，收益率在 -20%~20% 之间）。
复权一致性检查： 随机抽取 10 只股票，手动计算除权除息日前后的复权收益率，验证与数据源提供的复权因子是否一致。
时间序列连续性： 检查是否存在交易日缺失（如因停牌导致连续缺失超过 20 个交易日，需标记处理）。

通过以上流程，你可以获得一套高质量、可直接用于因子计算的本地数据集。下一阶段将基于清洗后的数据，进行因子计算与分层试验。