4、Pandas数据处理：数据读取、清洗、重采样、缺失值处理、时间索引操作

金融回测系统的地基是数据。无论策略逻辑多么精妙，如果喂给回测引擎的价格序列存在跳空、错位、时区混乱或缺失值，最终得到的绩效曲线都可能是虚假的。Pandas 是 Python 金融数据分析的核心库，其 DataFrame 和 Series 结构天然适配时间序列数据。本章将围绕金融数据处理的四个关键环节展开：数据读取与格式校验、数据清洗与去重、时间序列重采样、缺失值处理与时间索引的高级操作。我们将从原理出发，结合真实交易数据的常见问题，给出可落地的操作流程与代码范式。

上图展示了金融数据从原始输入到回测引擎的完整流水线。每个环节都可能引入错误，因此必须建立标准化的处理流程。下面我们逐一深入每个模块。

4.1 数据读取：从文件到 DataFrame 的严谨转换

核心原理：金融数据通常以 CSV、Parquet、HDF5 或数据库表形式存储。Pandas 的 read_csv() 是最常用的入口，但默认参数往往不适合金融数据。关键在于显式指定时间列、数据类型和解析规则，避免 Pandas 自动推断带来的隐患。

操作流程：

1. 使用 parse_dates 参数将时间列转换为 DatetimeIndex。
2. 通过 dtype 参数指定价格、成交量为数值类型，避免将 "1,234.56" 解析为字符串。
3. 设置 index_col 为时间列，使 DataFrame 直接以时间为索引。
4. 使用 encoding='utf-8' 或 'gbk' 处理中文文件。

import pandas as pd
df = pd.read_csv('HS300_1min.csv',
                 parse_dates=['datetime'],
                 dtype={'open': 'float64', 'high': 'float64',
                        'low': 'float64', 'close': 'float64',
                        'volume': 'int64'},
                 index_col='datetime',
                 encoding='utf-8')
print(df.head())
print(df.index.dtype)  # 确认是 datetime64[ns]

4.2 数据清洗：去重、排序与异常值过滤

核心原理：金融数据源经常出现重复时间戳（同一秒内多次报价）、乱序记录和极端异常值（如价格为 0 或负数）。清洗的目标是保证时间索引的唯一性和单调递增性。

判断方法：

• 检查索引是否唯一：df.index.is_unique
• 检查索引是否单调递增：df.index.is_monotonic_increasing
• 检查价格字段的合理范围：df[df['close'] <= 0]

操作流程：

# 1. 去重：保留每个时间戳的最后一条记录
df = df[~df.index.duplicated(keep='last')]

# 2. 排序
df = df.sort_index()

# 3. 过滤异常价格（例如：价格偏离 20 日均线 50% 以上视为异常）
ma20 = df['close'].rolling(20).mean()
std20 = df['close'].rolling(20).std()
df = df[(df['close'] > ma20 - 5*std20) & (df['close'] < ma20 + 5*std20)]

# 4. 删除成交量小于等于 0 的记录
df = df[df['volume'] > 0]

4.3 重采样：从高频到低频的聚合艺术

核心原理：回测策略可能基于 1 分钟线，但数据源只有 tick 级数据，或者需要将日线转为周线进行长周期测试。Pandas 的 resample() 方法提供了灵活的时间桶聚合功能。

操作流程：

# 将 1 分钟数据重采样为 5 分钟 K 线
df_5min = df.resample('5min').agg({
    'open': 'first',
    'high': 'max',
    'low': 'min',
    'close': 'last',
    'volume': 'sum'
}).dropna()

# 将日线重采样为周线（按周五收盘价）
df_weekly = df_daily.resample('W-FRI').agg({
    'open': 'first',
    'high': 'max',
    'low': 'min',
    'close': 'last',
    'volume': 'sum'
})

案例拆解：某量化团队需要将 5000 万条 tick 数据转换为 1 分钟 K 线。初始做法是直接 resample('1min')，但发现内存占用过高。优化方案是：先按日期分组，再对每组进行重采样，最后合并。这样将内存峰值从 8GB 降至 2GB，耗时从 15 分钟降至 3 分钟。

4.4 缺失值处理：插值、填充与删除的权衡

核心原理：金融时间序列的缺失值通常由停牌、数据源中断或非交易时段造成。处理策略取决于缺失的原因和后续分析需求。

判断方法：

• 先区分是"自然缺失"（非交易时间）还是"异常缺失"（交易时间无数据）。
• 检查缺失比例：df.isnull().mean()
• 观察缺失模式：连续缺失还是随机缺失。

操作流程：

# 1. 前向填充（适用于停牌后恢复，用最后有效价格）
df['close_ffill'] = df['close'].ffill()

