如何在聚宽中实现 ETF 动量轮动策略？结合均线、成交额与多时段分钟趋势确认（附 Python 源码）

聚宽 ETF 动量轮动策略：结合均线、成交额与分钟趋势分批建仓

在量化交易中，ETF 轮动策略因其波动小、抗风险能力强而广受欢迎。结合动量效应（Momentum）、流动性筛选（成交额）以及日内分钟趋势确认，可以有效降低追高风险并平滑建仓成本。本文将详细介绍如何在 JoinQuant（聚宽）平台上实现这一完整的策略架构。

一、 核心策略逻辑设计

1. 流动性筛选：
   从候选 ETF 池中，筛选出过去 5 个交易日日均成交额前 $N$ 名的 ETF，确保交易流动性，规避滑点风险。
2. 动量得分计算（加权线性回归）：
   使用聚宽内置的 PLRC（价格与日期线性回归系数）或通过历史收盘价进行加权线性回归。对近期价格赋予更高权重，计算斜率作为动量得分，选择得分最高的标的。
3. 趋势过滤（均线）：
   标的价格必须高于其 20 日均线（MA20），以确保处于大周期上升通道。
4. 日内分批建仓与分钟趋势确认：
   在每日的 13:05、13:40、14:10 和 14:55 四个时间点，获取当前分钟 Bar 的收盘价。若当前分钟价格高于前 5 分钟均价（即分钟线趋势向上），则触发分批建仓（每次买入目标仓位的 25%）。

二、 聚宽 Python 策略源码

请在聚宽回测环境中运行以下代码（建议选择分钟频率回测，并开启真实价格模式）：

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

def initialize(context):
    # 开启真实价格模式
    set_option('use_real_price', True)
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    
    # 候选 ETF 资金池（涵盖宽基、行业、跨境等）
    g.etf_pool = [
        '510300.XSHG', # 沪深300ETF
        '510500.XSHG', # 中证500ETF
        '518880.XSHG', # 黄金ETF
        '513100.XSHG', # 纳指ETF
        '159915.XSHE', # 创业板ETF
        '510050.XSHG', # 上证50ETF
    ]
    
    g.top_n = 3          # 每日流动性筛选保留数量
    g.target_num = 1     # 最终轮动持仓的 ETF 数量
    g.batch_ratio = 0.25 # 每次分批建仓比例 (25%)
    
    # 记录当日已建仓次数
    g.buy_count = 0
    g.target_etf = None
    
    # 设定定时运行任务
    # 1. 盘前筛选与动量计算
    run_daily(before_market_open, time='09:00')
    # 2. 盘中特定分钟时间点进行趋势确认与分批建仓
    run_daily(market_trade_check, time='13:05')
    run_daily(market_trade_check, time='13:40')
    run_daily(market_trade_check, time='14:10')
    run_daily(market_trade_check, time='14:55')
    # 3. 每日收盘后重置状态
    run_daily(after_market_close, time='15:30')

def before_market_open(context):
    # 1. 流流动性筛选：获取过去5天的成交额
    hist_money = history(5, unit='1d', field='money', security_list=g.etf_pool)
    mean_money = hist_money.mean().sort_values(ascending=False)
    active_etfs = list(mean_money.index[:g.top_n])
    
    # 2. 计算动量得分与均线过滤
    momentum_scores = {}
    for etf in active_etfs:
        # 获取过去12天的收盘价用于计算 PLRC12
        close_data = attribute_history(etf, 12, '1d', ['close'])
        closes = close_data['close'].values
        
        # 均线过滤：判断是否在20日均线之上
        ma20_data = attribute_history(etf, 20, '1d', ['close'])
        ma20 = ma20_data['close'].mean()
        current_price = closes[-1]
        
        if current_price > ma20:
            # 加权线性回归计算斜率 (对近期价格赋予更高权重)
            x = np.arange(12)
            weights = np.exp(x / 12.0) # 指数权重
            # 加权拟合
            poly = np.polyfit(x, closes, 1, w=weights)
            slope = poly[0] / current_price # 归一化斜率
            momentum_scores[etf] = slope
            
    # 3. 确定今日目标轮动标的
    if momentum_scores:
        sorted_etfs = sorted(momentum_scores.items(), key=lambda item: item[1], reverse=True)
        g.target_etf = sorted_etfs[0][0]
        log.info(f"今日选定动量最强且在MA20之上的ETF: {g.target_etf}，动量得分: {sorted_etfs[0][1]:.4f}")
    else: 
        g.target_etf = None
        log.info("今日无满足MA20过滤条件的ETF，保持空仓。")
        
    # 4. 调仓换股：若持仓不等于目标，则在开盘前或开盘立即清仓非目标ETF
    for hold_etf in list(context.portfolio.positions.keys()):
        if hold_etf != g.target_etf:
            order_target(hold_etf, 0)
            log.info(f"非目标持仓，清仓: {hold_etf}")

def market_trade_check(context):
    """盘中特定时间点运行，确认分钟趋势并分批建仓"""
    if not g.target_etf:
        return
        
    # 获取当前目标 ETF 的分钟线数据
    # 确认分钟趋势：当前价格是否高于过去5分钟均价
    minute_data = attribute_history(g.target_etf, 5, '1m', ['close'])
    ma5_minute = minute_data['close'].mean()
    current_minute_price = minute_data['close'][-1]
    
    if current_minute_price > ma5_minute:
        # 分钟趋势向上，执行分批建仓
        total_value = context.portfolio.total_value
        batch_value = total_value * g.batch_ratio
        
        # 检查可用现金是否足够
        if context.portfolio.available_cash >= batch_value:
            order_value(g.target_etf, batch_value)
            g.buy_count += 1
            log.info(f"时间 {context.current_dt.time()} 趋势确认：当前价 {current_minute_price} > 5分均价 {ma5_minute:.4f}。执行第 {g.buy_count} 批建仓，买入金额: {batch_value:.2f}")
        else:
            log.warn("可用资金不足，无法执行本批次建仓。")
    else:
        log.info(f"时间 {context.current_dt.time()} 趋势未确认：当前价 {current_minute_price} <= 5分均价 {ma5_minute:.2f}，暂不建仓。")

def after_market_close(context):
    log.info(f"今日交易结束。累计建仓次数: {g.buy_count}/4")
    g.buy_count = 0 # 重置计数器

三、 策略关键技术解析

1. 加权线性回归（Weighted Linear Regression）：
   传统的动量计算（如简单收益率）容易受到单日异常暴涨暴跌的影响。本策略采用 np.polyfit 并引入指数权重 weights = np.exp(x / 12.0)，使越接近当天的价格对斜率拟合的影响越大，从而更灵敏地捕捉近期的真实动量趋势。
2. 多时段分批建仓（Multi-period Batching）：
   通过 run_daily 在 13:05、13:40、14:10、14:55 触发交易，将资金均分为 4 份。这种日内分批建仓法能够有效平滑日内波动带来的随机性，避免在日内最高点一次性“站岗”。
3. 分钟趋势过滤：
   在建仓时刻，策略通过 attribute_history(..., '1m') 获取前 5 分钟的均价。只有当当前价格突破 5 分钟均线时才执行买入，确保在微观层面上，买入动作发生在价格向上脉冲的瞬间，提高了建仓的胜率。