在量化交易的实践中，很多开发者往往沉迷于复杂的机器学习模型或高频策略，却忽略了最朴素也最有效的逻辑：资产间的轮动效应。 本文详细介绍了如何从零开始，使用 Python 构建一套完整的 ETF 动量轮动交易系统。文章将带你一步步完成数据获取、收益率与相关性计算、轮动策略逻辑编写、回测验证及实盘注意事项等全流程。核心内容包括：利用 pytdx 获取历史行情数据，通过矩阵运算计算资产间相关性并可视化，编写基于“强者恒强”逻辑的轮动选股代码，以及进行包含交易成本的模拟回测。本文旨在为量化新手和希望优化策略的开发者提供一套清晰、可复现的工程化解决方案，帮助读者在控制风险的同时，捕捉市场板块轮动带来的超额收益。
特别是在 ETF 投资领域，不同板块、不同风格的产品在不同市场周期下表现差异巨大。如果我们能捕捉到这种强弱转换的规律，通过定期持有表现最强的品种，就能在控制回撤的同时获得不错的超额收益。这就是“动量轮动”策略的核心思想，它不需要预测明天的涨跌，只需要判断当下谁更强。
对于个人投资者而言，实现这一策略的最大门槛并非数学公式，而是如何将想法转化为可执行的代码流程。从数据获取、清洗，到相关性计算、信号生成，再到回测验证，每一个环节都藏着不少坑。比如，如何处理停牌导致的数据缺失？如何避免未来函数？交易成本对高频轮动的影响有多大？这些问题如果不在代码层面解决，策略在实盘中很容易失效。
本文将带你从零开始，用 Python 构建一套完整的 ETF 轮动交易系统。我们会跳过晦涩的理论推导，直接聚焦于工程落地：如何搭建环境、如何高效获取历史行情、如何利用矩阵运算计算相关性、以及如何编写健壮的轮动逻辑。无论你是刚入门量化的小白，还是希望优化现有框架的开发者，这套流程都能为你提供清晰的实操参考。接下来的内容将严格按照策略开发的自然顺序展开，确保你读完就能动手复现。
① 策略核心逻辑与数据准备清单
动量轮动策略的本质是“强者恒强”。其基本逻辑是：在一个固定的候选池（例如沪深 300、中证 500、创业板指等相关 ETF）中，每隔一段时间（如每周或每月）计算各标的在过去 N 天的收益率，选出收益率最高的前 M 只 ETF 进行持有，直到下一个调仓日。为了进一步降低风险，我们通常还会引入相关性分析，避免选出的几只 ETF 高度同质化（例如同时选了三个走势几乎一样的科技类 ETF），从而实现真正的分散配置。
在动手写代码之前，必须准备好以下数据清单，这是策略运行的基石：
② Python 量化环境搭建与库安装
工欲善其事，必先利其器。Python 拥有极其丰富的量化生态，我们只需要几个核心库即可完成全流程。建议使用 conda 或 venv 创建独立的虚拟环境，避免依赖冲突。
你需要安装以下关键库：
- pandas：数据处理的核心，用于时间序列操作和矩阵计算。
- numpy：提供高效的数值运算支持。
- matplotlib 和 seaborn：用于绘制收益曲线和相关性热力图。
- pytdx：一个基于通达信数据接口的 Python 库，用于获取 A 股历史行情（注：此处仅作为技术工具提及，具体数据源需符合合规要求）。
- scikit-learn（可选）：如果需要更复杂的聚类分析来辅助选股。
安装命令非常简单，在终端执行：
pip install pandas numpy matplotlib seaborn pytdx
安装完成后，建议在 Jupyter Notebook 或 VS Code 中进行开发，以便随时查看中间变量的数据形态。
③ 获取 ETF 历史行情数据方法
数据获取是量化策略的第一步，也是最容易出错的环节。我们需要编写一个函数，批量下载候选列表中所有 ETF 的历史日线数据，并统一格式。
在使用开源接口时，要注意以下几点：
以下是一个简化的数据获取示例逻辑：
from pytdx.hq import TdxHq_API
import pandas as pd
from datetime import datetime
def fetch_etf_history(code_list, start_date, end_date):
\”\”\”
使用 pytdx 获取 ETF 历史行情数据。
注意：pytdx 默认返回前复权数据，且需要将代码转换为市场代码格式（如 \’sh510300\’）。
\”\”\”
api = TdxHq_API()
all_data = {
}
# 连接到通达信服务器（默认使用深圳行情服务器）
if api.connect(\’119.147.212.81\’, 7709):
for code in code_list:
try:
# 将代码转换为 pytdx 格式（假设为上海 ETF，如 510300 -> sh510300）
market_code = f\’sh{
code}\’ if code.startswith(\’51\’) else f\’sz{
code}\’
# 获取日线数据
data = api.get_security_bars(9, 1 if market_code.startswi
