Python 实现基于 20 日动量的行业 ETF 轮动模型

这是一个基于 QMT 平台的 行业 ETF 动量轮动策略。

策略逻辑

1. 标的池：选取代表不同行业的 ETF（如半导体、医药、证券、新能源、消费等）。
2. 动量因子：计算过去 N 天（例如 20 天）的收益率作为动量指标。
3. 排序与筛选：根据动量因子从高到低排序，选取排名前 M 只（例如前 2 只）ETF。
4. 轮动交易：
   • 卖出：如果持仓的 ETF 不在最新的优选列表中，则清仓。
   • 买入/调仓：对优选列表中的 ETF 进行等权重买入或调仓。
5. 风控/择时（可选）：本示例为纯轮动，不包含大盘择时，旨在展示轮动框架。

策略代码

# -*- coding: gbk -*-
import pandas as pd
import numpy as np

def init(ContextInfo):
    """
    初始化函数，设定策略参数和股票池
    """
    # 1. 设定资金账号 (请替换为您自己的资金账号)
    ContextInfo.accID = '6000000000' 
    ContextInfo.set_account(ContextInfo.accID)
    
    # 2. 设定行业ETF池 (示例代码，涵盖主要行业)
    # 512480: 半导体, 512010: 医药, 512880: 证券, 515030: 新能源车
    # 512690: 酒ETF, 515000: 科技ETF, 512170: 医疗ETF, 510300: 沪深300(作为防守或基准)
    ContextInfo.etf_pool = [
        '512480.SH', '512010.SH', '512880.SH', '515030.SH',
        '512690.SH', '515000.SH', '512170.SH', '510300.SH'
    ]
    ContextInfo.set_universe(ContextInfo.etf_pool)
    
    # 3. 策略参数
    ContextInfo.lookback_days = 20  # 动量回顾周期 (N日收益率)
    ContextInfo.hold_count = 2      # 持仓数量 (持有排名前几的ETF)
    ContextInfo.rebalance_period = 1 # 调仓频率 (每隔多少个bar调仓，1代表每天)
    
    # 4. 费率设置 (回测用)
    # 佣金万三，印花税千一(ETF通常免印花税，此处仅为示例)
    ContextInfo.set_commission(0, [0.0001, 0.0001, 0.0003, 0.0003, 0, 5])
    
    print("策略初始化完成，ETF池数量：", len(ContextInfo.etf_pool))

def handlebar(ContextInfo):
    """
    K线周期运行函数
    """
    # 获取当前K线索引
    index = ContextInfo.barpos
    
    # 确保有足够的历史数据计算动量
    if index < ContextInfo.lookback_days:
        return

    # 控制调仓频率 (例如每隔N天调仓一次)
    # 如果是日线回测，index即代表天数
    # 如果是实盘，通常建议在尾盘或固定时间运行，此处简化为每个bar运行
    if index % ContextInfo.rebalance_period != 0:
        return

    # --- 1. 获取数据 ---
    # 获取过去 lookback_days + 1 天的收盘价，用于计算收益率
    # 使用 get_market_data_ex 接口
    # count = lookback_days + 1 (因为要计算 N 天前的价格作为基数)
    market_data = ContextInfo.get_market_data_ex(
        fields=['close'], 
        stock_code=ContextInfo.etf_pool, 
        period='1d', 
        count=ContextInfo.lookback_days + 1,
        dividend_type='front' # 前复权
    )
    
    # --- 2. 计算动量 ---
    momentum_scores = {}
    
    for code in ContextInfo.etf_pool:
        if code in market_data:
            df = market_data[code]
            # 检查数据长度是否足够
            if len(df) >= ContextInfo.lookback_days + 1:
                # 获取最新收盘价和N天前的收盘价
                current_price = df['close'].iloc[-1]
                old_price = df['close'].iloc[0] # 取第一条数据作为基准
                
                # 计算收益率 (动量)
                if old_price > 0:
                    ret = (current_price - old_price) / old_price
                    momentum_scores[code] = ret
    
    # 如果没有有效数据，直接返回
    if not momentum_scores:
        return

    # --- 3. 排序与筛选 ---
    # 将字典按收益率从大到小排序
    sorted_momentum = sorted(momentum_scores.items(), key=lambda x: x[1], reverse=True)
    
    # 选取排名前 hold_count 的 ETF 代码
    target_list = [item[0] for item in sorted_momentum[:ContextInfo.hold_count]]
    
    # 打印当天的排名情况 (调试用)
    # date_str = timetag_to_datetime(ContextInfo.get_bar_timetag(index), '%Y-%m-%d')
    # print(f"{date_str} 动量排名: {sorted_momentum[:3]}... 目标持仓: {target_list}")

    # --- 4. 交易执行 ---
    
    # 获取当前持仓
    # 注意：回测模式下，get_trade_detail_data 返回的是回测账户状态
    # 实盘模式下，返回的是真实账户状态
    positions = get_trade_detail_data(ContextInfo.accID, 'stock', 'position')
    current_holdings = [obj.m_strInstrumentID + '.' + obj.m_strExchangeID for obj in positions if obj.m_nVolume > 0]
    
    # A. 卖出逻辑：不在目标列表中的持仓，全部卖出
    for code in current_holdings:
        if code not in target_list:
            # 目标价值设为0，即清仓
            order_target_value(code, 0, ContextInfo, ContextInfo.accID)
            # print(f"卖出: {code}")

    # B. 买入逻辑：在目标列表中的，调整至目标仓位
    # 计算每只ETF的目标市值 (总资产 / 持仓数量)
    # 获取账户总资产
    account_info = get_trade_detail_data(ContextInfo.accID, 'stock', 'account')
    if len(account_info) > 0:
        total_asset = account_info[0].m_dBalance
        
        # 预留少量现金防止手续费不足，乘以 0.98
        target_value_per_etf = (total_asset * 0.98) / ContextInfo.hold_count
        
        for code in target_list:
            # 使用 order_target_value 自动计算买卖数量达到目标金额
            # 23代表买入(股票)，1101代表单股单账号普通下单(此处order_target_value封装了passorder)
            # 注意：order_target_value 在回测中非常方便，实盘中需谨慎使用，因为它会市价下单
            order_target_value(code, target_value_per_etf, ContextInfo, ContextInfo.accID)
            # print(f"买入/调仓: {code}, 目标金额: {target_value_per_etf}")

