TL;DR：如果你刚开始学量化，不建议一上来就研究几千只股票。股票数量多、停牌多、涨跌停多、财务和行业因素复杂，新手极易在数据清洗与交易规则里迷路。相对而言，ETF 轮动 更加清晰、透明且抗噪。本文将带你基于 QuantDash 统一金融数据 API，在 Python 环境下实现一个完整的多资产 ETF 动量轮动策略：从数据批量获取、日期索引对齐、持仓矩阵生成，到加入真实摩擦成本的回测绩效评估，手把手带你完成一整套可复现的量化研究流[1]。

---

1. 什么是 ETF 轮动？

ETF 轮动是一种典型的相对强弱（动量）策略。
假设你筛选并关注以下几个具有低相关性的资产方向：

1. 宽基指数 ETF（如沪深300 ETF、中证500 ETF）

2. 科技与成长主题 ETF（如科创50 ETF、创业板 ETF）

3. 红利与价值主题 ETF（如红利 ETF）

4. 商品与黄金 ETF（如黄金 ETF）

5. 跨境与海外市场 ETF（如纳指 ETF）

每隔一段时间（如每周、双周或每月），你计算每个 ETF 过去一段时间的累计收益率（动量指标），然后选择表现最强（涨幅最大）的一个或几个持有。

背后的假设：市场趋势具有一定的延续性，近期强势的资产在短期内有较大概率维持强势（动量效应）。当然，趋势随时会反转，因此我们必须通过严格的回测、加入真实的交易成本、并配合空仓阈值来全面评估其可行性。

---

2. 准备 ETF 数据

QuantDash 统一支持股票和 ETF 的行情数据，其标的代码格式完全标准化（如 A 股 ETF 通常使用交易所后缀 .SH 或 .SZ）[1]。

首先，确保安装了 SDK：

pip install quantdash pandas matplotlib

我们可以先拉取单只 ETF 历史 K 线作为测试[1]：

from quantdash import QuantDash

# 初始化客户端，会自动读取环境变量 QUANTDASH_API_KEY
qd = QuantDash(api_key="demo_public_token")  # 演示环境可使用公共测试 Token

df = qd.klines.get(
    symbol="510300.SH", 
    period="1d", 
    count=300, 
    adjust="forward", 
    to_dataframe=True
)
print(df.tail())

如果想看全市场 ETF 的实时行情快照，可以直接调用[1]：

etfs = qd.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
]

(注：在生产环境中，你可以根据基金规模、日均成交额等流动性指标，通过 universes 接口动态筛选出你的交易候选池)。

---

3. 批量获取 ETF 日 K

在做多资产回测时，最繁琐的往往是各标的历史日期的对齐（由于停牌、上市时间不同、节假日等原因）[4]。
QuantDash 的 batch 接口 完美解决了这一痛点，传入列表即可由 SDK 内部全自动处理分块并发，极速返回整理好的数据字典[4]：

from quantdash import QuantDash

