4. 数据获取与清洗：从零搭建你的量化数据管道

做量化择时，第一步不是写策略，而是搞定数据。

我见过太多人，策略逻辑写得天花乱坠，结果一跑回测，数据全是坑——缺失值、不复权、时间轴对不齐……嗯，那感觉就像你开着一辆法拉利，结果油箱里加的是水。

这一章，我们就来亲手搭建一条干净、可靠的数据管道。我会用 tushare 获取日线数据，然后一步步清洗、复权、对齐、重采样。你跟着做一遍，以后的数据处理就能少踩 80% 的坑。

4.1 用 tushare 获取日线数据

先说说 tushare。这是国内用得最多的免费数据源之一。我个人习惯用 tushare pro 版本，数据质量比旧版好不少。

你需要先去 tushare.pro 注册，拿到你的 token。然后安装：

pip install tushare

获取单只股票的日线数据，代码很简单：

import tushare as ts

# 设置 token（换成你自己的）
ts.set_token('你的token_here')
pro = ts.pro_api()

# 获取平安银行 2023年全年日线数据
df = pro.daily(
    ts_code='000001.SZ',
    start_date='20230101',
    end_date='20231231'
)

print(df.head())

返回的数据包含这些字段：

字段名	含义	说明
trade_date	交易日期	格式 YYYYMMDD
open	开盘价	元
high	最高价	元
low	最低价	元
close	收盘价	元
pre_close	前收盘价	用于复权计算
change	涨跌额	元
pct_chg	涨跌幅	%
vol	成交量	手
amount	成交额	千元

4.2 数据缺失值处理——别让空值毁了你的策略

数据拿到手，第一件事不是分析，而是检查有没有缺失值。我在项目中遇到过好几次，策略回测曲线漂亮得不行，结果一查，是因为某几天数据是 NaN，策略根本没交易——虚惊一场。

先看看缺失情况：

# 检查缺失值
print(df.isnull().sum())

# 看看哪些行有缺失
print(df[df.isnull().any(axis=1)])

常见的处理方式有三种：

• 删除缺失行：如果缺失比例很小（比如 < 1%），直接 dropna() 最省事。
• 前向填充：用上一个交易日的值填充。对于价格数据，这其实挺合理——你想想看，停牌一天，复牌后价格参考前一天。
• 插值：用前后值线性插值。适合成交量这种连续变化的数据。

# 前向填充（我最常用的方式）
df.fillna(method='ffill', inplace=True)

# 或者线性插值
df.interpolate(method='linear', inplace=True)

4.3 复权价格计算——还原真实的收益率

为什么要复权？说白了，股票分红送股后，价格会跳空。如果你直接用原始价格算收益率，那结果就是扭曲的。

我个人习惯用前复权。前复权调整历史价格，让价格序列连续，这样你看到的收益率就是真实的持有期收益率。

tushare 提供了复权因子，我们可以自己算：

# 获取复权因子
adj = pro.adj_factor(ts_code='000001.SZ', 
                     start_date='20230101', 
                     end_date='20231231')

# 合并到日线数据
df = df.merge(adj[['trade_date', 'adj_factor']], on='trade_date')

# 计算前复权价格
df['close_adj'] = df['close'] * df['adj_factor']
df['open_adj'] = df['open'] * df['adj_factor']
df['high_adj'] = df['high'] * df['adj_factor']
df['low_adj'] = df['low'] * df['adj_factor']

这里有个细节：tushare 的 adj_factor 是累积复权因子。你用收盘价乘以它，得到的就是前复权价格。后复权则是除以它，得到的是以上市首日为基准的价格。

4.4 数据对齐——让多只股票站在同一起跑线

做量化择时，你很少只盯一只股票。但不同股票的交易日期不完全一样——有的停牌，有的刚上市。这就需要对时间轴。

我常用的方法是：先构建一个完整的时间索引，然后对所有股票做 outer join。

import pandas as pd

# 假设我们有两只股票的数据
stocks = ['000001.SZ', '600036.SH']
data_dict = {}

for code in stocks:
    df = pro.daily(ts_code=code, start_date='20230101', end_date='20231231')
    df = df.sort_values('trade_date')
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    df.set_index('trade_date', inplace=True)
    data_dict[code] = df['close_adj']

# 合并成一个 DataFrame，缺失值自动变成 NaN
aligned = pd.DataFrame(data_dict)
print(aligned.head())

对齐之后，你会发现有些日期只有部分股票有数据。这时候再用前向填充处理一下：

aligned.fillna(method='ffill', inplace=True)
# 如果最前面几天还是 NaN，直接 drop 掉
aligned.dropna(inplace=True)

4.5 重采样——从日线到周线、月线

有些择时策略用日线，有些用周线甚至月线。重采样就是把高频数据降为低频。

重采样的核心是确定每个周期的代表价格：

• 开盘价：周期第一天的开盘价
• 收盘价：周期最后一天的收盘价
• 最高价：周期内的最高价
• 最低价：周期内的最低价
• 成交量：周期内成交量之和

# 日线重采样为周线
weekly = df.resample('W').agg({
    'open_adj': 'first',
    'high_adj': 'max',
    'low_adj': 'min',
    'close_adj': 'last',
    'vol': 'sum'
})

# 日线重采样为月线
monthly = df.resample('M').agg({
    'open_adj': 'first',
    'high_adj': 'max',
    'low_adj': 'min',
    'close_adj': 'last',
    'vol': 'sum'
})

这里要注意：resample 的 'W' 默认以周日为周的最后一天。A 股市场周五是最后交易日，所以我习惯用 'W-FRI'：

weekly = df.resample('W-FRI').agg({...})

4.6 完整的数据处理流水线

把上面所有步骤串起来，就是一个完整的数据管道。我个人习惯把它封装成一个函数：

def get_clean_data(ts_code, start_date, end_date):
    # 1. 获取数据
    df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date)
    df = df.sort_values('trade_date')
    
    # 2. 获取复权因子并计算复权价
    adj = pro.adj_factor(ts_code=ts_code, start_date=start_date, end_date=end_date)
    df = df.merge(adj[['trade_date', 'adj_factor']], on='trade_date')
    df['close_adj'] = df['close'] * df['adj_factor']
    
    # 3. 处理缺失值
    df.fillna(method='ffill', inplace=True)
    df.dropna(inplace=True)
    
    # 4. 设置日期索引
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    df.set_index('trade_date', inplace=True)
    
    return df

这样，每次拿数据只需要一行代码：

df_000001 = get_clean_data('000001.SZ', '20200101', '20231231')

嗯，干净利落。

---

数据获取与清洗，听起来枯燥，但它是所有量化策略的地基。地基不稳，上面盖多高的楼都没用。这一章的内容，你花半小时跑一遍代码，以后就能省下无数个排查数据问题的夜晚。