摘要： 本文是ETF轮动量化策略系列教程的第三篇，正式进入Python实战阶段。我们将以最经典的20日相对动量轮动策略为例，手把手带你完成从环境搭建、通过 pytdx 从通达信服务器免费获取数据、数据清洗、动量计算到持仓信号生成的完整流程。文章重点强调如何用 shift() 函数规避回测中最致命的“未来函数”陷阱，并最终生成一份可直接用于回测的每日持仓表。所有代码均提供详细注释，复制即可运行。

---

大家好，我是你们的老朋友。前两篇我们解决了“为什么轮动”以及“轮动的核心逻辑是什么”。今天，终于到了最激动人心的环节——打开编辑器，亲手写代码。

我们的目标很明确：实现一个最经典的20日相对动量轮动策略，并生成一份“无未来函数”的每日持仓信号表。这篇内容会非常“干”，建议你边看边敲，把每一行代码都跑通。

一、策略规则回顾

在动手之前，我们先明确今天要实现的策略规则：

• 候选池：沪深300ETF（510300）、中证500ETF（510500）、创业板ETF（159915）、国债ETF（511010）
• 评判指标：过去20个交易日的累计涨跌幅（20日动量）
• 调仓频率：每周五收盘后计算，下周一开盘执行
• 选股规则：全仓买入20日动量值最高的那只ETF
• 风控规则：如果动量值最高的那个也是负数，则空仓（持有货币ETF或现金）

二、环境准备与数据获取

2.1 安装依赖库

首先，我们需要安装几个核心库。打开终端，逐行执行：

pip install pandas numpy matplotlib pytdx

• pandas：数据处理的核心，我们用它来操纵表格数据
• numpy：数值计算
• matplotlib：画图（下一篇回测时用）
• pytdx：开源的通达信数据接口，可以从券商行情服务器直接获取历史K线数据，免费且无需注册

2.2 导入库并设置显示参数

    import pandas as pd
    import numpy as np
    import matplotlib.pyplot as plt
    from pytdx.hq import TdxHq
    from datetime import datetime, timedelta

    # 设置pandas显示选项，让表格更易读
    pd.set_option('display.max_columns', 10)
    pd.set_option('display.width', 1000)
    pd.set_option('display.max_rows', 50)

    # 设置matplotlib支持中文显示
    plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS', 'DejaVu Sans']
    plt.rcParams['axes.unicode_minus'] = False

    print("所有库导入成功！")

2.3 通过 pytdx 获取ETF历史行情数据

pytdx 通过连接通达信的行情服务器获取数据。我们需要先了解几个关键概念：

• 市场代码：上海交易所为 1，深圳交易所为 0
• ETF代码：如 510300（上海）、159915（深圳）
• K线类型：日线为 9，周线为 5，月线为 6

    # 候选池定义：ETF代码 -> (市场代码, 简称)
    # 市场代码：1=上海，0=深圳
    etf_dict = {
        '510300': (1, '沪深300'),
        '510500': (1, '中证500'),
        '159915': (0, '创业板'),
        '511010': (1, '国债ETF')
    }

    # 初始化pytdx连接
    # 使用通达信的免费行情服务器，如果连接失败可以换其他IP
    tdx_host = '119.147.212.81'  # 通达信行情服务器主IP
    tdx_port = 7709

    try:
        hq = TdxHq()
        hq.connect(tdx_host, tdx_port)
        print(f'成功连接到通达信服务器 {tdx_host}:{tdx_port}')
    except Exception as e:
        print(f'连接失败，请检查网络或更换服务器IP: {e}')
        # 备用服务器列表
        backup_hosts = [
            '120.76.152.177',
            '124.71.224.34',
            '120.79.60.82'
        ]
        for host in backup_hosts:
            try:
                hq = TdxHq()
                hq.connect(host, tdx_port)
                print(f'成功连接到备用服务器 {host}:{tdx_port}')
                break
            except:
                continue
        else:
            raise ConnectionError('所有通达信服务器均连接失败，请检查网络后重试')

2.4 分批获取历史数据函数