qd = QuantDash(api_key="demo_public_token")
dfs = qd.klines.batch(
    symbols=ETF_POOL,
    period="1d",
    count=1000,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

我们将获取到的各标的数据中的收盘价，整理并对齐到同一个时间矩阵中：

import pandas as pd

def build_close_matrix(dfs: dict[str, pd.DataFrame]) -> pd.DataFrame:
    series = []
    for symbol, df in dfs.items():
        if df is None or df.empty:
            continue
        # 按交易日期排序，并以日期为索引
        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 价格矩阵就是我们接下来生成信号与执行回测的坚实底座。

---

4. 定义轮动信号

我们使用一个极其直观但逻辑闭环的动量轮动规则：

1. 每天计算每个 ETF 过去 N 个交易日（本例采用 60 日）的累计收益率作为动量指标。

2. 在所有 ETF 中，选择动量指标最高的那只。

3. 安全垫/空仓机制：如果所有 ETF 的 60 日动量收益率均小于 0（说明全市场处于系统性下跌期），组合选择空仓。

4. 【规避未来函数】：今天生成的交易决策，明天才持有，严格作 shift(1) 延迟处理，避免未来函数。

# 设定动量观察周期为 60 个交易日
lookback = 60

# 计算累计收益率矩阵
momentum = close / close.shift(lookback) - 1

# 找到每天动量值最大的 ETF 标的及其对应的动量值
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，则该列为 1，其余列全为 0。如果所有 ETF 动量都小于 0，则全为 0。

---

5. 计算组合收益与交易成本

在得到决策信号后，我们开始计算组合实际获取的日收益，并引入真实的交易磨损成本：

# 1. 计算单个标的的日收益率
ret = close.pct_change().fillna(0)

# 2. 规避未来函数：信号后移一天，即今日开盘根据昨晚收盘信号进行建仓/调仓
position = signal.shift(1).fillna(0)

# 3. 计算未考虑成本的组合基础日收益率
portfolio_ret = (position * ret).sum(axis=1)

# 4. 引入交易摩擦：设定费率及滑点
fee = 0.0003       # 单边佣金/印花税
slippage = 0.0002  # 单边交易滑点

# 计算每日调仓换手率（持仓权重变化的绝对值之和）
turnover = position.diff().abs().sum(axis=1).fillna(0)
cost = turnover * (fee + slippage)

# 5. 扣除摩擦成本，得到最终的实际收益与净值曲线
portfolio_ret_after_cost = portfolio_ret - cost
equity_after_cost = (1 + portfolio_ret_after_cost).cumprod()

回测的每一步都足够真实、没有“偷看未来收盘价”的作弊可能，这才能给后续的参数扫描和实盘监控提供有价值的参考。

---

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("=== 策略回测报告 ===")
for k, v in metrics.items():
    print(f"{k}: {v:.4f}")

绘制净值曲线图

import matplotlib.pyplot as plt

plt.figure(figsize=(10, 5))
equity_after_cost.plot(title="ETF Rotation Equity Curve (After Cost)")
plt.grid(True)
plt.tight_layout()
plt.savefig("etf_rotation_equity.png", dpi=160)
plt.show()

7. 穿透分析：输出每天持有了什么？

只看单一的累计收益率曲线是不够的，回测系统不能做成黑箱。我们输出最近 20 天的持仓详情，透视策略的运作轨迹：

# 找到每日持仓列名称
holding = position.idxmax(axis=1)
# 判断当天组合是否有持仓（如果全为0，表示空仓避险）
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("\n=== 最近 20 天持仓与净值透视 ===")
print(report.tail(20))

通过穿透报告你可以清楚了解：

• 策略最近持有什么标的。

• 哪一天发生了 ETF 切换，是否符合强弱转换的逻辑。

• 空仓（CASH）发生在哪些暴跌阶段，避险机制是否确实起到了防守作用。

---

8. 参数扫描：寻找更稳健的动量窗口

“60 日动量”好，还是“20 日动量”好？动量周期是轮动策略最敏感的超参数之一，我们需要通过参数扫描（Parameter Sweep）来确保策略的稳健度：

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_after_cost = portfolio_ret - cost
    equity = (1 + portfolio_ret_after_cost).cumprod()
    
    # 评价指标
    metrics = calc_metrics(equity, portfolio_ret_after_cost)
    metrics["lookback"] = lookback
    return metrics

# 扫描 20 日、40 日、60 日、120 日动量周期
results = [run_rotation(close, lb) for lb in [20, 40, 60, 120]]
result_df = pd.DataFrame(results).sort_values("sharpe", ascending=False)

print("\n=== 参数扫描结果 (按夏普比率降序) ===")
print(result_df.to_string(index=False))

在量化开发中，不要一味去追求并采用夏普比率最高的那一个特定最优参数（过度拟合风险）。相反，更理性的选择是：寻找一个宽广的参数平台（比如 40 日至 80 日周期均表现出相近且不错的绩效），这样的参数在未来实盘中才具有更强的抗噪能力。

---

9. Cursor / DeepSeek 专属 AI 编程 Prompt

为了方便你围绕这套 ETF 轮动回测框架进行个性化升级（例如：修改选股逻辑、引入动量权重分配、引入止损线），你可以直接把这段 Prompt 复制给 Cursor、Claude 3.5 Sonnet 或 DeepSeek：

你是一位精通 Python 的量化开发专家。我正在使用 QuantDash 作为 ETF 数据源，设计并迭代我的【ETF多资产动量轮动回测系统】。
请记住我的基础回测组件代码框架：

1. 批量获取数据并对齐矩阵：
dfs = qd.klines.batch(symbols=ETF_POOL, period="1d", count=1000, adjust="forward", to_dataframe=True)

2. 整理为 close 价格矩阵，索引为 trade_date (DatetimeIndex 格式)：
def build_close_matrix(dfs):
    series = [df.sort_values("trade_date").set_index("trade_date")["close"].rename(sym) for sym, df in dfs.items()]
    return pd.concat(series, axis=1).sort_index().dropna(how="all")

3. 生成信号（signal 矩阵：行代表日期，列代表 ETF 标的，1为持有，0为空仓）。

4. 规避未来函数：持仓 position = signal.shift(1).fillna(0)。

5. 扣除交易成本（单边交易摩擦 fee + slippage = 0.0005），计算最终净值曲线。

需求：
请在此基础框架上，帮我升级为一个【多周期动量共振 + 波动率过滤轮动策略】。
规则修改：
- 计算每个 ETF 过去 20 日、60 日和 120 日的动量收益率，并按 0.3, 0.4, 0.3 的权重分配，计算一个综合动量得分。
- 另外，计算每个 ETF 过去 20 日的年化波动率。如果某只得分最高的 ETF 其 20 日年化波动率大于 40%，则强制空仓（不参与轮动，视为异常高波），回到 CASH 状态。
- 请提供结构优美、写好完整中文注释的模块化 Python 代码，并包含保存图表、回测绩效输出和最近 20 天持仓状态打印。

总结：从玩具回测到专业研究

ETF 轮动是普通人学习量化的好入口。它足够简单，能让你快速跑出结果；又足够真实，能覆盖数据对齐、持仓矩阵、交易成本、参数扫描这些关键问题。

对于新手来说，在这个完整的流里你至少实践了以下八个关键环节：

1. 选股池界定（不同资产低相关性组合）

2. 多标的数据批量抓取（基于 QuantDash 接口的高速 Batch）

3. 时序对齐（Pandas 跨标的时间轴标准清洗）

4. 决策信号生成（动量矩阵计算）

5. 规避未来函数（严格执行 shift(1) 延迟成交）

6. 组合日收益累计（非事件驱动的向量化极速回测）

7. 真实摩擦磨损建模（佣金与滑点扣除）

8. 评价体系构建（夏普比率与回撤测算）

QuantDash 在这个过程里承担的是数据底座的角色。你用统一的 Python SDK 获取 ETF K 线和实时行情，把精力放在策略假设和回测逻辑上，而不是反复修数据接口。先写一个简单版本，再一点点加规则，量化研究就是这样长出来的。

---

“三步走”落地指引

• 第一步：获取完整源码。请访问官方开源托管仓库获取本文完整 ETF 轮动代码及更多跨市场高频套利回测 Demo：https://github.com/quantdash-net/QuantDash（请认准官方 quantdash-net 组织，欢迎 Star 支持）。

• 第二步：申请专属密钥。注册获取您的个人免费/生产级 API Key：https://quantdash.net/。

• 第三步：查阅开发细节。更多高频行情、多市场 K 线接口及实时 universes 全量数据集参数请参考：https://docs.quantdash.net/。

---

免责声明: 本文仅用于量化技术教学和策略回测方法演示目的，不构成任何投资建议。市场有风险，投资需谨慎。