用 Python 做一个 ETF 轮动策略:普通人也能理解的多资产量化入门(附 GitHub 源码)

发布时间:2026/7/26 10:51:29

用 Python 做一个 ETF 轮动策略:普通人也能理解的多资产量化入门(附 GitHub 源码) TL;DR如果你刚开始学量化不建议一上来就研究几千只股票。股票数量多、停牌多、涨跌停多、财务和行业因素复杂新手极易在数据清洗与交易规则里迷路。相对而言ETF 轮动更加清晰、透明且抗噪。本文将带你基于QuantDash统一金融数据 API在 Python 环境下实现一个完整的多资产 ETF 动量轮动策略从数据批量获取、日期索引对齐、持仓矩阵生成到加入真实摩擦成本的回测绩效评估手把手带你完成一整套可复现的量化研究流[1]。1. 什么是 ETF 轮动ETF 轮动是一种典型的相对强弱动量策略。假设你筛选并关注以下几个具有低相关性的资产方向宽基指数 ETF如沪深300 ETF、中证500 ETF科技与成长主题 ETF如科创50 ETF、创业板 ETF红利与价值主题 ETF如红利 ETF商品与黄金 ETF如黄金 ETF跨境与海外市场 ETF如纳指 ETF每隔一段时间如每周、双周或每月你计算每个 ETF 过去一段时间的累计收益率动量指标然后选择表现最强涨幅最大的一个或几个持有。背后的假设市场趋势具有一定的延续性近期强势的资产在短期内有较大概率维持强势动量效应。当然趋势随时会反转因此我们必须通过严格的回测、加入真实的交易成本、并配合空仓阈值来全面评估其可行性。2. 准备 ETF 数据QuantDash 统一支持股票和 ETF 的行情数据其标的代码格式完全标准化如 A 股 ETF 通常使用交易所后缀 .SH 或 .SZ[1]。首先确保安装了 SDKpip install quantdash pandas matplotlib我们可以先拉取单只 ETF 历史 K 线作为测试[1]from quantdash import QuantDash # 初始化客户端会自动读取环境变量 QUANTDASH_API_KEY qd QuantDash(api_keydemo_public_token) # 演示环境可使用公共测试 Token df qd.klines.get( symbol510300.SH, period1d, count300, adjustforward, to_dataframeTrue ) print(df.tail())如果想看全市场 ETF 的实时行情快照可以直接调用[1]etfs qd.quotes.get(universesCN_ETF, to_dataframeTrue) 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_keydemo_public_token) dfs qd.klines.batch( symbolsETF_POOL, period1d, count1000, adjustforward, to_dataframeTrue, show_progressTrue, )我们将获取到的各标的数据中的收盘价整理并对齐到同一个时间矩阵中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, axis1).sort_index() close close.dropna(howall) return close close build_close_matrix(dfs) print(close.tail())这个 close 价格矩阵就是我们接下来生成信号与执行回测的坚实底座。4. 定义轮动信号我们使用一个极其直观但逻辑闭环的动量轮动规则每天计算每个 ETF 过去 N 个交易日本例采用 60 日的累计收益率作为动量指标。在所有 ETF 中选择动量指标最高的那只。安全垫/空仓机制如果所有 ETF 的 60 日动量收益率均小于 0说明全市场处于系统性下跌期组合选择空仓。【规避未来函数】今天生成的交易决策明天才持有严格作 shift(1) 延迟处理避免未来函数。# 设定动量观察周期为 60 个交易日 lookback 60 # 计算累计收益率矩阵 momentum close / close.shift(lookback) - 1 # 找到每天动量值最大的 ETF 标的及其对应的动量值 selected momentum.idxmax(axis1) best_momentum momentum.max(axis1) # 初始化一个与价格矩阵同形状的信号矩阵 signal pd.DataFrame(0, indexclose.index, columnsclose.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(axis1) # 4. 引入交易摩擦设定费率及滑点 fee 0.0003 # 单边佣金/印花税 slippage 0.0002 # 单边交易滑点 # 计算每日调仓换手率持仓权重变化的绝对值之和 turnover position.diff().abs().sum(axis1).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(titleETF Rotation Equity Curve (After Cost)) plt.grid(True) plt.tight_layout() plt.savefig(etf_rotation_equity.png, dpi160) plt.show()7. 穿透分析输出每天持有了什么只看单一的累计收益率曲线是不够的回测系统不能做成黑箱。我们输出最近 20 天的持仓详情透视策略的运作轨迹# 找到每日持仓列名称 holding position.idxmax(axis1) # 判断当天组合是否有持仓如果全为0表示空仓避险 has_position position.sum(axis1) 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(axis1) best_momentum momentum.max(axis1) # 持仓矩阵 signal pd.DataFrame(0, indexclose.index, columnsclose.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(axis1) turnover position.diff().abs().sum(axis1).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, ascendingFalse) print(\n 参数扫描结果 (按夏普比率降序) ) print(result_df.to_string(indexFalse))在量化开发中不要一味去追求并采用夏普比率最高的那一个特定最优参数过度拟合风险。相反更理性的选择是寻找一个宽广的参数平台比如 40 日至 80 日周期均表现出相近且不错的绩效这样的参数在未来实盘中才具有更强的抗噪能力。9. Cursor / DeepSeek 专属 AI 编程 Prompt为了方便你围绕这套 ETF 轮动回测框架进行个性化升级例如修改选股逻辑、引入动量权重分配、引入止损线你可以直接把这段 Prompt 复制给 Cursor、Claude 3.5 Sonnet 或 DeepSeek你是一位精通 Python 的量化开发专家。我正在使用 QuantDash 作为 ETF 数据源设计并迭代我的【ETF多资产动量轮动回测系统】。 请记住我的基础回测组件代码框架 1. 批量获取数据并对齐矩阵 dfs qd.klines.batch(symbolsETF_POOL, period1d, count1000, adjustforward, to_dataframeTrue) 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, axis1).sort_index().dropna(howall) 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 轮动是普通人学习量化的好入口。它足够简单能让你快速跑出结果又足够真实能覆盖数据对齐、持仓矩阵、交易成本、参数扫描这些关键问题。对于新手来说在这个完整的流里你至少实践了以下八个关键环节选股池界定不同资产低相关性组合多标的数据批量抓取基于 QuantDash 接口的高速 Batch时序对齐Pandas 跨标的时间轴标准清洗决策信号生成动量矩阵计算规避未来函数严格执行 shift(1) 延迟成交组合日收益累计非事件驱动的向量化极速回测真实摩擦磨损建模佣金与滑点扣除评价体系构建夏普比率与回撤测算QuantDash在这个过程里承担的是数据底座的角色。你用统一的 Python SDK 获取 ETF K 线和实时行情把精力放在策略假设和回测逻辑上而不是反复修数据接口。先写一个简单版本再一点点加规则量化研究就是这样长出来的。“三步走”落地指引第一步获取完整源码。请访问官方开源托管仓库获取本文完整 ETF 轮动代码及更多跨市场高频套利回测 Demohttps://github.com/quantdash-net/QuantDash请认准官方 quantdash-net 组织欢迎 Star 支持。第二步申请专属密钥。注册获取您的个人免费/生产级 API Keyhttps://quantdash.net/。第三步查阅开发细节。更多高频行情、多市场 K 线接口及实时 universes 全量数据集参数请参考https://docs.quantdash.net/。免责声明: 本文仅用于量化技术教学和策略回测方法演示目的不构成任何投资建议。市场有风险投资需谨慎。

相关新闻