聚宽 ETF 轮动策略：如何用月线 MACD 构建多资产择时？（附 Python 源码）

策略逻辑与设计思路

本策略旨在通过月线级别的 MACD 指标对四大核心 ETF 资产（创业板 ETF、纳指 ETF、黄金 ETF、红利 ETF）进行趋势择时，并在每个月的第一个交易日进行轮动调仓。其核心逻辑如下：

1. 资产池构建：
   • 创业板 ETF (159915.XSHE)
   • 纳指 ETF (513100.XSHG)
   • 黄金 ETF (518880.XSHG)
   • 红利 ETF (510880.XSHG)
2. 月线 MACD 计算：
   • 在每月第一个交易日开盘前，获取各 ETF 过去 120 个交易日的日线收盘价，并将其重采样或合成为月线数据（或直接使用 get_bars 获取月线 Bar）。
   • 计算月线 MACD 的 DIF（快线 - 慢线）和 DEA（信号线）。
3. 轮动与择时规则：
   • 多头筛选：只考虑月线 MACD 处于多头状态（即 DIF > DEA 且 MACD 柱 > 0）的资产。
   • 动量轮动：在满足多头状态的资产中，选择 MACD 柱状图（HIST）绝对值或增长率最强的资产进行全仓买入。
   • 空仓避险：若所有资产的月线 MACD 均不满足多头条件，则全仓持有现金或买入国债/货币基金避险。

聚宽 Python 策略源码

请在聚宽回测环境中运行以下完整代码。建议回测频率选择分钟或天，并开启真实价格（动态复权）模式。

import pandas as pd
import numpy as np
from jqlib.technical_analysis import *
from jqdata import *

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 过滤掉比error级别低的log
    log.set_level('order', 'error')
    
    # 设定交易税费：买入万三，卖出万三加千一印花税
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 定义轮动资产池
    g.etf_pool = {
        '159915.XSHE': '创业板ETF',
        '513100.XSHG': '纳指ETF',
        '518880.XSHG': '黄金ETF',
        '510880.XSHG': '红利ETF'
    }
    
    # 每月第一个交易日，开盘时运行轮动函数
    run_monthly(handle_rotation, monthday=1, time='09:30', reference_security='000300.XSHG')

def handle_rotation(context):
    log.info("====== 开始执行月度 ETF 轮动调仓 ======")
    
    selected_asset = None
    max_macd_hist = -99999
    
    # 遍历资产池，计算月线 MACD
    for etf, name in g.etf_pool.items():
        # 获取最近 12 个月的月线 Bar 数据
        bars = get_bars(etf, count=12, unit='1M', fields=['date', 'close'], include_now=False, df=True)
        if len(bars) < 12:
            continue
            
        close_prices = bars['close'].values
        
        # 手动计算 MACD (12, 26, 9)
        dif, dea, macd_hist = calculate_macd(close_prices)
        
        log.info(f"{name}({etf}) -> 月线 DIF: {dif:.4f}, DEA: {dea:.4f}, HIST: {macd_hist:.4f}")
        
        # 择时条件：DIF > DEA 且 HIST > 0 (处于上升趋势)
        if dif > dea and macd_hist > 0:
            # 轮动条件：选择 MACD 柱状图最强的资产
            if macd_hist > max_macd_hist:
                max_macd_hist = macd_hist
                selected_asset = etf

    # 获取当前持仓
    current_positions = list(context.portfolio.positions.keys())
    
    if selected_asset:
        log.info(f"本月选出最强多头资产: {g.etf_pool[selected_asset]}({selected_asset})")
        # 卖出不在选择列表中的资产
        for asset in current_positions:
            if asset != selected_asset:
                log.info(f"卖出非目标资产: {g.etf_pool[asset]}")
                order_target(asset, 0)
        
        # 买入目标资产
        cash = context.portfolio.available_cash
        if cash > 1000:
            log.info(f"全仓买入目标资产: {g.etf_pool[selected_asset]}")
            order_value(selected_asset, cash)
    else:
        log.info("未有资产满足多头择时条件，全仓空仓避险。")
        for asset in current_positions:
            order_target(asset, 0)

def calculate_macd(prices, fastperiod=12, slowperiod=26, signalperiod=9):
    """手动计算 MACD 指标"""
    df = pd.DataFrame(prices, columns=['close'])
    df['ema_fast'] = df['close'].ewm(span=fastperiod, adjust=False).mean()
    df['ema_slow'] = df['close'].ewm(span=slowperiod, adjust=False).mean()
    df['dif'] = df['ema_fast'] - df['ema_slow']
    df['dea'] = df['dif'].ewm(span=signalperiod, adjust=False).mean()
    df['macd_hist'] = (df['dif'] - df['dea']) * 2
    
    return df['dif'].iloc[-1], df['dea'].iloc[-1], df['macd_hist'].iloc[-1]

核心 API 与函数解析

1. run_monthly(func, monthday=1, time='09:30')：
   • 聚宽专用的定时运行 API。设置 monthday=1 确保策略在每个月的第一个交易日开盘时准时执行，避免了日频频繁交易带来的高额摩擦成本。
2. get_bars(security, count, unit='1M', ...)：
   • 获取历史 K 线数据。通过设置 unit='1M'，直接获取月线级别的 Bar 数据，无需手动将日线重采样，极大地简化了多周期数据处理的复杂度。
3. set_option('use_real_price', True)：
   • 强烈建议开启的真实价格模式。在 ETF 轮动中，分红派息会影响实际持仓和价格，开启此模式可确保回测与模拟盘的撮合价格与真实市场完全一致。