第二章：数据获取与清洗——用pandas搞定历史行情数据

做量化交易，第一步就是搞数据。没有干净的数据，再牛的策略也是白搭。我个人习惯把数据获取和清洗看作整个交易系统的地基——地基不稳，楼盖得再高也得塌。

这一章，咱们就聊聊怎么用pandas把历史行情数据拿到手，然后处理那些烦人的缺失值、对齐不同时间周期的数据、以及重采样。嗯，都是实战中每天都要面对的问题。

2.1 数据获取：从哪儿拿，怎么拿

获取历史行情数据，说白了就两条路：要么从本地文件读，要么从网络API拉。我建议初学者先从本地文件开始，省得被网络问题搞心态。

2.1.1 从CSV文件读取

最常见的格式就是CSV。pandas的read_csv函数，我用了不下上千次。这里有个小坑：日期列默认会被当成字符串，你得手动转成datetime类型。

import pandas as pd

# 读取CSV文件
df = pd.read_csv('stock_data.csv')

# 把日期列转成datetime类型
df['date'] = pd.to_datetime(df['date'])

# 把日期列设为索引
df.set_index('date', inplace=True)

print(df.head())

我在项目中遇到过一个问题：有些数据源导出的CSV文件，日期格式是"2024/01/15"这种斜杠分隔的。直接用pd.to_datetime能自动识别，但偶尔会遇到混合格式的情况。这时候我建议加上format参数，明确告诉pandas你的日期长什么样。

2.1.2 从API获取数据

如果你用的是Tushare、AKShare这类数据源，代码会稍微复杂一点。但核心逻辑不变：拿到数据后，第一时间检查数据质量。

import akshare as ak

# 获取股票历史数据
stock_data = ak.stock_zh_a_hist(symbol="000001", 
                                period="daily", 
                                start_date="20240101", 
                                end_date="20241231", 
                                adjust="qfq")

# 看一眼数据长什么样
print(stock_data.info())
print(stock_data.head())

你想想看，API返回的数据格式五花八门。有的列名叫"收盘价"，有的叫"close"，有的干脆叫"CLOSE"。我习惯在拿到数据后，第一时间统一列名，省得后面写代码时搞混。

2.2 处理缺失值：别让空数据坑了你

缺失值，是量化交易里最烦人的问题之一。为什么会缺失？原因很多：停牌、数据源没更新、节假日...嗯，这里要注意：千万不要直接删掉所有缺失值，你得先搞清楚缺失的原因。

2.2.1 检查缺失值

# 检查每列的缺失值数量
print(df.isnull().sum())

# 查看缺失值占比
print(df.isnull().mean() * 100)

我曾经遇到过一个案例：某只股票连续停牌5天，数据里全是NaN。如果直接删掉这些行，那复牌后的跳空缺口就完全看不到了，策略信号会出大问题。

2.2.2 填充缺失值

处理缺失值，我一般按这个优先级来：

1. 向前填充：用上一个非空值填充，适合停牌场景
2. 向后填充：用下一个非空值填充，适合数据延迟场景
3. 插值法：线性插值，适合短时间缺失
4. 固定值填充：比如用0填充成交量

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

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

# 用0填充成交量
df['volume'].fillna(0, inplace=True)

2.3 数据对齐：让不同周期的数据对上号

做多因子策略时，你经常需要把日线数据和分钟线数据对齐。说白了，就是让不同时间频率的数据，在同一个时间点上都有值。

2.3.1 按时间索引对齐

pandas的reindex方法，就是干这个的。我习惯先创建一个完整的时间序列，然后用它来对齐所有数据。

# 创建一个完整的时间序列（交易日历）
trade_dates = pd.date_range(start='2024-01-01', 
                            end='2024-12-31', 
                            freq='B')  # B代表工作日

# 对齐日线数据
df_daily = df_daily.reindex(trade_dates)

