第2章数据采集与清洗

数据采集与清洗：从原始数据到可用因子

做量化投资，有个残酷的现实我得先告诉你——80%的时间都花在数据上，只有20%的时间真正在做策略。这不是夸张，是我这些年踩坑踩出来的经验。

你想想看，再牛的因子模型，喂进去的是垃圾数据，出来的只能是垃圾结果。所以这一章，咱们就聊聊数据采集与清洗的那些事儿。

API数据获取：别当手动搬运工

我记得刚入行那会儿，有个同事每天手动从网站复制粘贴行情数据，结果有一次粘贴错了小数点，整个回测全废了。嗯，从那以后我就养成了一个习惯——能用API绝不用手动。

API说白了就是数据接口，你写几行代码，数据就自动拉下来了。常用的金融数据API有这些：

数据源	特点	适合场景
Wind	国内最全，收费	机构级研究
Tushare	免费开源，社区活跃	个人研究、快速验证
AKShare	纯Python，接口丰富	爬虫替代方案
Yahoo Finance	全球数据，免费	美股研究

我个人习惯用Tushare做快速验证，毕竟免费又方便。但要注意，免费API通常有频率限制，我曾在一次批量拉取时被限流，结果数据缺了一大块。所以建议你加个延时，比如每次请求后sleep(0.5)。

import tushare as ts
import time

# 初始化
pro = ts.pro_api('你的token')

# 获取日线数据，加延时避免限流
def fetch_stock_data(code, start, end):
    df = pro.daily(ts_code=code, start_date=start, end_date=end)
    time.sleep(0.5)  # 我踩过的坑，不加这行容易被封
    return df

# 示例：获取茅台数据
df_maotai = fetch_stock_data('600519.SH', '20230101', '20231231')
print(df_maotai.head())

我的小技巧： 写个重试机制，网络波动时自动重试3次。我曾经因为没加重试，凌晨跑的数据采集任务全挂了，第二天才发现。

网页爬虫基础：当API不够用时

说实话，API虽然方便，但很多另类数据根本没有API接口。比如某个论坛的情绪指标、某家公司的招聘数据、某个商品的销量排名……这些都得靠爬虫。

爬虫的核心就三步：请求网页 → 解析内容 → 提取数据。我用的是requests + BeautifulSoup的组合，轻量又好用。

import requests
from bs4 import BeautifulSoup
import pandas as pd

# 请求网页
url = 'https://example.com/stock-comments'
headers = {'User-Agent': 'Mozilla/5.0'}  # 伪装成浏览器，别被反爬
resp = requests.get(url, headers=headers)

# 解析内容
soup = BeautifulSoup(resp.text, 'html.parser')

# 提取数据
comments = []
for item in soup.select('.comment-item'):
    text = item.find('.text').text.strip()
    date = item.find('.date').text.strip()
    comments.append({'text': text, 'date': date})

df_comments = pd.DataFrame(comments)
print(f'共爬取 {len(df_comments)} 条评论')

避坑指南： 我曾经爬一个论坛，没加headers直接请求，结果IP被封了三天。后来学乖了，不仅加headers，还用了代理池。另外，尊重网站的robots.txt，别做违法的事。

数据清洗流程：脏数据才是常态

数据拉下来了，你以为就完事了？太天真了。真实世界的数据，十个里面有八个是脏的。我见过缺失值、异常值、重复值、格式不统一……各种妖魔鬼怪。

清洗流程我总结为四步：

缺失值处理——要么删，要么补
异常值检测——别让极端值毁了你的因子
标准化——让不同量纲的数据能放在一起比
一致性检查——日期格式、股票代码格式统一

缺失值处理

缺失值分三种：随机缺失、非随机缺失、完全随机缺失。处理方式也分几种：

直接删除：缺失比例小于5%时可以用，但别删太多
均值/中位数填充：简单粗暴，适合数值型数据
前向/后向填充：时间序列数据常用，比如用昨天的数据补今天的
插值法：线性插值、多项式插值，更精确但计算量大

import pandas as pd
import numpy as np

# 模拟有缺失的数据
df = pd.DataFrame({
    'date': pd.date_range('2023-01-01', periods=10),
    'close': [100, 102, np.nan, 105, 107, np.nan, 110, 112, 115, np.nan],
    'volume': [1000, 1200, 1100, np.nan, 1300, 1250, 1400, np.nan, 1500, 1450]
})

# 方法1：删除缺失行
df_drop = df.dropna()
print(f'删除后剩余 {len(df_drop)} 行')

# 方法2：前向填充（时间序列常用）
df_ffill = df.fillna(method='ffill')
print('前向填充结果：')
print(df_ffill)

# 方法3：线性插值（我个人偏好这个）
df_interp = df.interpolate(method='linear')
print('线性插值结果：')
print(df_interp)

  我的经验： 对于金融时间序列，我一般先用前向填充，再用线性插值做微调。别用均值填充，因为金融数据有趋势，均值会破坏序列的连续性。

异常值检测

异常值是什么？就是那些明显不合理的数据点。比如某只股票一天涨了1000%，或者成交量突然变成平时的100倍。这些很可能是数据错误，而不是真正的市场信号。

