如何在聚宽（JoinQuant）中编写纯 ETF 双轮动策略？（附 Python 源码）

聚宽双轮动 ETF 策略设计与实现

双轮动 ETF 策略的核心逻辑在于：在给定的 ETF 候选池中，通过计算过去一段时间的**动量（Momentum）或相对强弱（Relative Strength）**指标，筛选出表现最强势的若干只 ETF 进行等权重或比例配置；同时，当标的动量转弱或跌破均线时，及时切换到其他强势标的或空仓避险。

以下是在聚宽（JoinQuant）平台实现该策略的完整设计方案与 Python 源码。

一、 策略核心设计

1. 标的选择（ETF 候选池）：
   选择相关性较低、代表不同大类资产的 ETF。例如：

   • 权益类：沪深300 ETF (510300.XSHG)、创业板 ETF (159915.XSHE)
   • 跨境/商品类：纳指 ETF (513100.XSHG)、黄金 ETF (518880.XSHG)

2. 动量指标计算：
   使用过去 $N$ 个交易日的累积收益率作为动量因子。为避免未来函数，在开盘前或盘中调仓时，必须使用历史数据（如 attribute_history 或 get_bars）计算，且不包含当前未收盘的 Bar。

3. 调仓与仓位分配：

   • 调仓频率：按月或按周轮动（使用 run_monthly 或 run_weekly），降低交易频次和摩擦成本。
   • 仓位分配：对筛选出的前 $K$ 只强势 ETF 进行等比例分配（如双轮动即选择前 2 只，各配 50% 仓位）。
   • 避险机制：若候选 ETF 的动量收益率低于 0（绝对动量过滤），则不买入，对应仓位留存现金或买入货币基金。

二、 聚宽策略源码实现

请在聚宽策略编辑框中粘贴以下 Python 3 代码进行回测：

import pandas as pd
import numpy as np
from jqdata import *

def initialize(context):
    # 1. 设定基准与基本设置
    set_benchmark('000300.XSHG')
    set_option('use_real_price', True) # 开启真实价格模式
    log.set_level('order', 'info')
    
    # 2. 设定交易税费（场内基金不收印花税，佣金一般无最低免五限制，此处按万三设置）
    set_order_cost(OrderCost(close_tax=0, open_commission=0.0003, close_commission=0.0003, min_commission=0), type='fund')
    
    # 3. 策略参数配置
    g.etf_pool = [
        '510300.XSHG', # 沪深300 ETF
        '159915.XSHE', # 创业板 ETF
        '513100.XSHG', # 纳指 ETF
        '518880.XSHG'  # 黄金 ETF
    ]
    g.momentum_window = 20 # 动量计算窗口（20个交易日）
    g.keep_count = 2       # 双轮动：每次持有前2只强势ETF
    
    # 4. 定时运行设置：每月第一个交易日 09:30 运行调仓
    run_monthly(handle_rotation, monthday=1, time='09:30', reference_security='000300.XSHG')

def handle_rotation(context):
    log.info("====== 开始执行月度轮动调仓 ======")
    
    # 1. 获取历史数据计算动量
    momentum_scores = {}
    for etf in g.etf_pool:
        # 获取过去 N+1 天的收盘价，不包含今天（避免未来函数）
        hist = attribute_history(etf, g.momentum_window + 1, '1d', ['close'], skip_paused=True)
        if len(hist) < g.momentum_window + 1:
            continue
        
        # 计算动量：(昨日收盘价 - N天前收盘价) / N天前收盘价
        close_prev = hist['close'].iloc[0]
        close_last = hist['close'].iloc[-1]
        momentum = (close_last - close_prev) / close_prev
        momentum_scores[etf] = momentum
        
    # 2. 按动量得分从大到小排序
    sorted_etfs = sorted(momentum_scores.items(), key=lambda x: x[1], reverse=True)
    log.info(f"当前各ETF动量得分: {sorted_etfs}")
    
    # 3. 筛选出动量大于0的前 K 只 ETF
    target_list = []
    for etf, score in sorted_etfs[:g.keep_count]:
        if score > 0: # 绝对动量过滤，只有收益为正才买入
            target_list.append(etf)
            
    log.info(f"本期目标持仓列表: {target_list}")
    
    # 4. 卖出不在目标列表中的持仓
    current_positions = list(context.portfolio.long_positions.keys())
    for etf in current_positions:
        if etf not in target_list:
            order_target(etf, 0)
            log.info(f"卖出不符合轮动条件的ETF: {etf}")
            
    # 5. 重新分配仓位并买入
    if len(target_list) > 0:
        # 计算单只 ETF 分配的资金（等权重分配）
        total_value = context.portfolio.total_value
        target_value_per_etf = total_value / g.keep_count
        
        for etf in target_list:
            # 调整至目标价值
            order_target_value(etf, target_value_per_etf)
            log.info(f"调仓/买入 ETF: {etf}，目标价值: {target_value_per_etf:.2f}元")
    else:
        log.info("无符合绝对动量要求的标的，本期选择空仓避险。")

三、 关键API与避险机制解析

1. attribute_history 规避未来数据：
   在 handle_rotation 中，我们获取了 g.momentum_window + 1 长度的数据。由于是在 09:30 运行，昨日的收盘价是 hist['close'].iloc[-1]，而 $N$ 天前的收盘价是 hist['close'].iloc[0]。这样计算出的动量完全基于历史已实现价格，不包含调仓当天的任何未来信息。

2. order_target_value 自动仓位再平衡：
   使用 order_target_value(security, value) 可以非常方便地将账户资金等比例分配。它会自动计算当前持仓价值与目标价值的差额，自动执行买入或卖出，无需手动计算交易股数，极大地简化了多标的再平衡的代码逻辑。