# 对齐分钟线数据（取每天收盘时刻）
df_minute = df_minute.resample('D').last()
df_minute = df_minute.reindex(trade_dates)

你想想看，如果两个数据框的索引不完全一致，直接做运算会出大问题。pandas虽然会自动对齐，但对齐后可能出现大量NaN。我建议在对齐后，再检查一下对齐结果。

2.3.2 多表合并

有时候你需要把不同股票的数据合并到一个表里。这时候merge或concat就派上用场了。

# 合并多只股票的数据
stock_list = ['000001', '000002', '000003']
dfs = []

for stock in stock_list:
    df = get_stock_data(stock)
    df.columns = [f'{stock}_{col}' for col in df.columns]
    dfs.append(df)

# 按日期合并
combined_df = pd.concat(dfs, axis=1)

2.3 数据重采样：把高频数据变低频

重采样，就是把数据从一个时间频率转换到另一个。最常见的场景：把分钟线数据转成日线数据。

2.3.1 降采样（高频转低频）

降采样时，你需要指定聚合方式。开盘价取第一个，收盘价取最后一个，最高价取最大值，最低价取最小值，成交量求和。

# 把5分钟线转成日线
daily_data = minute_data.resample('D').agg({
    'open': 'first',
    'high': 'max',
    'low': 'min',
    'close': 'last',
    'volume': 'sum'
})

我在项目中遇到过一个问题：有些数据源在收盘后还会更新数据，导致"last"取到的不是真正的收盘价。后来我加了个时间过滤，只取15:00之前的数据。

2.3.2 升采样（低频转高频）

升采样相对少见，但做回测时偶尔会用到。比如把日线数据转成小时线，用于模拟盘中交易。

# 把日线转成小时线（向前填充）
hourly_data = daily_data.resample('H').ffill()

# 或者用插值
hourly_data = daily_data.resample('H').interpolate()

2.4 实战：构建一个完整的数据清洗流程

说了这么多，咱们来写一个完整的清洗函数。这是我个人项目中一直在用的模板，你可以直接拿来用。

def clean_market_data(df, date_col='date', freq='D'):
    """
    清洗市场数据
    """
    # 1. 日期处理
    df[date_col] = pd.to_datetime(df[date_col])
    df.set_index(date_col, inplace=True)
    
    # 2. 排序
    df.sort_index(inplace=True)
    
    # 3. 去重
    df = df[~df.index.duplicated(keep='first')]
    
    # 4. 处理缺失值
    # 成交量用0填充
    if 'volume' in df.columns:
        df['volume'].fillna(0, inplace=True)
    
    # 价格数据用向前填充
    price_cols = ['open', 'high', 'low', 'close']
    for col in price_cols:
        if col in df.columns:
            df[col] = df[col].fillna(method='ffill')
    
    # 5. 重采样到指定频率
    if freq != 'D':
        df = df.resample(freq).agg({
            'open': 'first',
            'high': 'max',
            'low': 'min',
            'close': 'last',
            'volume': 'sum'
        })
    
    # 6. 删除异常值（比如价格为负）
    for col in price_cols:
        if col in df.columns:
            df.loc[df[col] <= 0, col] = None
    
    return df

这个函数我用了快两年，基本覆盖了日常80%的清洗需求。当然，具体场景还得具体调整。比如期货数据有换月问题，股票数据有复权问题，这些都需要额外处理。

2.5 本章小结

数据获取与清洗，说白了就是三个步骤：拿到数据、处理缺失、对齐频率。听起来简单，但每个环节都有坑。我建议你养成一个习惯：每次拿到新数据，先花10分钟做数据质量检查，看看有没有异常值、缺失值、重复值。这10分钟，能帮你省下后面几小时的调试时间。

嗯，数据准备好了，下一章咱们就可以开始计算技术指标了。不过在那之前，我建议你把今天的内容动手实践一遍。找一只股票的历史数据，跑一遍清洗流程，看看你能发现什么问题。