通达信服务器单次请求通常最多返回800条记录，我们需要分批获取较长的历史数据。

    def get_etf_daily_data(hq, code, market, start_date='20180101', end_date='20251231'):
        """
        从通达信服务器获取ETF日线数据，支持分批获取长周期数据
        
        Parameters:
        - hq: pytdx连接对象
        - code: ETF代码（字符串）
        - market: 市场代码（1=上海，0=深圳）
        - start_date: 起始日期，格式YYYYMMDD
        - end_date: 结束日期，格式YYYYMMDD
        
        Returns:
        - DataFrame，包含日期和收盘价
        """
        start = datetime.strptime(start_date, '%Y%m%d')
        end = datetime.strptime(end_date, '%Y%m%d')
        
        all_data = []
        batch_size = 600  # 每批获取约600天数据，留有余量
        
        current_start = start
        
        while current_start < end:
            current_end = min(current_start + timedelta(days=batch_size), end)
            
            try:
                data = hq.get_security_bars(
                    market=market,
                    code=code,
                    start=current_start.strftime('%Y%m%d'),
                    end=current_end.strftime('%Y%m%d'),
                    k_type=9,      # 日线
                    fq=1           # 前复权
                )
                
                if data and len(data) > 0:
                    all_data.extend(data)
                    print(f'  获取 {code} 从 {current_start.strftime("%Y-%m-%d")} 到 {current_end.strftime("%Y-%m-%d")} 的数据，共 {len(data)} 条')
                
            except Exception as e:
                print(f'  获取 {code} 数据时出错：{e}，跳过此段')
            
            current_start = current_end + timedelta(days=1)
        
        if len(all_data) == 0:
            print(f'  警告：未能获取到 {code} 的任何数据')
            return pd.DataFrame()
        
        # 转换为DataFrame
        df = pd.DataFrame(all_data)
        # pytdx返回的字段包含：open, high, low, close, volume, amount, year, month, day...
        df['date'] = pd.to_datetime(df[['year', 'month', 'day']])
        df = df[['date', 'close']].copy()
        df.set_index('date', inplace=True)
        df.sort_index(inplace=True)
        
        return df

2.5 下载所有ETF数据并合并

    # 存放每只ETF数据的字典
    etf_data = {}

    # 逐个获取历史数据
    for code, (market, name) in etf_dict.items():
        print(f'\n正在获取 {name}({code}) 的数据...')
        
        df = get_etf_daily_data(
            hq=hq,
            code=code,
            market=market,
            start_date='20180101',
            end_date='20251231'
        )
        
        if len(df) > 0:
            # 重命名收盘价列，以ETF代码作为列名
            df.rename(columns={'close': code}, inplace=True)
            etf_data[code] = df
            print(f'  {name} 数据获取完成，共 {len(df)} 条记录')
        else:
            print(f'  {name} 数据获取失败！')

    # 使用完毕后关闭连接
    hq.disconnect()
    print('\n已断开与通达信服务器的连接')
    print('所有ETF数据获取完毕！')

    # 将所有ETF的收盘价合并成一张大表
    price_df = pd.DataFrame()
    for code in etf_dict.keys():
        if code in etf_data:
            price_df = pd.concat([price_df, etf_data[code]], axis=1)

    # 删除所有ETF都没有数据的日期
    price_df.dropna(how='all', inplace=True)
    # 对缺失值做前向填充处理（停牌日期沿用前一日价格）
    price_df.ffill(inplace=True)
    # 删除开头无法填充的NaN行
    price_df.dropna(inplace=True)

    print(f'\n合并后的数据形状：{price_df.shape}')
    print(f'数据起始日期：{price_df.index[0].strftime("%Y-%m-%d")}')
    print(f'数据截止日期：{price_df.index[-1].strftime("%Y-%m-%d")}')
    print('\n最近5个交易日的收盘价：')
    print(price_df.tail())

运行后，你应该能看到一张类似这样的表格：

            510300  510500  159915  511010
2024-12-25   4.050   7.350   2.950  103.200
2024-12-26   4.055   7.380   2.970  103.210
2024-12-27   4.060   7.400   2.990  103.180
2024-12-30   4.045   7.320   2.950  103.220
2024-12-31   4.050   7.300   2.920  103.250

这张表就是我们后续所有计算的原材料。

小提示：由于通达信服务器对历史数据的支持有限，如果你发现获取的数据最早只到2019年左右，这是正常的。对于策略回测来说，4-5年的数据已经足够。

三、计算20日动量并生成信号

3.1 动量计算

20日动量 = (今日收盘价 - 20个交易日前收盘价) / 20个交易日前收盘价

