3. 数据获取：用Python获取ETF历史行情数据，数据清洗与预处理
好了，到了真正动手的环节了。
前面我们聊了策略逻辑，聊了ETF怎么选。但说句实话，没有数据，一切都是空谈。量化交易说白了就是「用数据说话」的行业。你策略再漂亮，拿到的数据是脏的、错的，那结果就是垃圾进垃圾出。
我个人习惯，在开始任何策略开发前，先花30%的时间把数据这块搞定。别嫌慢，这步省了，后面调试会让你哭出来。
3.1 数据源的选择：免费的午餐在哪里？
获取ETF行情数据，渠道其实不少。我列几个常用的，你根据自己情况选。
| 数据源 | 特点 | 推荐指数 |
|---|---|---|
| Tushare | 老牌金融数据接口，数据全，但新用户有积分限制 | ⭐⭐⭐⭐ |
| AkShare | 开源免费，接口丰富，更新快，我目前主力用这个 | ⭐⭐⭐⭐⭐ |
| Baostock | 免费，无需注册，数据质量稳定 | ⭐⭐⭐⭐ |
| 东方财富/新浪 | 直接爬取，但容易被封，维护成本高 | ⭐⭐⭐ |
我的建议：新手直接上AkShare。免费、不用折腾注册、文档也写得清楚。我在项目中遇到过几次数据源变更的情况，AkShare的社区活跃度最高，基本当天就有人更新修复。
3.2 用AkShare获取ETF历史数据
先装库，这个不用我多说了吧？
pip install akshare装好之后，我们来看看怎么拿数据。以我最常跟踪的「沪深300ETF（510300）」为例。
import akshare as ak
import pandas as pd
# 获取ETF历史行情
# 参数说明：symbol是ETF代码，start_date和end_date控制时间范围
df = ak.fund_etf_hist_em(
symbol="510300",
period="daily",
start_date="20200101",
end_date="20241231",
adjust="qfq" # 前复权，这个很重要
)
print(df.head())
print(df.columns.tolist())
注意：adjust参数一定要设成"qfq"（前复权）。我曾经吃过这个亏——直接用原始价格做回测，结果策略表现好得离谱。后来才发现是因为没复权，分红除权造成的价格跳空被当成了交易信号。嗯，那感觉就像考试抄到了答案，结果发现抄的是隔壁班的卷子。
跑完代码，你会看到类似这样的数据：
| 日期 | 开盘 | 收盘 | 最高 | 最低 | 成交量 | 成交额 |
|---|---|---|---|---|---|---|
| 2020-01-02 | 3.892 | 3.912 | 3.918 | 3.882 | 1.2亿 | 4.7亿 |
| 2020-01-03 | 3.908 | 3.885 | 3.915 | 3.878 | 0.9亿 | 3.5亿 |
3.3 数据清洗：脏数据是策略的隐形杀手
数据拿到手了，但别急着用。你想想看，真实市场里什么情况都可能发生：停牌、数据缺失、异常波动。这些脏数据不处理，策略回测结果就是骗自己的。
我一般按以下步骤清洗：
- 检查缺失值——看看有没有空数据
- 处理停牌日——停牌期间价格不变，但成交量是0
- 去除无效数据——比如价格<=0的记录
- 统一索引——把日期设为索引，方便后续操作
# 数据清洗流程
def clean_etf_data(df):
# 1. 选择需要的列，重命名
df = df[['日期', '开盘', '收盘', '最高', '最低', '成交量']]
df.columns = ['date', 'open', 'close', 'high', 'low', 'volume']
# 2. 日期格式转换
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)
# 3. 按日期排序
df.sort_index(inplace=True)
# 4. 处理缺失值
# 如果某天数据缺失，用前一天的收盘价填充（保守做法）
df.fillna(method='ffill', inplace=True)
# 5. 去除异常值
# 价格不可能为0或负数
df = df[df['close'] > 0]
df = df[df['volume'] >= 0]
# 6. 检查停牌日
# 成交量极低的日子可能是停牌或流动性枯竭
# 我一般会标记出来，但不删除
df['is_suspended'] = df['volume'] < 1000 # 假设1000手以下视为停牌
return df
# 执行清洗
df_clean = clean_etf_data(df)
print(f"清洗前数据量: {len(df)}")
print(f"清洗后数据量: {len(df_clean)}")
核心原则：清洗数据时，宁可漏掉一些边缘数据，也不要引入错误数据。我在做CTA策略时，就因为没处理停牌日的数据，导致策略在复牌那天开出了天量仓位，回测曲线直接崩了。
3.4 数据预处理：为策略计算铺路
清洗完了，数据基本能用了。但为了后续计算方便，我还会做几项预处理。
说白了，就是给数据「加料」，让策略模块可以直接调用。
# 数据预处理
def preprocess_etf_data(df):
# 1. 计算收益率（对数收益率更符合金融假设）
df['log_return'] = np.log(df['close'] / df['close'].shift(1))
# 2. 计算简单收益率
df['simple_return'] = df['close'].pct_change()
# 3. 计算日均价格（用于某些策略）
df['avg_price'] = (df['open'] + df['high'] + df['low'] + df['close']) / 4
# 4. 计算振幅
df['range'] = (df['high'] - df['low']) / df['close']
# 5. 删除第一行（因为收益率计算会产生NaN）
df.dropna(inplace=True)
return df
# 应用预处理
df_final = preprocess_etf_data(df_clean)
print(df_final.tail())你可能会问：为什么要算对数收益率？
嗯，这里简单解释一下。对数收益率有个好处——它是可加的。你想想看，如果你要算一周的总收益率，简单收益率需要连乘，而对数收益率直接相加就行。这在后面做统计分析时会方便很多。
3.5 本章知识体系
为了让你对整个流程有个直观印象，我画了张图。数据获取到预处理，其实就是一条流水线：
3.6 多ETF批量获取
做轮动策略，肯定不止跟踪一只ETF。我习惯写个循环，一次性把候选池里的数据都拉下来。
# 候选ETF池
etf_pool = {
'510300': '沪深300ETF',
'510500': '中证500ETF',
'159915': '创业板ETF',
'510050': '上证50ETF',
'513100': '纳指ETF'
}
# 批量获取并清洗
def load_etf_data(symbol, name, start='20200101', end='20241231'):
df = ak.fund_etf_hist_em(
symbol=symbol,
period="daily",
start_date=start,
end_date=end,
adjust="qfq"
)
df = clean_etf_data(df)
df = preprocess_etf_data(df)
df['etf_name'] = name
df['etf_code'] = symbol
return df
# 执行批量加载
all_etf_data = {}
for code, name in etf_pool.items():
print(f"正在获取 {name} ({code})...")
all_etf_data[code] = load_etf_data(code, name)
print(f" 完成，共 {len(all_etf_data[code])} 条记录")
# 合并成一个DataFrame，方便后续分析
combined_df = pd.concat(all_etf_data.values(), keys=all_etf_data.keys())
print(combined_df.head())
小技巧：第一次拉数据时，建议把结果保存到本地CSV文件。这样下次跑策略时直接读本地文件，不用每次都请求接口。我在项目中就是这么干的——每天收盘后跑一次更新脚本，把数据增量追加到本地文件里。
# 保存到本地
combined_df.to_csv('etf_data_pool.csv')
# 下次直接读取
df_loaded = pd.read_csv('etf_data_pool.csv', index_col=0, parse_dates=True)好了，数据这块就聊到这儿。你拿到干净的数据后，下一章我们就可以开始真正搭建轮动策略了。记住，数据是你的原材料，质量决定了最终产品的档次。