如何在聚宽 ETF 轮动策略中增加场内溢价率过滤、异常放量过滤和阶段回撤止盈止损？

聚宽 ETF 轮动风控策略设计方案

在 ETF 轮动策略中，除了常规的动量（如 20 日收益率）排序外，加入场内溢价率过滤、异常放量过滤和阶段回撤止盈止损可以显著提升策略的稳健性，避免买入高溢价资产、追高见顶资产，并在市场系统性下跌时及时锁定利润或控制亏损。

一、 核心风控逻辑与聚宽 API 实现

1. 场内溢价率过滤

• 原理：场内 ETF 的交易价格（close）若大幅高于其单位净值（IOPV 或 unit_net_value），代表溢价率过高，随时面临套利资金砸盘。当 $\text{溢价率} = (\text{收盘价} / \text{单位净值}) - 1 > \text{阈值}$（如 1%）时，应过滤该 ETF。
• 聚宽 API：使用 get_extras 获取 unit_net_value（基金单位净值），结合 attribute_history 获取最新收盘价。

2. 异常放量过滤

• 原理：若某 ETF 短期成交量异常放大（例如当日成交量超过过去 20 日平均成交量的 3 倍），通常意味着情绪过热或主力资金出逃，此时应避免买入。
• 聚宽 API：使用 attribute_history 获取 volume，计算 current_volume > 3 * MA_volume。

3. 阶段回撤止盈止损

• 原理：记录持仓个股的建仓后最高价。若当前价格相比持仓期间的最高价回撤超过一定比例（如 5%），则触发移动止损（Trailing Stop）。
• 聚宽 API：利用全局变量 g 维护一个字典记录持仓标的的最高价，在 handle_data 或 run_daily 中每日更新并判断。

二、 完整策略源码（Python 3）

以下是结合上述风控机制的完整聚宽策略代码：

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

def initialize(context):
    # 开启真实价格模式
    set_option('use_real_price', True)
    set_benchmark('000300.XSHG')
    
    # 交易费率设置
    set_order_cost(OrderCost(open_tax=0, close_tax=0, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='fund')
    
    # 策略参数
    g.etf_pool = [
        '510300.XSHG', # 沪深300ETF
        '510500.XSHG', # 中证500ETF
        '159915.XSHE', # 创业板ETF
        '513100.XSHG', # 纳指ETF
        '518880.XSHG'  # 黄金ETF
    ]
    
    g.momentum_window = 20      # 动量计算周期
    g.max_premium_rate = 0.01   # 最大允许溢价率 (1%)
    g.volume_ratio_limit = 3.0  # 异常放量倍数 (3倍)
    g.stop_loss_rate = 0.05     # 移动止损阈值 (5%)
    
    # 记录持仓历史最高价，用于移动止损
    g.highest_prices = {}
    
    # 每日运行轮动与风控
    run_daily(market_open, time='09:30')

def market_open(context):
    # 1. 移动止损风控（优先执行）
    handle_stop_loss(context)
    
    # 2. 轮动选股逻辑
    today = context.current_dt.date()
    valid_etfs = []
    
    for etf in g.etf_pool:
        # 过滤停牌
        current_data = get_current_data()
        if current_data[etf].paused:
            continue
            
        # 溢价率过滤
        if is_over_premium(etf, today):
            log.info(f"{etf} 溢价率过高，过滤")
            continue
            
        # 异常放量过滤
        if is_abnormal_volume(etf):
            log.info(f"{etf} 异常放量，过滤")
            continue
            
        valid_etfs.append(etf)
        
    # 3. 动量排序与调仓
    if not valid_etfs:
        return
        
    # 计算 20 日收益率
    scores = {}
    for etf in valid_etfs:
        hist = attribute_history(etf, g.momentum_window, '1d', ['close'])
        if len(hist) == g.momentum_window:
            return_rate = (hist['close'][-1] - hist['close'][0]) / hist['close'][0]
            scores[etf] = return_rate
            
    if not scores:
        return
        
    # 选出动量最强的 1 只 ETF
    best_etf = max(scores, key=scores.get)
    
    # 如果最强 ETF 动量为正，则调仓
    if scores[best_etf] > 0:
        # 卖出不在持仓中的其他 ETF
        for hold_etf in list(context.portfolio.positions.keys()):
            if hold_etf != best_etf:
                order_target(hold_etf, 0)
                if hold_etf in g.highest_prices:
                    del g.highest_prices[hold_etf]
                    
        # 买入最强 ETF
        if best_etf not in context.portfolio.positions:
            cash = context.portfolio.available_cash
            order_value(best_etf, cash)
            g.highest_prices[best_etf] = get_current_data()[best_etf].last_price
    else:
        # 动量全部为负，空仓避险
        for hold_etf in list(context.portfolio.positions.keys()):
            order_target(hold_etf, 0)
        g.highest_prices.clear()

def is_over_premium(etf, date):
    """判断是否超过最大溢价率"""
    # 获取最新收盘价
    close_price = get_current_data()[etf].last_price
    # 获取最新单位净值
    net_value_df = get_extras('unit_net_value', [etf], end_date=date, count=1)
    if not net_value_df.empty:
        net_value = net_value_df[etf].iloc[-1]
        premium_rate = (close_price / net_value) - 1
        return premium_rate > g.max_premium_rate
    return False

def is_abnormal_volume(etf):
    """判断是否异常放量"""
    hist = attribute_history(etf, 21, '1d', ['volume'])
    if len(hist) < 21:
        return False
    current_vol = hist['volume'].iloc[-1]
    ma_vol = hist['volume'].iloc[:-1].mean()
    return current_vol > g.volume_ratio_limit * ma_vol

def handle_stop_loss(context):
    """移动止损逻辑"""
    current_data = get_current_data()
    for etf in list(context.portfolio.positions.keys()):
        curr_price = current_data[etf].last_price
        
        # 初始化或更新最高价
        if etf not in g.highest_prices:
            g.highest_prices[etf] = curr_price
        else:
            g.highest_prices[etf] = max(g.highest_prices[etf], curr_price)
            
        # 计算回撤
        drawdown = (g.highest_prices[etf] - curr_price) / g.highest_prices[etf]
        if drawdown > g.stop_loss_rate:
            log.warning(f"{etf} 触发移动止损，当前价: {curr_price}, 最高价: {g.highest_prices[etf]}, 回撤: {drawdown:.2%}")
            order_target(etf, 0)
            del g.highest_prices[etf]

三、 策略优势与注意事项

1. 避免套利砸盘：跨境 ETF（如纳指 ETF、日经 ETF）在牛市中极易出现高溢价，通过 get_extras('unit_net_value') 过滤，能有效防止买在场内情绪最高点。
2. 防止假突破：异常放量过滤机制可以规避主力资金对冲或对倒交易导致的“放量诱多”陷阱。
3. 动态保护利润：移动止损（Trailing Stop）相比固定止损更具灵活性，在 ETF 开启大级别主升浪时能一路追踪，在趋势反转回撤达到 5% 时果断离场，锁住大部分利润。