在pandas中，pct_change(periods=20) 一步到位：

    # 计算20日动量（百分比收益率）
    momentum_20 = price_df.pct_change(periods=20)

    # 前20天没有足够数据计算动量，删除这些行
    momentum_20.dropna(inplace=True)

    print('20日动量计算结果（最近5个交易日）：')
    print(momentum_20.tail())

输出示例：

            510300    510500    159915    511010
2024-12-25  0.0150    0.0220    0.0350   -0.0010
2024-12-26  0.0135    0.0185    0.0320   -0.0008
2024-12-27  0.0120    0.0150    0.0280   -0.0012
2024-12-30  0.0080    0.0080    0.0150   -0.0005
2024-12-31  0.0050    0.0020    0.0050   -0.0003

正值表示过去20天上涨，负值表示下跌。可以看到这几天创业板（159915）的动量最强。

3.2 生成每日最优ETF信号

    # 找出每行（每个交易日）动量值最大的列名（ETF代码）
    best_etf = momentum_20.idxmax(axis=1)
    # 同时找出每行的最大动量值
    max_val = momentum_20.max(axis=1)

    # 构建信号DataFrame
    signal_df = pd.DataFrame({
        'best_etf': best_etf,
        'max_momentum': max_val
    })

    print('每日最优ETF信号（最近10个交易日）：')
    print(signal_df.tail(10))

四、关键步骤：规避“未来函数”与加入风控

4.1 什么是未来函数（Look-ahead Bias）？

这是一个致命错误，新手几乎必犯。

错误做法：在周五收盘后，用周五的收盘价计算动量，然后认为自己在周五当天就以收盘价买入了这只ETF。

为什么错：周五的收盘价在收盘那一刻才确定，你不可能在收盘前就知道收盘价并以此成交。这等于你“穿越”了，回测收益会被严重高估。

正确做法：周五收盘后计算信号 → 下周一开盘时执行交易。信号与执行之间必须有至少一个交易日的延迟。

4.2 用 shift() 实现信号滞后

    # 核心操作：将信号向后移动一天
    # 今天生成的信号，明天才能执行
    signal_df['best_etf_shifted'] = signal_df['best_etf'].shift(1)
    signal_df['max_momentum_shifted'] = signal_df['max_momentum'].shift(1)

    # 对比移动前后的信号
    comparison = signal_df[['best_etf', 'best_etf_shifted']].tail(10).copy()
    comparison.columns = ['当日计算出的最优ETF', '次日实际持仓的ETF']
    print('\n信号shift前后对比（注意日期错位）：')
    print(comparison)

你会看到，“当日计算出的最优ETF”那一列，比“次日实际持仓”早了一个交易日。这就对了。

4.3 加入动量过滤（风控）

如果计算出的最优动量值都是负数，我们选择空仓：

    # 获取shift后的信号
    final_signal = signal_df['best_etf_shifted'].copy()

    # 风控逻辑：如果shift后的最大动量值 < 0，则空仓
    final_signal[signal_df['max_momentum_shifted'] < 0] = 'cash'

    # 删除前几行NaN值
    final_signal.dropna(inplace=True)

    print('\n最终持仓信号（加入风控后，最近20个交易日）：')
    print(final_signal.tail(20))

五、实现周频调仓

5.1 为什么要控制调仓频率？

如果不加控制，我们上面的信号是“每日轮动”——每天检查一次动量，每天可能换仓。这会导致：

• 交易成本极高
• 在震荡市中频繁“左右打脸”
• 策略实盘可行性差

所以我们要改成每周调仓一次。

5.2 提取每周的调仓信号

    # 给信号DataFrame添加“星期几”的标记（0=周一, 4=周五）
    signal_df['weekday'] = signal_df.index.dayofweek

    # 选择每周五作为信号计算日
    friday_signals = signal_df[signal_df['weekday'] == 4].copy()

    # 对这些周五信号应用shift，让它们在下周一生效
    friday_signals['trade_signal'] = friday_signals['best_etf'].shift(1)
    friday_signals['max_shifted'] = friday_signals['max_momentum'].shift(1)

    # 应用风控过滤
    friday_signals.loc[friday_signals['max_shifted'] < 0, 'trade_signal'] = 'cash'

    print('\n每周调仓信号（周五生成，下周一执行）：')
    print(friday_signals[['best_etf', 'max_momentum', 'trade_signal']].tail(20))

