聚宽全天候 ETF 轮动策略：如何用月线 MACD 信号进行多资产轮动与篮子下单？

聚宽全天候 ETF 轮动策略：月线 MACD 动态配置与篮子下单实现

全天候资产配置策略旨在通过分散投资于低相关性的资产（如股票、黄金、国债）来抵御不同的经济周期。结合月线 MACD 趋势跟踪信号，可以进一步过滤震荡，捕捉大级别趋势。本文将展示如何在聚宽（JoinQuant）平台实现这一策略，并利用 batch_submit_orders 接口进行高效的篮子下单。

一、 策略核心逻辑

1. 候选资产池：
   • 创业板 ETF (159915.XSHE)
   • 纳指 ETF (513100.XSHG)
   • 红利 ETF (510880.XSHG)
   • 黄金 ETF (518880.XSHG)
   • 国债 ETF (511010.XSHG)
2. 交易信号：每月第一个交易日，获取各 ETF 的月线（1M）历史数据，计算 MACD 指标。若 MACD 金叉（DIF > DEA）且 DIF > 0，则判定为买入/持有信号。
3. 仓位配置：对所有产生买入信号的 ETF 进行等权重配置（例如，若有 2 个 ETF 满足条件，则各占 50% 仓位；若无满足条件的，则全仓持有国债 ETF 或空仓）。
4. 年度清仓：每年最后一个交易日收盘前，强制平仓所有标的，释放资金，以便明年初重新初始化配置。
5. 高效下单：使用聚宽 batch_submit_orders 接口，将所有调仓买卖单打包成一个篮子订单一次性提交，避免逐笔下单带来的资金占用冲突。

二、 聚宽 Python 策略源码

请在聚宽回测环境中创建策略，并选择分钟级或天级回测，开启真实价格模式（set_option('use_real_price', True)）。

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=5), type='fund')
    
    # 3. 定义全局变量
    g.etf_pool = [
        '159915.XSHE', # 创业板 ETF
        '513100.XSHG', # 纳指 ETF
        '510880.XSHG', # 红利 ETF
        '518880.XSHG', # 黄金 ETF
        '511010.XSHG'  # 国债 ETF
    ]
    
    # 4. 定时运行任务
    # 每月第一个交易日 09:45 进行月度调仓
    run_monthly(handle_rotation, monthday=1, time='09:45')
    # 每年最后一个交易日 14:50 进行年度清仓
    run_daily(annual_clear, time='14:50')

def get_macd_signal(security, end_dt):
    """
    获取月线 MACD 信号
    """
    # 获取过去 30 个月的月线收盘价
    bars = get_bars(security, count=30, unit='1M', fields=['date', 'close'], include_now=True, end_dt=end_dt, df=True)
    if len(bars) < 26:
        return False
    
    close = bars['close']
    # 计算 EMA
    ema12 = close.ewm(span=12, adjust=False).mean()
    ema26 = close.ewm(span=26, adjust=False).mean()
    dif = ema12 - ema26
    dea = dif.ewm(span=9, adjust=False).mean()
    
    # 最新一期的指标值
    last_dif = dif.iloc[-1]
    last_dea = dea.iloc[-1]
    
    # 信号判定：DIF > DEA 且 DIF > 0
    if last_dif > last_dea and last_dif > 0:
        return True
    return False

def handle_rotation(context):
    """
    月度轮动调仓逻辑
    """
    # 排除 12 月底清仓后的 1 月初，防止立即无信号买入
    if context.current_dt.month == 1 and context.portfolio.total_value == context.portfolio.available_cash:
        log.info("1月初重新初始化，等待信号...")
        
    log.info("====== 开始月度 ETF 轮动评估 ======")
    buy_list = []
    
    # 1. 评估每个 ETF 的月线 MACD 信号
    for etf in g.etf_pool:
        if get_macd_signal(etf, context.current_dt):
            buy_list.append(etf)
            
    log.info("满足买入信号的 ETF 列表: %s" % buy_list)
    
    # 2. 计算目标权重
    target_weights = {}
    if len(buy_list) > 0:
        weight = 1.0 / len(buy_list)
        for etf in buy_list:
            target_weights[etf] = weight
    else:
        # 若无信号，全仓持有国债 ETF 防御
        target_weights['511010.XSHG'] = 1.0
        log.info("无多头信号，全仓国债 ETF 防御")

    # 3. 构建篮子下单列表
    order_list = []
    total_value = context.portfolio.total_value
    
    # 先处理需要卖出或减仓的标的
    current_positions = context.portfolio.positions
    for etf in current_positions.keys():
        if etf not in target_weights:
            # 不在目标持仓中，全卖
            order_list.append({
                'security': etf,
                'amount': -current_positions[etf].total_amount
            })
            
    # 再处理需要买入或加仓的标的
    for etf, weight in target_weights.items():
        target_val = total_value * weight
        current_val = current_positions[etf].value if etf in current_positions else 0
        diff_val = target_val - current_val
        
        # 获取当前最新价计算股数
        current_data = get_current_data()
        price = current_data[etf].last_price
        
        if price > 0:
            # 转换为 100 的整数倍（场内基金交易规则）
            shares = int(diff_val / price / 100) * 100
            if shares != 0:
                order_list.append({
                    'security': etf,
                    'amount': shares
                })

    # 4. 执行篮子下单
    if len(order_list) > 0:
        log.info("提交篮子订单: %s" % order_list)
        batch_submit_orders(order_list)

def annual_clear(context):
    """
    年度清仓逻辑：在每年最后一个交易日执行
    """
    # 判断是否为当年最后一个交易日
    # 获取本年所有交易日
    trade_days = get_trade_days(start_date=pd.Timestamp(context.current_dt).strftime('%Y-01-01'), 
                                end_date=pd.Timestamp(context.current_dt).strftime('%Y-12-31'))
    
    if context.current_dt.date() == trade_days[-1]:
        log.info("====== 触及年度最后一个交易日，执行强制清仓 ======")
        clear_list = []
        for etf, pos in context.portfolio.positions.items():
            if pos.total_amount > 0:
                clear_list.append({
                    'security': etf,
                    'amount': -pos.total_amount
                })
        if len(clear_list) > 0:
            batch_submit_orders(clear_list)
            log.info("年度清仓订单已提交")

三、 核心 API 函数解析

1. get_bars(..., unit='1M')：
   用于获取月线 K 线数据。通过设置 include_now=True，可以确保当前未结束的自然月数据也能被实时纳入计算，保证信号的及时性。
2. batch_submit_orders(orders)：
   聚宽特有的篮子下单 API。它接受一个包含多个订单字典的 list。相比于循环调用 order()，batch_submit_orders 会在底层进行统一的验资验券，避免了因先买后卖导致的“可用资金不足”报错，非常适合多标的轮动策略。
3. set_option('use_real_price', True)：
   强烈建议开启。在 ETF 轮动中，分红和拆分会影响实际持仓。开启真实价格模式后，系统会自动在除权除息日调整账户的持股数量和现金，使回测结果更贴近实盘。