第四章：数据获取与清洗——从原始数据到可用特征
做量化投资，有一句老话叫「垃圾进，垃圾出」。你模型再牛，数据不对，结果就是废的。今天咱们聊聊数据获取与清洗，这是整个基本面量化流程的起点，也是我踩坑最多的地方。
核心逻辑：原始数据 → 获取 → 清洗（缺失值/异常值） → 标准化 → 可用特征矩阵
4.1 数据获取：Tushare vs Akshare
我个人习惯用 Tushare 做专业回测，用 Akshare 做快速验证。为什么？Tushare 的数据质量高，但需要 token 和积分；Akshare 免费开源，但偶尔会有字段缺失。
先看 Tushare 怎么用。你需要在官网注册，拿到 token。然后安装：
pip install tushare获取股票日线数据，我一般这么写：
import tushare as ts
# 设置 token（建议写在配置文件里，别硬编码）
ts.set_token('你的token')
pro = ts.pro_api()
# 获取平安银行 2023年全年日线数据
df = pro.daily(
ts_code='000001.SZ',
start_date='20230101',
end_date='20231231'
)
print(df.head())小技巧：我在项目中遇到过 Tushare 接口返回空数据的情况，后来发现是日期格式问题。Tushare 要求日期格式为 YYYYMMDD，别写成 2023-01-01。
Akshare 就更直接了，不需要 token：
import akshare as ak
# 获取 A 股个股实时行情
df = ak.stock_zh_a_spot()
print(df.head())嗯，这里要注意：Akshare 的字段名是中文的，比如「最新价」「涨跌幅」。你如果习惯英文列名，记得做一次 rename。
4.2 数据清洗：缺失值处理
拿到数据后，第一件事不是建模，而是检查数据质量。你想想看，如果某只股票停牌了，那天的数据就是 NaN。直接扔进模型，会出大问题。
我常用的缺失值处理方法有三种：
| 方法 | 适用场景 | 代码示例 |
|---|---|---|
| 删除法 | 缺失比例 < 5% | df.dropna() |
| 前向填充 | 时间序列数据（停牌） | df.fillna(method='ffill') |
| 插值法 | 连续型财务指标 | df.interpolate() |
举个例子，处理停牌数据：
# 前向填充：用上一个交易日的数据填充停牌日
df['close'].fillna(method='ffill', inplace=True)
# 如果停牌超过 5 天，我建议直接删除该时间段
# 因为长时间停牌后的复牌，价格往往有跳空
df['close'] = df['close'].fillna(method='ffill', limit=5)
df.dropna(subset=['close'], inplace=True)避坑指南：我曾经在回测中用了前向填充处理所有缺失值，结果发现某只股票停牌了 3 个月，复牌后直接跌停。前向填充把停牌前的价格延续了 3 个月，导致回测收益虚高。所以，停牌超过 5 天的，我建议直接剔除。
4.3 数据标准化：让不同量纲的数据可比较
基本面数据里，有的指标是百分比（ROE），有的是绝对值（营收），还有的是对数（市值）。直接放一起建模，量纲差异会主导模型。说白了，就是「大数欺负小数」。
我常用的标准化方法：
- Z-score 标准化：适用于数据近似正态分布，比如市盈率、市净率
- Min-Max 标准化：适用于有明确上下界的指标，比如资产负债率（0~1）
- Rank 标准化：适用于极端值较多的数据，比如小市值股票的换手率
代码实现：
from sklearn.preprocessing import StandardScaler, MinMaxScaler
import pandas as pd
import numpy as np
# 假设 df 包含多只股票的财务指标
# 列：['roe', 'debt_ratio', 'market_cap']
# Z-score 标准化
scaler = StandardScaler()
df_zscore = pd.DataFrame(
scaler.fit_transform(df[['roe', 'debt_ratio', 'market_cap']]),
columns=['roe_z', 'debt_ratio_z', 'market_cap_z'],
index=df.index
)
# Min-Max 标准化
scaler_mm = MinMaxScaler()
df_mm = pd.DataFrame(
scaler_mm.fit_transform(df[['roe', 'debt_ratio', 'market_cap']]),
columns=['roe_mm', 'debt_ratio_mm', 'market_cap_mm'],
index=df.index
)个人经验：我在做多因子模型时，发现 Z-score 对极端值敏感。有一次 ROE 数据里有个异常值（某公司一次性资产处置导致 ROE 飙到 200%），Z-score 后其他股票都被压缩到 -0.5~0.5 之间。后来我改用 Rank 标准化，把每个指标按百分位排名，效果好了很多。
4.4 实战：完整的数据清洗流水线
把上面三步串起来，就是一个完整的流水线。我一般封装成一个函数：
def clean_pipeline(df, method='zscore'):
"""
数据清洗流水线
:param df: 原始 DataFrame，包含日期、股票代码、指标列
:param method: 标准化方法，可选 'zscore', 'minmax', 'rank'
:return: 清洗后的 DataFrame
"""
# 1. 删除全为 NaN 的列
df = df.dropna(axis=1, how='all')
# 2. 前向填充停牌数据（最多 5 天）
df = df.fillna(method='ffill', limit=5)
# 3. 删除仍有缺失的行
df = df.dropna()
# 4. 标准化
if method == 'zscore':
scaler = StandardScaler()
scaled = scaler.fit_transform(df.select_dtypes(include=[np.number]))
elif method == 'minmax':
scaler = MinMaxScaler()
scaled = scaler.fit_transform(df.select_dtypes(include=[np.number]))
elif method == 'rank':
scaled = df.select_dtypes(include=[np.number]).rank(pct=True)
df_scaled = pd.DataFrame(
scaled,
columns=df.select_dtypes(include=[np.number]).columns,
index=df.index
)
return df_scaled核心要点：数据清洗的顺序很重要。先处理缺失值，再做标准化。如果先标准化再填充，填充值会受标准化后的分布影响，导致偏差。
好了，这一章的内容就这些。数据获取与清洗是基本功，但也是最容易出 bug 的地方。我建议你拿到任何新数据源，先花 10 分钟做一次 EDA（探索性数据分析），看看缺失值分布、异常值、量纲差异。磨刀不误砍柴工。