5.3 将周度信号填充到每日

为了后续计算每日净值，我们需要把周度信号扩展到每一天：

    # 将周度信号重新索引到原始的每日日期范围
    daily_signal = friday_signals['trade_signal'].reindex(price_df.index)

    # 用前向填充法（ffill），让本周一至下周四都沿用上周五生成的信号
    daily_signal.ffill(inplace=True)

    # 在第一个信号出现之前，默认空仓
    daily_signal.fillna('cash', inplace=True)

    print('\n每日持仓信号（前向填充后，最近20个交易日）：')
    print(daily_signal.tail(20))

六、持仓信号可视化

在进入回测之前，我们先直观感受一下策略让我们在不同时间段持有什么。

    def visualize_signal(daily_signal):
        """
        可视化持仓信号的分布
        """
        # 统计各标的持有天数占比
        signal_counts = daily_signal.value_counts()
        
        print('\n=== 持仓分布统计 ===')
        for etf, count in signal_counts.items():
            pct = count / len(daily_signal) * 100
            # 获取ETF中文名称
            etf_name = '现金'
            for code, (_, name) in etf_dict.items():
                if etf == code:
                    etf_name = name
                    break
            print(f'{etf_name}({etf}): {count} 天 ({pct:.1f}%)')
        
        # 绘制信号分布图
        fig, ax = plt.subplots(figsize=(14, 4))
        
        # 将持仓标的映射为数字，方便画图
        all_assets = list(etf_dict.keys()) + ['cash']
        asset_map = {asset: i for i, asset in enumerate(all_assets)}
        numeric_signal = daily_signal.map(asset_map)
        
        ax.plot(numeric_signal.index, numeric_signal.values, linewidth=0.8, color='steelblue')
        ax.set_yticks(range(len(all_assets)))
        ax.set_yticklabels(all_assets)
        ax.set_title('20日动量轮动策略 - 每日持仓信号', fontsize=14)
        ax.set_xlabel('日期')
        ax.set_ylabel('持仓标的')
        ax.grid(True, alpha=0.3)
        
        plt.tight_layout()
        plt.show()

    # 执行可视化
    visualize_signal(daily_signal)

七、保存信号数据，为回测做准备

最后，我们把计算好的每日持仓信号保存下来，下一篇回测时直接使用。

    # 构建最终输出表：每日价格 + 持仓信号
    final_df = price_df.copy()
    final_df['hold'] = daily_signal

    # 保存到本地CSV文件
    final_df.to_csv('etf_rotation_signal.csv', encoding='utf-8-sig')
    print('\n信号数据已保存至 etf_rotation_signal.csv')

    # 展示最终表格的前后10行
    print('\n=== 最终输出表格（前10行）===')
    print(final_df.head(10))
    print('\n=== 最终输出表格（后10行）===')
    print(final_df.tail(10))

八、本篇总结

在本文中，我们完成了以下工作：

步骤	内容	关键技术点
环境搭建	安装pandas, numpy, matplotlib, pytdx	pip install
数据获取	用pytdx连接通达信服务器，分批下载4只ETF的历史日线数据	TdxHq(), get_security_bars()
数据清洗	合并表格、对齐日期、处理缺失值	pd.concat(), ffill()
动量计算	计算每只ETF的20日涨跌幅	pct_change(periods=20)
信号生成	每日选出动量最强的ETF	idxmax(axis=1)
避坑	用shift(1)消除未来函数	回测中最重要的一步！
风控	动量全负时空仓	max_momentum < 0 → cash
调仓频率	从日频改为周频	筛选周五信号 + ffill()
输出	保存每日持仓信号表	to_csv()

下一篇预告：策略回测与业绩评估

有了这份无未来函数的持仓信号，下一篇文章我们将：

• 根据信号计算策略的每日收益率
• 绘制策略净值曲线
• 计算年化收益、最大回撤、夏普比率等核心指标
• 与“等权买入持有”基准进行对比，看看这个简单策略到底有没有创造超额收益！

---

完整代码获取：建议你把本文的所有代码块按顺序拼成 .py 文件或 Jupyter Notebook，一次性跑通。如果遇到通达信服务器连接不上的问题，可以尝试：

1. 更换脚本中提供的备用服务器IP
2. 检查防火墙是否阻止了7709端口
3. 在评论区留言，我会协助排查

咱们下篇见！