# 2. 线性插值（适用于短期数据中断）
df['close_interp'] = df['close'].interpolate(method='linear')

# 3. 时间加权插值（考虑时间间隔）
df['close_time'] = df['close'].interpolate(method='time')

# 4. 删除缺失值（仅当缺失比例小于 1% 且非连续）
df_clean = df.dropna(subset=['close'])

4.5 时间索引高级操作：切片、偏移与时区处理

核心原理：金融数据的时间索引不仅是排序键，更是策略逻辑的一部分。需要掌握按时间切片、日期偏移、时区转换和交易日历对齐。

操作流程：

# 1. 时间切片（注意包含端点）
df_2024 = df['2024-01-01':'2024-12-31']
df_morning = df.between_time('09:30', '11:30')

# 2. 日期偏移（计算前一个交易日）
from pandas.tseries.offsets import BDay
df['prev_trade'] = df.index - BDay(1)

# 3. 时区处理（美股数据转北京时间）
df_et = df.tz_localize('America/New_York')
df_beijing = df_et.tz_convert('Asia/Shanghai')

# 4. 重采样到交易日历（跳过节假日）
import exchange_calendars as xcals
cn_cal = xcals.get_calendar('XSHG')
df_trading = df.reindex(cn_cal.sessions_in_range('2024-01-01', '2024-12-31'))

常见误区：

• 使用 shift(1) 时，默认按行偏移，而不是按时间偏移。如果数据有缺失，应使用 df['close'].shift(1, freq='D') 按日历日偏移。
• 时区转换后，索引的 tz 属性会改变，但底层时间戳不变。务必确认回测引擎期望的时区。
• 不要用 pd.to_datetime(df['date']) 后直接作为索引，应先排序再设置索引，否则后续切片可能出错。

4.6 案例闭环：从脏数据到可回测的 5 分钟 K 线

背景与约束：某私募团队获得一份中证 500 指数成分股的 1 分钟 tick 数据，共 3 个月，约 1.2 亿行。数据存在以下问题：时间戳为字符串且含时区偏移、部分股票有重复记录、停牌期间价格为 0、成交量缺失。团队需要在 2 小时内生成干净的 5 分钟 K 线用于因子测试。

初始现象：直接读取后，df.shape 显示 1.2 亿行，但 df.index.is_unique 为 False，且 df['close'].min() 为 0。

分析判断过程：

1. 时间戳解析：原始格式为 "2024-03-01 09:30:00+08:00"，需用 pd.to_datetime(..., utc=True) 统一为 UTC，再转换到 Asia/Shanghai。
2. 重复记录：同一股票同一秒内有多条报价，保留最后一条（最新成交价）。
3. 零价格：停牌期间价格为 0，应转为 NaN，后续用 ffill() 填充，但成交量设为 0。
4. 缺失成交量：用 0 填充，因为停牌无成交。

实际操作步骤：

# 步骤 1：分块读取，避免内存爆炸
chunks = pd.read_csv('tick_data.csv', chunksize=5000000, parse_dates=['timestamp'])
df_list = []
for chunk in chunks:
    chunk['timestamp'] = pd.to_datetime(chunk['timestamp'], utc=True).tz_convert('Asia/Shanghai')
    chunk = chunk.set_index('timestamp')
    chunk = chunk[~chunk.index.duplicated(keep='last')]
    chunk['close'] = chunk['close'].replace(0, np.nan)
    chunk['volume'] = chunk['volume'].fillna(0)
    df_list.append(chunk)
df = pd.concat(df_list).sort_index()

# 步骤 2：按股票分组重采样
def resample_stock(group):
    return group.resample('5min').agg({
        'open': 'first', 'high': 'max', 'low': 'min',
        'close': 'last', 'volume': 'sum'
    })
df_5min = df.groupby('stock_code').apply(resample_stock).dropna(subset=['close'])

# 步骤 3：验证结果
assert df_5min.index.is_monotonic_increasing
assert df_5min['close'].isnull().sum() == 0

结果对比：处理前，直接重采样会得到 1800 万行包含零价格和重复时间戳的数据；处理后，得到 1200 万行干净数据，因子计算耗时减少 40%，且回测结果与手动核对的历史行情一致。

风险与代价：使用 groupby().apply() 在 1.2 亿行数据上较慢（约 20 分钟）。优化方案是使用 groupby().resample() 的向量化操作，但需要确保每个股票的时间索引对齐。此外，停牌股票的 ffill() 会引入未来信息，必须在策略中设置交易状态标志。

复盘经验：

• 清洗规则必须与数据源特性匹配，不能盲目套用通用模板。
• 对于高频数据，分块处理是必须的，但要注意分块边界的时间戳连续性。
• 验证环节不能只看行数，要检查关键统计量（如最高价是否大于最低价、成交量是否非负）。