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

原创

用户7083614

发布于 2026-09-06 10:31:44

一句话回答： 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

避坑要点（诚实提示）

坑	说明
货币/债 ETF 混入	低波动会“霸榜”尾部或污染排名，务必先剔除
折溢价	ETF 二级市场价可能偏离净值，极端行情尤甚
流动性陷阱	小规模 ETF 成交稀、冲击成本高，设成交额门槛
换手成本	轮动频率越高手续费越高，别过度交易
动量反转	动量会失效甚至反转，务必回测并控制回撤

和其他数据源比

能力	免费聚合库	AlphaFeed
全市场 ETF 快照	需自行遍历、易被限流	`universes="CN_ETF"` 一行
批量 K 线	自行实现并发	`klines.batch` 自动分批
涨跌幅/成交额字段	需清洗	快照直接给

常见问题（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 Python SDK 文档：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python

pandas

#量化交易

#股票数据