聚宽全天候 ETF 轮动：如何用月线 MACD 实现分层仓位配置？

策略设计思路

全天候 ETF 轮动策略的核心在于资产多属性配置与趋势跟踪。我们将候选 ETF 分为三层：

1. 权益进攻层：创业板 ETF（159915）、纳指 ETF（513100）—— 跟踪高弹性、高成长资产。
2. 红利防守层：红利 ETF（510880）—— 提供稳健的股息现金流与低波动防御。
3. 避险避险层：黄金 ETF（518880）、十年国债 ETF（511260）—— 在权益市场系统性下跌时提供避险通道。

仓位分层与 MACD 信号逻辑

• 运行频率：按月调仓（使用 run_monthly），避免日内与周内噪音。
• 信号计算：获取各 ETF 的月线数据，计算 MACD 指标（DIF > DEA 视为多头信号）。
• 动态配置：
  • 第一层（进攻仓，50% 权重）：若创业板或纳指 ETF 的月线 MACD 呈多头状态，优先配置 MACD 动能更强的权益 ETF；若均无信号，则将此仓位降级配置到红利 ETF 或国债 ETF。
  • 第二层（防守仓，30% 权重）：若红利 ETF 的月线 MACD 为多头，配置红利 ETF；否则配置到国债 ETF。
  • 第三层（避险仓，20% 权重）：若黄金 ETF 的月线 MACD 为多头，配置黄金 ETF；否则配置到国债 ETF。

聚宽 Python 策略源码

请在聚宽回测环境中运行以下代码，并开启真实价格模式（set_option('use_real_price', True)）：

import pandas as pd
import numpy as np
from jqlib.technical_analysis import *  # 导入聚宽技术分析库

def initialize(context):
    # 开启真实价格模式
    set_option('use_real_price', True)
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    
    # 定义资产池
    g.etf_dict = {
        'cyb': '159915.XSHE',      # 创业板 ETF
        'nd': '513100.XSHG',       # 纳指 ETF
        'hl': '510880.XSHG',       # 红利 ETF
        'hj': '518880.XSHG',       # 黄金 ETF
        'gz': '511260.XSHG'        # 国债 ETF
    }
    
    # 设定交易费率
    set_order_cost(OrderCost(close_tax=0, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='fund')
    
    # 每月第一个交易日 09:30 运行调仓
    run_monthly(handle_rotation, monthday=1, time='09:30', reference_security='000300.XSHG')

def get_macd_signal(security, context):
    """
    计算月线 MACD 信号
    返回: True (DIF > DEA, 多头), False (DIF <= DEA, 空头)
    """
    # 获取过去 120 个交易日（约 6 个月）的日线数据合成月线，或直接获取月线 bar
    # 聚宽 get_bars 支持 '1M' (一月) 标准 bar
    bars = get_bars(security, count=30, unit='1M', fields=['close'], include_now=True)
    if len(bars) < 26:
        return False
    
    close_prices = bars['close']
    
    # 手动计算 MACD (12, 26, 9)
    ema12 = pd.Series(close_prices).ewm(span=12, adjust=False).mean()
    ema26 = pd.Series(close_prices).ewm(span=26, adjust=False).mean()
    dif = ema12 - ema26
    dea = dif.ewm(span=9, adjust=False).mean()
    
    return dif.iloc[-1] > dea.iloc[-1]

def handle_rotation(context):
    # 1. 获取所有标的的月线 MACD 信号
    signals = {}
    for key, security in g.etf_dict.items():
        signals[key] = get_macd_signal(security, context)
        
    # 2. 决策各分层仓位的目标标的
    target_weights = {sec: 0.0 for sec in g.etf_dict.values()}
    
    # --- 第一层：权益进攻仓 (50% 权重) ---
    # 比较创业板与纳指，若有多头信号，选择动能更强（此处简化为均有多头时各分 25%，或单边多头占 50%）
    if signals['cyb'] and signals['nd']:
        target_weights[g.etf_dict['cyb']] += 0.25
        target_weights[g.etf_dict['nd']] += 0.25
    elif signals['cyb']:
        target_weights[g.etf_dict['cyb']] += 0.50
    elif signals['nd']:
        target_weights[g.etf_dict['nd']] += 0.50
    else:
        # 权益无信号，降级至红利（若红利有多头信号）或国债
        if signals['hl']:
            target_weights[g.etf_dict['hl']] += 0.50
        else:
            target_weights[g.etf_dict['gz']] += 0.50
            
    # --- 第二层：红利防守仓 (30% 权重) ---
    if signals['hl']:
        target_weights[g.etf_dict['hl']] += 0.30
    else:
        target_weights[g.etf_dict['gz']] += 0.30
        
    # --- 第三层：避险仓 (20% 权重) ---
    if signals['hj']:
        target_weights[g.etf_dict['hj']] += 0.20
    else:
        target_weights[g.etf_dict['gz']] += 0.20
        
    # 3. 执行调仓
    adjust_portfolio(context, target_weights)

def adjust_portfolio(context, target_weights):
    total_value = context.portfolio.total_value
    
    # 先卖出权重降低或不再持有的标的，释放可用资金
    for security in list(context.portfolio.positions.keys()):
        if security not in target_weights or target_weights[security] == 0:
            order_target(security, 0)
            
    # 再买入或调整至目标权重
    for security, weight in target_weights.items():
        target_value = total_value * weight
        # 避免小额频繁交易，设置 1% 的调仓阈值
        current_value = context.portfolio.positions[security].value
        if abs(target_value - current_value) > total_value * 0.01:
            order_target_value(security, target_value)
            log.info(f"调整 {security} 权重至 {weight*100}%")