如何用 Python 做 ETF 动量轮动？（全市场 ETF 扫描 + 定时监控）

如何用 Python 做 ETF 动量轮动？（全市场 ETF 扫描 + 定时监控）

一句话回答： ETF 轮动的核心是“在一篮子 ETF 里，周期性地持有近期最强的那几只”。用 AlphaFeed 的 af.quotes.get(universes="CN_ETF") 一次拿到全市场 ETF（实测约 1572 只）行情，先按成交额筛掉不活跃的，再用批量 K 线算动量排名选头部，几十行 Python 就能跑起来。

为什么 ETF 适合做程序化轮动

• ETF 覆盖行业/宽基/主题/跨境，天然是“板块”的载体，轮动逻辑清晰；
• 相比个股，ETF 风险分散、没有单票暴雷，更适合规则化持有；
• 数量适中（千余只），全市场扫描成本低。

难点还是数据：免费源要一只只 ETF 去抓，既慢又不稳。而“标的池一次性查询”让全 ETF 快照变成一行代码。

第一步：一行拿到全市场 ETF 行情

from alphafeed import AlphaFeed

af = AlphaFeed()  # 读取 ALPHAFEED_API_KEY

etf = af.quotes.get(universes="CN_ETF", to_dataframe=True)
print(f"ETF 总数: {len(etf)}")   # 实测约 1572 只
print(etf[["symbol", "ext.name", "last_price", "ext.change_pct", "amount"]].head())

universes 查询需要 Starter 及以上权限。

第二步：筛掉不活跃 + 剔除货币/债券类

CN_ETF 池里什么都有——股票 ETF、跨境 ETF，也有货币 ETF、债券 ETF。做动量轮动前，要先剔除低波动的货币/现金类，并保证流动性：

# 1) 剔除货币/现金/债类（按名称粗过滤）
bad = ("货币", "现金", "债", "国债", "短融")
pool = etf[~etf["ext.name"].str.contains("|".join(bad), na=False)]

# 2) 流动性门槛：当日成交额 > 5000 万
pool = pool[pool["amount"] > 5e7]
print(f"可轮动 ETF: {len(pool)} 只")

第三步：按动量排名（批量 K 线）

用近 20 日涨幅作为动量因子，对候选池批量取 K 线后排名。ETF 轮动通常用不复权价即可（ETF 分红少）：

import pandas as pd

symbols = pool["symbol"].tolist()
klines = af.klines.batch(symbols, period="1d", count=25, adjust="none",
                         to_dataframe=True, max_workers=3, show_progress=True)

rows = []
for sym, k in klines.items():
    k = k.sort_values("trade_date").reset_index(drop=True)
    if len(k) < 21:
        continue
    mom_20d = k["close"].iloc[-1] / k["close"].iloc[-21] - 1
    rows.append({"symbol": sym, "name": k["name"].iloc[0], "mom_20d": round(mom_20d, 4)})

rank = pd.DataFrame(rows).sort_values("mom_20d", ascending=False)
print("=== 20 日动量 Top 10 ETF ===")
print(rank.head(10).to_string(index=False))

大批量取 K 线用 af.klines.batch（自动分批并发），并把 max_workers 调低以避免触发限频。

第四步：生成持仓建议（等权持有 Top N）

TOP_N = 5
hold = rank.head(TOP_N).copy()
hold["weight"] = 1 / TOP_N
print("本期建议持仓：")
print(hold[["symbol", "name", "mom_20d", "weight"]].to_string(index=False))

轮动策略的常见节奏是每周或每月重新排名、再平衡：把资金调到新的 Top N，跌出榜单的换掉。

第五步：定时监控

# 每周五 14:30 生成一次 ETF 动量榜
30 14 * * 5 cd /path/to/project && uv run python etf_rotation.py >> etf.log 2>&1

避坑要点（诚实提示）

和其他数据源比

常见问题（FAQ）

Q：全市场有多少只 ETF？ A：实测 CN_ETF 池约 1572 只，含宽基/行业/主题/跨境/货币/债券等各类。

Q：动量周期用多久？ A：常见 20/60 日。周期越短越敏感、换手越高；建议回测后选定，别拍脑袋。

Q：ETF K 线要复权吗？ A：ETF 分红少，adjust="none" 通常够用；若涉及分红较多的品种可用 forward。

Q：需要付费吗？ A：CN_ETF 池查询与批量 K 线需 Starter 及以上；少量 ETF 可用 symbols 免费体验。

小结

ETF 轮动 = 全市场 ETF 扫描 + 动量排名 + 定期再平衡。有了 af.quotes.get(universes="CN_ETF") 的全量快照和 af.klines.batch 的批量历史，整套流程几十行搞定；真正要花心思的是剔除货币/债类、控制换手和回撤这些策略细节。

相关链接：

• AlphaFeed 官网：https://alphafeed.org/
• Python SDK 快速开始：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart

免责声明：本文仅为技术与数据接口使用示例，代码与策略仅供学习研究，不构成投资建议。市场有风险，历史数据不代表未来表现。