常用的检测方法：

3σ原则：超出均值±3倍标准差的数据视为异常
箱线图法：超出1.5倍IQR（四分位距）的数据视为异常
Z-score法：Z-score绝对值大于3的视为异常

# 使用Z-score检测异常值
from scipy import stats

# 模拟数据，包含一个异常值
data = [100, 102, 101, 103, 102, 100, 101, 500, 102, 103]

z_scores = np.abs(stats.zscore(data))
threshold = 3
outliers = np.where(z_scores > threshold)[0]

print(f'异常值索引: {outliers}')
print(f'异常值: {[data[i] for i in outliers]}')

# 处理：用中位数替换异常值
median_val = np.median(data)
data_clean = [median_val if i in outliers else val for i, val in enumerate(data)]
print(f'清洗后数据: {data_clean}')

注意： 我曾经在回测中没处理异常值，结果一个错误数据让夏普比率从2.0变成了0.5。所以别偷懒，异常值一定要处理。但也要小心别把真正的极端行情（比如熔断）当成异常值删掉了。

标准化：让因子站在同一起跑线

你想想看，市盈率可能是几倍到几百倍，换手率是百分之零点几到百分之几十，这两个因子怎么放在一起比较？标准化就是干这个的。

常用的标准化方法：

方法	公式	适用场景
Z-score标准化	(x - μ) / σ	数据近似正态分布时
Min-Max标准化	(x - min) / (max - min)	数据有明确边界时
Rank标准化	排序后映射到[0,1]	因子有极端值时

from sklearn.preprocessing import StandardScaler, MinMaxScaler
import pandas as pd

# 模拟因子数据
df_factors = pd.DataFrame({
    'pe': [15, 20, 8, 50, 12, 30],
    'turnover': [0.5, 1.2, 0.3, 2.5, 0.8, 1.5],
    'momentum': [0.05, -0.02, 0.08, -0.1, 0.03, 0.06]
})

# Z-score标准化（我个人最常用）
scaler = StandardScaler()
df_zscore = pd.DataFrame(scaler.fit_transform(df_factors), 
                         columns=df_factors.columns)
print('Z-score标准化结果：')
print(df_zscore)

# Min-Max标准化
scaler_mm = MinMaxScaler()
df_minmax = pd.DataFrame(scaler_mm.fit_transform(df_factors),
                         columns=df_factors.columns)
print('\nMin-Max标准化结果：')
print(df_minmax)

我的建议： 对于大多数因子，我推荐用Z-score标准化。但如果你的因子有极端值（比如市值因子），先用Rank标准化会更稳健。记住，标准化要在训练集上拟合，再应用到测试集，避免数据泄露。

数据存储：CSV还是Parquet？

数据清洗完了，得存起来吧？存哪里？怎么存？这里也有讲究。

CSV：人类可读，通用性强，但文件大、读写慢。适合小数据量（几万行以内）或者需要人工检查的场景。

Parquet：列式存储，压缩率高，读写快。适合大数据量（百万行以上）或者频繁读写的场景。我个人的量化数据库，全部用Parquet格式。

import pandas as pd

# 模拟清洗后的数据
df_clean = pd.DataFrame({
    'date': pd.date_range('2023-01-01', periods=1000),
    'code': ['600519.SH'] * 1000,
    'close': np.random.randn(1000).cumsum() + 100,
    'volume': np.random.randint(1000, 10000, 1000)
})

# 保存为CSV
df_clean.to_csv('stock_data.csv', index=False)
print('CSV保存完成')

# 保存为Parquet（需要安装pyarrow或fastparquet）
df_clean.to_parquet('stock_data.parquet', index=False)
print('Parquet保存完成')

# 对比文件大小
import os
csv_size = os.path.getsize('stock_data.csv')
parquet_size = os.path.getsize('stock_data.parquet')
print(f'CSV大小: {csv_size/1024:.1f} KB')
print(f'Parquet大小: {parquet_size/1024:.1f} KB')
print(f'压缩比: {csv_size/parquet_size:.1f}x')

  我的选择： 日常研究用CSV，方便查看。正式回测和实盘用Parquet，速度快、省空间。另外，建议按日期分区存储，比如每天一个Parquet文件，这样增量更新很方便。

本章知识体系

说了这么多，咱们用一张图来总结一下数据采集与清洗的完整流程：

数据采集与清洗，说白了就是给因子模型准备食材。食材不好，再好的厨子也做不出好菜。我见过太多人花几个月写策略，却不愿意花几天清洗数据，最后回测结果漂亮，实盘一塌糊涂。

嗯，这一章的内容就到这里。记住，数据质量决定了你策略的天花板。下一章咱们聊聊因子构建，到时候你会发现，数据清洗做得好，后面的事情会顺利很多。

📚 课程目录

数据采集与清洗：从原始数据到可用因子

API数据获取：别当手动搬运工

网页爬虫基础：当API不够用时

数据清洗流程：脏数据才是常态

缺失值处理

异常值检测

标准化：让因子站在同一起跑线

数据存储：CSV还是Parquet？

本章知识体系