3. Python 金融数据处理基础

高频策略的成败，90% 取决于数据质量。本章将系统讲解 Pandas 在金融时间序列中的核心操作，包括数据加载、清洗、重采样以及通过 API 获取实时行情。掌握这些技能后，你就能为任何策略搭建可靠的数据管道。

3.1 Pandas 核心数据结构与金融数据加载

Pandas 的两大核心对象是 Series（一维序列）和 DataFrame（二维表）。金融数据通常以 OHLCV（开盘、最高、最低、收盘、成交量）格式存储，最适合用 DataFrame 表示。

对象	维度	金融应用场景
Series	1D	单只股票的收盘价序列、收益率序列
DataFrame	2D	多只股票的 OHLCV 表、因子矩阵、订单簿快照

从 CSV 加载历史数据的标准写法：

import pandas as pd

# 加载日线数据，指定日期列为索引
df = pd.read_csv('AAPL_daily.csv', parse_dates=['Date'], index_col='Date')
print(df.head())

3.2 时间序列数据处理

金融时间序列具有高频、非均匀间隔、多层级（tick/分钟/日/月）的特点。Pandas 提供了强大的时间索引操作。

3.2.1 索引与切片

# 按年份切片
df_2023 = df['2023']

# 按月份范围切片
df_jan_feb = df['2024-01':'2024-02']

# 按具体日期精确取值
df_specific = df.loc['2024-03-15']

3.2.2 滚动窗口与滞后

高频策略中，移动平均、滚动波动率、滞后因子是基础算子。

# 20 日滚动均值
df['SMA_20'] = df['Close'].rolling(window=20).mean()

# 5 日滚动标准差（波动率）
df['Volatility_5'] = df['Close'].rolling(window=5).std()

# 滞后 1 期（前一交易日收盘价）
df['Close_Lag1'] = df['Close'].shift(1)

3.3 数据清洗与重采样

真实金融数据常包含缺失值、异常跳空、非交易时段噪音。清洗是策略稳健性的第一道防线。

3.3.1 缺失值处理

方法	适用场景	代码示例
前向填充	非交易时段（周末、节假日）	df.fillna(method='ffill')
线性插值	日内 tick 数据短时缺失	df.interpolate(method='linear')
丢弃	缺失比例极低（<1%）	df.dropna()

3.3.2 重采样：将高频数据降频

高频策略常需将 tick 数据聚合为分钟线，或将分钟线聚合为日线。

# 将 1 分钟数据重采样为 5 分钟 OHLCV
df_5min = df.resample('5T').agg({
    'Open': 'first',
    'High': 'max',
    'Low': 'min',
    'Close': 'last',
    'Volume': 'sum'
})

# 将日线重采样为周线
df_weekly = df.resample('W').agg({
    'Open': 'first',
    'High': 'max',
    'Low': 'min',
    'Close': 'last',
    'Volume': 'sum'
})

3.4 金融数据 API 调用（Yahoo Finance）

使用 yfinance 库可以零成本获取全球股票、ETF、指数的历史行情。这是策略回测和实时监控的常用数据源。

3.4.1 安装与基础调用

# 安装：pip install yfinance
import yfinance as yf

# 下载单只股票日线数据
aapl = yf.download('AAPL', start='2023-01-01', end='2024-12-31')
print(aapl.head())

3.4.2 批量下载与多级列索引

# 同时下载多只股票
tickers = ['AAPL', 'MSFT', 'GOOGL']
data = yf.download(tickers, start='2024-01-01', end='2024-06-30')

# 此时 data 为多级列索引：(Open, AAPL), (Open, MSFT), ...
# 提取所有股票的收盘价
close_prices = data['Close']
print(close_prices.head())

3.4.3 实时行情与历史分笔

# 获取当前最新行情（延迟约 15 分钟）
ticker = yf.Ticker('AAPL')
info = ticker.info
print(f"当前价格: {info['currentPrice']}")

# 获取历史分笔数据（最近 7 天）
hist = ticker.history(period='7d', interval='1m')
print(hist.tail())

3.5 综合实战：构建分钟级数据管道

将以上知识点串联，构建一个从 API 获取 → 清洗 → 重采样 → 存储的完整流程。

import yfinance as yf
import pandas as pd

# 1. 获取原始分钟数据
raw = yf.download('AAPL', period='5d', interval='1m')

# 2. 清洗：前向填充缺失值
clean = raw.fillna(method='ffill')

# 3. 重采样为 5 分钟 OHLCV
resampled = clean.resample('5T').agg({
    'Open': 'first',
    'High': 'max',
    'Low': 'min',
    'Close': 'last',
    'Volume': 'sum'
})

# 4. 添加技术指标
resampled['SMA_5'] = resampled['Close'].rolling(5).mean()
resampled['Return'] = resampled['Close'].pct_change()

# 5. 保存到本地
resampled.to_parquet('AAPL_5min.parquet')
print("数据管道执行完成，共 %d 条记录" % len(resampled))