主题切换
3.2 数据获取与清洗
概念详解
数据：量化策略的血液
在量化交易中有一句格言：Garbage In, Garbage Out。无论模型多么精妙，如果输入的数据是脏的，输出就一定是错的。更危险的是，脏数据导致的错误往往是"看起来合理但实际上错误"的，极难被发现。
数据获取与清洗通常占据量化研究者50-70%的工作时间。这不是效率问题，而是数据质量的极端重要性使然。
名词解释：复权
调整股价以反映分红、送股、拆股等事件，保持价格序列连续可比。前复权调整历史价格，后复权调整当前价格。
名词解释：除权除息 (Ex-Right/Ex-Dividend)
股票分红送股后，交易所对股价进行的调整。如果不做复权处理，价格序列会出现"跳跃"，看似价格暴跌实际是除权除息所致。
名词解释：前视偏差 (Look-Ahead Bias)
在回测中使用了当时不可获得的信息（如用年报数据在年报发布日之前交易）。这是量化回测中最隐蔽的错误之一。
金融数据的独特挑战
金融数据与一般的数据集有本质的不同：
- 时序依赖性：数据点不是独立的，昨天的价格影响今天的决策
- 生存偏差：回测往往只使用"现在还活着的股票"，忽略了已退市的
- 数据时效性：财报数据存在发布延迟，不能假设在报告期结束日就知道数据
- 复权复杂性：分红、送股、配股、拆股、并股都会在价格序列上留下"疤痕"
- 节假日与停牌：不同市场、不同股票的交易日历不同，处理不当会导致对齐错误
- 极端值和错误数据：数据库中的错误（如价格多一个零）比想象中常见得多
主要数据源概览
| 数据源 | 覆盖范围 | 数据类型 | 费用 | 适用场景 |
|---|---|---|---|---|
| AkShare | A股、期货、基金 | 日线、分钟线、基本面 | 免费 | 个人研究、入门学习 |
| Tushare Pro | A股、指数、宏观 | 全品类 | 免费/付费 | 个人和中小团队 |
| Wind | A股、港股、全球 | 全品类专业级 | 高额付费 | 专业机构 |
| yfinance | 全球股票、ETF | 日线 | 免费 | 海外市场研究 |
| Baostock | A股 | 日线、分钟线 | 免费 | 个人研究 |
| JoinQuant/RiceQuant | A股 | 日线+基本面 | 免费/付费 | 在线回测+数据一体 |
Python实战
案例1：清洗日线数据
此处有展示代码展开 ▼
python
df['Close'] = df['Close'].ffill()
df['Return'] = df['Close'].pct_change()
abnormal = df[df['Return'].abs() > 0.2]
print(f"异常天数: {len(abnormal)}")点击展开可浏览运行结果
缺失值（已前向填充）: 1 处
异常收益天数: 2
异常日期:
Close Return
2024-03-21 120.278364 0.500000
2024-03-22 80.460457 -0.331048案例2：从AkShare获取A股数据并完整清洗
此处有展示代码展开 ▼
python
import pandas as pd
import numpy as np
import akshare as ak
def fetch_and_clean_a_stock(symbol, start_date, end_date, adjust='qfq'):
"""
获取并清洗A股日线数据
symbol: 股票代码，如 '600519'
start_date: 起始日期 '20200101'
end_date: 截止日期 '20241231'
adjust: 复权方式 'qfq'(前复权)/'hfq'(后复权)/''(不复权)
"""
# 获取原始数据
raw = ak.stock_zh_a_hist(
symbol=symbol,
period='daily',
start_date=start_date,
end_date=end_date,
adjust=adjust
)
# 标准化列名
raw = raw.rename(columns={
'日期': 'date', '开盘': 'open', '收盘': 'close',
'最高': 'high', '最低': 'low', '成交量': 'volume',
'成交额': 'amount', '振幅': 'amplitude',
'涨跌幅': 'pct_change', '涨跌额': 'change',
'换手率': 'turnover'
})
# 设置日期索引并排序
raw['date'] = pd.to_datetime(raw['date'])
raw = raw.set_index('date').sort_index()
# === 数据清洗流水线 ===
df = raw.copy()
n_original = len(df)
# 1. 处理重复日期（取最后一条记录）
df = df[~df.index.duplicated(keep='last')]
# 2. 检测并标记极端价格错误
# 价格突然翻倍或腰斩（在没有涨跌停限制的情况下）
price_change = df['close'].pct_change()
extreme_mask = price_change.abs() > 0.15 # 单日涨跌超15%可能是数据错误
if extreme_mask.any():
print(f"检测到 {extreme_mask.sum()} 个极端价格变动日:")
for date in df.index[extreme_mask]:
print(f" {date.date()}: 变动={price_change[date]:.2%}")
# 3. 处理零成交量日（停牌日）
zero_volume = df['volume'] <= 0
df.loc[zero_volume, ['open', 'high', 'low']] = df.loc[zero_volume, 'close'].values
# 4. 前向填充（停牌延续最新价）
# 注意：不建议在金融数据上使用插值，因为价格在停牌期间并没有"中间值"
df['close'] = df['close'].ffill()
df['open'] = df['open'].ffill()
df['high'] = df['high'].ffill()
df['low'] = df['low'].ffill()
# 5. 衍生计算
df['return'] = df['close'].pct_change()
df['log_return'] = np.log(df['close'] / df['close'].shift(1))
df['range'] = (df['high'] - df['low']) / df['close'] # 日内振幅
# 6. 标记异常收益率日（可能是脏数据或极端事件）
ret_mean = df['return'].mean()
ret_std = df['return'].std()
df['is_abnormal'] = (df['return'] - ret_mean).abs() > 4 * ret_std
# 7. 计算可用交易日数
trading_days = (~zero_volume).sum()
# === 数据质量报告 ===
print("=" * 50)
print(f"数据清洗报告: {symbol}")
print("=" * 50)
print(f"原始记录数: {n_original}")
print(f"清洗后记录数: {len(df)}")
print(f"重复日期: {n_original - len(df)}")
print(f"零成交量日(停牌): {zero_volume.sum()} ({zero_volume.sum()/len(df):.1%})")
print(f"异常收益率日: {df['is_abnormal'].sum()} ({df['is_abnormal'].sum()/len(df):.1%})")
print(f"日期范围: {df.index.min().date()} ~ {df.index.max().date()}")
print(f"实际交易日: {trading_days}")
print(f"\n价格统计:")
print(f" 开盘: {df['open'].min():.2f} ~ {df['open'].max():.2f}")
print(f" 收盘: {df['close'].min():.2f} ~ {df['close'].max():.2f}")
print(f" 日均成交量: {df.loc[~zero_volume, 'volume'].mean():,.0f}")
return df点击展开可浏览运行结果
📘 本段代码定义了 1 个函数/类：函数 `fetch_and_clean_a_stock`（获取并清洗A股日线数据）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
案例3：多股票面板数据构建与前视偏差防护
此处有展示代码展开 ▼
python
import pandas as pd
import numpy as np
def build_panel_with_publication_dates(financial_data, report_dates,
price_data, lookback=252):
"""
构建避免前视偏差的面板数据集
关键思想：财务数据的实际发布日期（report_dates）通常晚于报告期结束日期。
回测时必须使用report_dates而非报告期日期来对齐。
financial_data: DataFrame, index=stock_code, columns=财务指标
report_dates: DataFrame, index=stock_code, values=发布日期
price_data: DataFrame, index=date, columns=stock_code
"""
aligned_data = {}
for stock in financial_data.index:
report_date = pd.Timestamp(report_dates.loc[stock, 'report_date'])
financials = financial_data.loc[stock]
# 只在发布日期之后才能使用该财务数据
stock_prices = price_data[stock]
valid_prices = stock_prices[stock_prices.index >= report_date]
if len(valid_prices) > 0:
for col in financials.index:
aligned_data[(stock, col)] = financials[col]
return aligned_data
class DataQualityMonitor:
"""
持续监控数据质量的工具类，适用于实时或批量数据流
"""
def __init__(self, price_bounds=None, volume_bounds=None):
self.price_bounds = price_bounds or {}
self.volume_bounds = volume_bounds or {}
self.alerts = []
def check_price_continuity(self, prices, max_gap_pct=0.20):
"""
检查价格连续性：单日变动超过max_gap_pct可能为错误
"""
returns = prices.pct_change()
jumps = returns[returns.abs() > max_gap_pct]
if len(jumps) > 0:
for date, ret in jumps.items():
self.alerts.append({
'type': 'price_jump',
'date': date,
'value': ret,
'severity': 'HIGH' if abs(ret) > 0.50 else 'MEDIUM'
})
return jumps
def check_volume_anomaly(self, volumes, zscore_threshold=5):
"""
成交量异常检测：相对历史均值偏离超过阈值
"""
vol_mean = volumes.rolling(20).mean()
vol_std = volumes.rolling(20).std()
vol_zscore = (volumes - vol_mean) / vol_std
anomalies = vol_zscore[vol_zscore.abs() > zscore_threshold]
for date, z in anomalies.items():
self.alerts.append({
'type': 'volume_anomaly',
'date': date,
'zscore': z,
'severity': 'MEDIUM'
})
return anomalies
def check_ohlc_logic(self, ohlc_df):
"""
检查OHLC数据的逻辑一致性
- Low <= Open, Close, High
- High >= Open, Close, Low
- Low <= High
"""
issues = {}
# Low 不能高于 Open, Close, High
issues['low_above_open'] = ohlc_df['low'] > ohlc_df['open']
issues['low_above_close'] = ohlc_df['low'] > ohlc_df['close']
issues['low_above_high'] = ohlc_df['low'] > ohlc_df['high']
# High 不能低于 Open, Close, Low
issues['high_below_open'] = ohlc_df['high'] < ohlc_df['open']
issues['high_below_close'] = ohlc_df['high'] < ohlc_df['close']
total_issues = sum(v.sum() for v in issues.values())
if total_issues > 0:
for name, mask in issues.items():
if mask.any():
dates = ohlc_df.index[mask]
self.alerts.append({
'type': 'ohlc_logic',
'rule': name,
'count': len(dates),
'dates': dates[:5].tolist(), # 只展示前5个
'severity': 'HIGH'
})
return issues, total_issues
def generate_report(self):
"""生成数据质量报告"""
df_alerts = pd.DataFrame(self.alerts)
if len(df_alerts) == 0:
print("✅ 数据质量检查通过，未发现异常")
return
print("=" * 60)
print(f"数据质量报告 - {len(df_alerts)} 个警报")
print("=" * 60)
severity_counts = df_alerts['severity'].value_counts()
for severity in ['HIGH', 'MEDIUM', 'LOW']:
if severity in severity_counts:
print(f" {severity}: {severity_counts[severity]} 个")
print(f"\n按类型分组:")
type_counts = df_alerts['type'].value_counts()
for atype, count in type_counts.items():
print(f" {atype}: {count} 个")
# 高严重性警报详情
high_alerts = df_alerts[df_alerts['severity'] == 'HIGH']
if len(high_alerts) > 0:
print(f"\n=== 高严重性警报 ({len(high_alerts)}个) ===")
for _, alert in high_alerts.head(10).iterrows():
print(f" [{alert['severity']}] {alert['type']}: {alert.to_dict()}")
return df_alerts点击展开可浏览运行结果
📘 本段代码定义了 2 个函数/类：函数 `build_panel_with_publication_dates`（构建避免前视偏差的面板数据集）、类 `DataQualityMonitor`（持续监控数据质量的工具类，适用于实时或批量数据流）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
常见误区
误区一：复权数据可以无限信任
事实：复权计算理论上很简单，但实际数据供应商的复权数据常有错误。特别是连续多年的大量分红、送股后，累积的复权因子可能出现严重的精度问题。检查方法：比较前复权价格 × 复权因子 和 不复权价格 的一致性。
误区二：缺失值用均值填充即可
事实：金融数据的缺失机制很重要。是随机缺失（交易暂停）、结构性缺失（退市）、还是数据录入错误？不同的缺失机制需要不同的处理方式，简单均值填充可能在停牌期"发明"出虚假的交易机会。
误区三：数据清洗是一次性的
事实：数据质量问题会随着数据更新不断引入。建立自动化数据质量监控（如DataQualityMonitor类），在每次数据更新后自动检查，是专业量化团队的标准做法。
误区四：财务数据发布即是可知的
事实：A股年报的法定截止日是次年4月30日，但有些公司会提前发布，有些会拖到最后。如果回测中假设所有公司的年报都在12月31日可知（即报告期结束日），这种前视偏差会显著高估基本面策略的表现。必须使用实际发布日期来对齐数据。
实战练习
数据清洗实战：使用AkShare获取一只A股股票过去五年的日线数据，完成全流程数据清洗。特别检查：(a) 是否存在价格跳跃日，(b) 停牌天数占比，(c) 复权前后价格的一致性。
异常值检测：对沪深300所有成分股的某一年日收益率，使用箱线图法和Z-score法分别识别异常值，比较两种方法的检测结果差异，讨论哪种方法更适合金融数据。
前视偏差影响量化：选取一个基本面因子（如PE、PB或ROE），设计两组回测：(a) 使用报告期结束日对齐（存在前视偏差），(b) 使用发布日期对齐（无前视偏差），比较两组结果的差异，量化前视偏差对你的策略表现的影响。
延伸阅读
- 《Data Wrangling with Python》—— Jacqueline Kazil and Katharine Jarmul
- 《Python for Finance》—— Yves Hilpisch，第4-6章的金融数据处理
- 各数据供应商的数据字典：理解每个字段的准确定义和生成逻辑
- 《Tidy Data》—— Hadley Wickham，整洁数据的概念框架
- 《Bad Data Handbook》—— Q. Ethan McCallum，数据实际问题的案例集
本章要点
- 数据清洗占据量化研究时间的50-70%，是回报率最高的投入之一
- 金融数据的独特挑战：时序依赖、生存偏差、前视偏差、复权和停牌
- 复权、前视偏差和停牌处理是最容易被忽视但影响最大的三个数据问题
- 建立自动化的数据质量监控比每次手动检查更高效、更可靠
- 与脏数据相处是常态，追求的是"足够干净"而非"完全完美"