数据获取与清洗（上）：使用yfinance获取历史数据、使用ib_insync连接盈透证券获取实时数据、数据格式标准化
做量化交易，尤其是高频策略，第一步就是搞定数据。没有数据，再牛的算法也是空中楼阁。我个人习惯把数据获取分成两条线：一条是历史数据，用来回测；另一条是实时数据，用来实盘。今天咱们就先把这两条线打通，顺便把数据格式统一一下。
一、用yfinance拿历史数据——简单粗暴但够用
yfinance这个库，说白了就是雅虎财经的非官方Python接口。我刚开始做美股回测时，第一个用的就是它。为什么？因为免费、简单、不用注册API Key。你想想看，对于快速验证策略想法来说，这简直是神器。
安装很简单：
pip install yfinance基本用法：
import yfinance as yf
# 下载苹果股票从2023年1月到2024年1月的日线数据
data = yf.download('AAPL', start='2023-01-01', end='2024-01-01')
print(data.head())输出结果长这样：
| Date | Open | High | Low | Close | Volume |
|---|---|---|---|---|---|
| 2023-01-03 | 130.28 | 131.74 | 128.12 | 129.93 | 74651200 |
| 2023-01-04 | 128.90 | 130.39 | 127.31 | 128.67 | 64857600 |
嗯，这里要注意：yfinance返回的数据是Pandas DataFrame格式，索引是日期。但它的列名是大写的，而且Open/High/Low/Close/Volume这些字段名跟我们后面要用的IB数据不太一样。所以标准化这一步必不可少。
二、用ib_insync连接盈透证券——实时数据的正确打开方式
如果你要做实盘交易，或者需要更精细的分钟级、tick级数据，那就得用盈透证券（Interactive Brokers）了。ib_insync这个库，我个人觉得是连接IB最好用的Python库，没有之一。
为什么？因为它把IB复杂的API封装成了异步模式，用起来特别顺手。我当年第一次用IB原生API时，被那些回调函数搞得头大。后来换了ib_insync，整个世界清净了。
安装：
pip install ib_insync连接并获取实时数据：
from ib_insync import *
# 创建连接
ib = IB()
ib.connect('127.0.0.1', 7497, clientId=1) # 7497是TWS的端口，7496是Gateway的
# 定义合约
contract = Stock('AAPL', 'SMART', 'USD')
# 请求历史数据（5分钟K线）
bars = ib.reqHistoricalData(
contract,
endDateTime='',
durationStr='2 D',
barSizeSetting='5 mins',
whatToShow='TRADES',
useRTH=True,
formatDate=1
)
# 转成DataFrame
df = util.df(bars)
print(df.head())输出结果：
| date | open | high | low | close | volume |
|---|---|---|---|---|---|
| 2024-01-02 09:30:00 | 185.64 | 186.23 | 185.12 | 185.89 | 1234567 |
看到区别了吗？IB返回的列名是小写的，而且时间格式是带时区的。这就是为什么我们需要做数据格式标准化。
三、数据格式标准化——统一战场
做量化交易最怕什么？数据格式不统一。你从yfinance拿到的数据是大写列名，从IB拿到的是小写列名，从其他数据源拿到的可能又是另一种格式。我见过最离谱的一次，有个同事把Open拼成了"OPen"，结果回测跑出来全是错的。
所以，我建议所有数据进来后，先做标准化处理。标准格式如下：
- datetime：统一为UTC时间，列名用'datetime'
- open, high, low, close：统一小写
- volume：统一小写
- symbol：添加股票代码列
标准化函数示例：
def standardize_data(df, source='yfinance'):
"""
将不同来源的数据标准化为统一格式
"""
df = df.copy()
if source == 'yfinance':
# yfinance的列名是大写的
df.columns = [col.lower() for col in df.columns]
# 重置索引，把日期变成普通列
df = df.reset_index()
df.rename(columns={'date': 'datetime'}, inplace=True)
elif source == 'ib':
# IB的列名已经是小写，但时间列名是'date'
df.rename(columns={'date': 'datetime'}, inplace=True)
# 统一添加symbol列
if 'symbol' not in df.columns:
df['symbol'] = 'AAPL' # 这里可以根据实际情况修改
# 确保列顺序一致
cols = ['datetime', 'symbol', 'open', 'high', 'low', 'close', 'volume']
df = df[cols]
return df
# 使用示例
yf_data = yf.download('AAPL', start='2023-01-01', end='2023-01-10')
yf_std = standardize_data(yf_data, source='yfinance')
print(yf_std.head())四、本章知识体系总览
下面这张图，是我做数据获取与清洗时的核心流程。你看一眼就能明白整个逻辑：
这张图的核心逻辑很简单：不管数据从哪来，最终都要经过标准化处理，变成统一的格式。这样你的回测引擎和实盘策略才能无缝对接。
五、一些实用技巧
- 数据缓存：yfinance每次请求都有延迟，我建议把下载的数据存成CSV或Parquet格式，避免重复下载。特别是做回测时，数据量大了网络请求很慢。
- IB连接重试：IB的连接有时会断开，尤其是长时间运行。我习惯在代码里加一个重试机制，每隔30秒检查一次连接状态。
- 时区处理：美股数据默认是美东时间，但你的策略可能用UTC。记得在标准化时统一时区，不然回测结果会差几个小时。
好了，数据获取与清洗的上半部分就到这里。记住：数据质量决定了策略的天花板。你花再多时间在数据清洗上都不为过。下一节我们会继续深入，聊聊数据清洗中的那些坑——缺失值处理、异常值检测、复权数据等等。