代码关键点解析

1. # -*- coding: gbk -*-:

   • 这是 QMT Python 策略文件的必须声明，否则中文注释会导致编译错误。

2. 数据获取 (get_market_data_ex):

   • 我们使用 get_market_data_ex 获取多只股票的历史数据。
   • count=ContextInfo.lookback_days + 1：为了计算 20 日涨幅，我们需要获取当前日期往前推 21 个数据点（第 1 个点是基准，第 21 个点是当前价）。
   • dividend_type='front'：使用前复权数据，保证收益率计算不受分红除权影响。

3. 动量计算:

   • 逻辑简单直接：(当前收盘价 - N天前收盘价) / N天前收盘价。
   • 使用了 sorted 函数对字典进行降序排列。

4. 交易执行 (order_target_value):

   • 这是一个非常便捷的高级交易函数。它会自动判断当前持仓和目标市值的差额，自动发出买入或卖出指令。
   • 例如：目标持有 5 万元，当前持有 3 万元，它会自动买入 2 万元；如果当前持有 0，则买入 5 万元。
   • 资金分配：代码中将总资产乘以 0.98，预留 2% 的资金用于支付手续费和防止因价格波动导致的废单。

5. 实盘与回测:

   • 该代码既可用于回测，也可用于实盘（需将 ContextInfo.accID 替换为真实账号，并将运行模式切换为实盘）。
   • 注意：实盘中使用 order_target_value 默认通常以市价或对手价下单，请确保 ETF 流动性充足。

常见问题 Q&A

Q: 如何修改动量周期？
A: 修改 init 函数中的 ContextInfo.lookback_days = 20，例如改为 60 即为 60 日动量。

Q: 为什么使用 get_market_data_ex 而不是 get_market_data？
A: get_market_data_ex 是 QMT 推荐的新版接口，返回的数据结构（Pandas DataFrame/Panel）更规范，且支持订阅和历史数据的一体化获取，处理多股数据更方便。

Q: 策略在盘中会频繁交易吗？
A: 代码中设置了 if index % ContextInfo.rebalance_period != 0: return。如果 rebalance_period 设为 1 且运行周期为日线，则每天收盘（回测）或每天盘中（实盘，取决于运行频率设置）检查一次。如果在实盘中想固定在每天 14:50 交易，可以在 handlebar 中增加时间判断逻辑：

import time
# 获取当前tick时间
timetag = ContextInfo.get_tick_timetag()
now_time = timetag_to_datetime(timetag, '%H%M%S')
# 仅在 14:50 到 14:55 之间运行逻辑
if not ('145000' <= now_time <= '145500'):
    return