用 Python 做一个 ETF 轮动策略:普通人也能理解的多资产量化入门
如果你刚开始学量化,我不建议一上来就研究几千只股票。
股票数量多、停牌多、涨跌停多、财务和行业因素复杂。新手很容易在数据处理、选股池、交易规则里迷路。
ETF 轮动反而是一个更适合入门的方向。
它的核心思想很简单:在一组 ETF 里,定期选择近期表现更强的品种持有。如果所有品种都不好,就降低仓位或空仓。
这篇文章我们用 AlphaFeed 写一个最小可运行的 ETF 轮动策略。它不追求复杂,也不承诺收益。重点是展示一个清晰、可复现、可继续迭代的量化研究流程。
1. 什么是 ETF 轮动
ETF 轮动是一种相对强弱策略。
假设你关注这些方向:
- 宽基指数 ETF。
- 科技或成长 ETF。
- 红利或价值 ETF。
- 债券或货币类 ETF。
- 黄金、商品或海外市场 ETF。
每隔一段时间,比如每周或每月,你计算每个 ETF 过去一段时间的收益率,选择表现最强的一个或几个持有。
背后的假设是:市场趋势有一定延续性。近期强的资产,短期内可能继续强。
当然,这不是永远有效。趋势会反转,轮动会失灵,所以必须通过回测和风险控制来评估。
2. 准备 ETF 数据
AlphaFeed 支持 ETF 行情和 K 线。代码格式和股票一致,比如 A 股 ETF 通常使用交易所后缀:
from alphafeed import AlphaFeed
af = AlphaFeed()
df = af.klines.get(
"510300.SH",
period="1d",
count=300,
adjust="forward",
to_dataframe=True,
)
print(df.tail())
如果你想看全量 ETF 实时行情,可以使用:
etfs = af.quotes.get(universes="CN_ETF", to_dataframe=True)
print(etfs.head())
为了方便演示,我们先手动设定一个 ETF 池。实际研究时,你可以根据规模、成交额、跟踪指数等条件筛选。
ETF_POOL = [
"510300.SH", # 沪深300 ETF 示例
"510500.SH", # 中证500 ETF 示例
"159915.SZ", # 创业板 ETF 示例
"588000.SH", # 科创50 ETF 示例
"518880.SH", # 黄金 ETF 示例
]
注意:不同账户和数据权限下,具体可用标的请以接口返回为准。
3. 批量获取 ETF 日 K
from alphafeed import AlphaFeed
af = AlphaFeed()
dfs = af.klines.batch(
ETF_POOL,
period="1d",
count=1000,
adjust="forward",
to_dataframe=True,
show_progress=True,
)
我们先把每个 ETF 的收盘价整理成一个矩阵:
import pandas as pd
def build_close_matrix(dfs: dict[str, pd.DataFrame]) -> pd.DataFrame:
series = []
for symbol, df in dfs.items():
one = df.sort_values("trade_date").set_index("trade_date")["close"]
one.name = symbol
series.append(one)
close = pd.concat(series, axis=1).sort_index()
close = close.dropna(how="all")
return close
close = build_close_matrix(dfs)
print(close.tail())
这个 close 就是后面所有研究的基础:
| trade_date | 510300.SH | 510500.SH | 159915.SZ | ... |
|---|---|---|---|---|
| 2026-06-01 | ... | ... | ... | ... |
| 2026-06-02 | ... | ... | ... | ... |
多资产策略的第一步,往往就是把不同标的对齐到同一个日期索引上。
4. 定义轮动信号
我们用一个非常简单的规则:
- 每天计算每个 ETF 过去 60 个交易日收益率。
- 选择收益率最高的 ETF。
- 如果最高收益率小于 0,则空仓。
- 今天生成的选择,明天才持有,避免未来函数。
lookback = 60
momentum = close / close.shift(lookback) - 1
selected = momentum.idxmax(axis=1)
best_momentum = momentum.max(axis=1)
signal = pd.DataFrame(0, index=close.index, columns=close.columns)
for date in close.index:
if pd.notna(best_momentum.loc[date]) and best_momentum.loc[date] > 0:
symbol = selected.loc[date]
signal.loc[date, symbol] = 1
这里的 signal 是一个持仓矩阵。
如果某天选择 510300.SH,那这一行里 510300.SH 为 1,其他 ETF 为 0。如果所有 ETF 动量都小于 0,则全为 0。
5. 计算组合收益
先计算每个 ETF 的日收益:
ret = close.pct_change().fillna(0)
再把信号后移一天:
position = signal.shift(1).fillna(0)
组合收益:
portfolio_ret = (position * ret).sum(axis=1)
equity = (1 + portfolio_ret).cumprod()
加入交易成本:
fee = 0.0003
slippage = 0.0002
turnover = position.diff().abs().sum(axis=1).fillna(0)
cost = turnover * (fee + slippage)
portfolio_ret_after_cost = portfolio_ret - cost
equity_after_cost = (1 + portfolio_ret_after_cost).cumprod()
完整回测逻辑并不复杂,但每一步都要清楚:
| 步骤 | 目的 |
|---|---|
| 计算动量 | 找近期更强的 ETF |
| 生成 signal | 记录当日选择 |
| signal 后移 | 避免未来函数 |
| 计算收益 | 得到组合日收益 |
| 加交易成本 | 更接近真实交易 |
| 生成净值 | 观察长期表现 |
6. 计算指标
import numpy as np
def calc_metrics(equity: pd.Series, returns: pd.Series) -> dict:
total_return = equity.iloc[-1] / equity.iloc[0] - 1
annual_return = (1 + total_return) ** (252 / len(equity)) - 1
drawdown = equity / equity.cummax() - 1
max_drawdown = drawdown.min()
sharpe = 0 if returns.std() == 0 else returns.mean() / returns.std() * np.sqrt(252)
return {
"total_return": total_return,
"annual_return": annual_return,
"max_drawdown": max_drawdown,
"sharpe": sharpe,
}
metrics = calc_metrics(equity_after_cost, portfolio_ret_after_cost)
print(metrics)
再画净值曲线:
import matplotlib.pyplot as plt
equity_after_cost.plot(figsize=(10, 5), title="ETF Rotation Equity")
plt.tight_layout()
plt.savefig("etf_rotation_equity.png", dpi=160)
7. 输出每天持有了什么
只看净值不够。轮动策略还应该能解释每天持仓。
holding = position.idxmax(axis=1)
has_position = position.sum(axis=1) > 0
holding = holding.where(has_position, "CASH")
report = pd.DataFrame({
"holding": holding,
"portfolio_ret": portfolio_ret_after_cost,
"equity": equity_after_cost,
})
print(report.tail(20))
这份报告能告诉你:
- 策略最近持有什么。
- 什么时候切换了 ETF。
- 空仓发生在哪些阶段。
- 净值变化和持仓是否对应。
策略不能只是一个黑箱。尤其是新手阶段,你必须能解释结果。
8. 参数扫描:20 日、60 日、120 日动量谁更稳
轮动策略最重要的参数之一是动量周期。
我们可以扫描多个 lookback:
def run_rotation(close: pd.DataFrame, lookback: int) -> dict:
momentum = close / close.shift(lookback) - 1
selected = momentum.idxmax(axis=1)
best_momentum = momentum.max(axis=1)
signal = pd.DataFrame(0, index=close.index, columns=close.columns)
for date in close.index:
if pd.notna(best_momentum.loc[date]) and best_momentum.loc[date] > 0:
signal.loc[date, selected.loc[date]] = 1
ret = close.pct_change().fillna(0)
position = signal.shift(1).fillna(0)
portfolio_ret = (position * ret).sum(axis=1)
turnover = position.diff().abs().sum(axis=1).fillna(0)
cost = turnover * 0.0005
portfolio_ret = portfolio_ret - cost
equity = (1 + portfolio_ret).cumprod()
metrics = calc_metrics(equity, portfolio_ret)
metrics["lookback"] = lookback
return metrics
results = [run_rotation(close, lb) for lb in [20, 40, 60, 120]]
result_df = pd.DataFrame(results).sort_values("sharpe", ascending=False)
print(result_df)
还是那句话:不要只看最优参数。你要看不同参数下结果是否都还能接受。
如果只有 40 日动量很好,20、60、120 都不行,那要警惕过拟合。
9. 可以继续升级的方向
ETF 轮动是一个很好的量化练习框架,因为它可以逐步加复杂度:
| 升级方向 | 示例 |
|---|---|
| 多持仓 | 每次持有动量前 2 或前 3 |
| 波动率调整 | 对高波动 ETF 降低权重 |
| 趋势过滤 | 只有指数在长期均线上方才开仓 |
| 再平衡频率 | 从每日切换改成每周或每月 |
| 成交额过滤 | 排除流动性较差的 ETF |
| 加入债券/黄金 | 提高资产类别分散度 |
比如多持仓版本,可以让策略更平滑:
top_n = 2
signal = pd.DataFrame(0, index=close.index, columns=close.columns)
for date in momentum.index:
row = momentum.loc[date].dropna()
row = row[row > 0].sort_values(ascending=False).head(top_n)
if len(row) > 0:
signal.loc[date, row.index] = 1 / len(row)
这时策略不再把所有资金压到一个 ETF 上,而是分散到近期表现较强的多个 ETF。
10. 为什么 ETF 轮动适合新手
ETF 轮动有几个优点:
- 标的数量少,容易理解。
- 数据结构简单,适合 pandas 练习。
- 不需要复杂财务数据。
- 策略逻辑直观,容易解释。
- 很适合学习多资产回测。
它也有缺点:
- 趋势反转时可能回撤。
- 参数容易过拟合。
- 频繁切换会产生交易成本。
- ETF 选择会影响结果。
- 过去有效不代表未来有效。
新手最应该学到的不是“ETF 轮动一定能赚钱”,而是完整研究流程:选池、取数、对齐、生成信号、延迟持仓、计算收益、加入成本、输出指标、检查参数稳定性。
这套流程学会了,以后研究股票、指数、美股、港股都能复用。
结语
ETF 轮动是普通人学习量化的好入口。
它足够简单,能让你快速跑出结果;又足够真实,能覆盖数据对齐、持仓矩阵、交易成本、参数扫描这些关键问题。
AlphaFeed 在这个过程里承担的是数据底座的角色。你用统一的 Python SDK 获取 ETF K 线和实时行情,把精力放在策略假设和回测逻辑上,而不是反复修数据接口。
先写一个简单版本,再一点点加规则。量化研究就是这样长出来的。
相关链接:

