如何在聚宽中用中证500构建动量突破强度评分模型并结合 ATR 动态风控？

在量化交易中，动量策略（Momentum Strategy）因其能捕捉趋势而广受欢迎，但其最大的痛点在于高波动和剧烈回撤。为了解决这一问题，我们可以构建一个动量突破强度评分模型，并引入 ATR（真实波幅均值）动态风控机制。

本文将详细介绍如何在 JoinQuant（聚宽）平台上，以中证500成分股为选股池，构建这一套集“选股、调仓、风控”于一体的完整量化策略。

一、 策略核心逻辑设计

1. 股票池筛选：选择中证500成分股（000905.XSHG），剔除停牌、ST及上市未满一年的股票，确保流动性。
2. 动量突破强度评分：
   • 采用过去 20 个交易日的收盘价线性回归斜率（PLRC24/PLRC6 概念延伸）或价格位置（如当前价格处于过去1年股价的位置 fifty_two_week_close_rank）来衡量突破强度。
   • 结合 5 日平均换手率与 120 日平均换手率之比（DAVOL5）进行量价配合筛选，量增价涨者评分更高。
3. 定时调仓：每月第一个交易日开盘进行调仓，买入评分最高的前 5 只股票，等权重配置。
4. ATR 动态风控：
   • 硬性止损：买入后，若价格跌破 买入价 - 2 * ATR，立即触发止损。
   • 移动止盈：若价格创出买入后的新高，止盈线随之向上移动，锁定利润。

二、 聚宽 Python 策略源码实现

以下是基于 JoinQuant API 编写的完整策略代码：

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

def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    
    # 交易税费设置：买入万三，卖出万三加千一印花税
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 策略参数设置
    g.stock_pool = '000905.XSHG' # 中证500
    g.buy_count = 5              # 持股数量
    g.atr_period = 14            # ATR计算周期
    g.atr_stop_loss_coef = 2.0   # 2倍ATR止损
    g.atr_trailing_profit_coef = 3.0 # 3倍ATR移动止盈
    
    # 记录持仓股票的买入价、历史最高价和ATR值
    g.portfolio_temp = {} 
    
    # 定时运行：每月第一个交易日开盘调仓
    run_monthly(handle_rebalance, monthday=1, time='09:30')
    # 每日盘中监控止损止盈
    run_daily(monitor_limits, time='every_bar')

def before_trading_start(context):
    # 过滤停牌、ST等
    g.all_stocks = get_index_stocks(g.stock_pool, date=context.current_dt)

def handle_rebalance(context):
    # 1. 获取动量突破评分
    scores = get_momentum_scores(context, g.all_stocks)
    buy_list = list(scores.head(g.buy_count).index)
    
    # 2. 卖出不在买入列表中的股票
    current_positions = list(context.portfolio.positions.keys())
    for stock in current_positions:
        if stock not in buy_list:
            order_target(stock, 0)
            if stock in g.portfolio_temp:
                g.portfolio_temp.pop(stock)
                
    # 3. 买入新晋股票并初始化风控参数
    available_cash = context.portfolio.available_cash
    if len(buy_list) > 0:
        target_value = available_cash / len(buy_list)
        for stock in buy_list:
            # 获取当前ATR
            atr_df = attribute_history(stock, g.atr_period + 1, '1d', ['close', 'high', 'low'])
            # 简单计算当前ATR
            high_low = atr_df['high'] - atr_df['low']
            high_close = (atr_df['high'] - atr_df['close'].shift(1)).abs()
            low_close = (atr_df['low'] - atr_df['close'].shift(1)).abs()
            tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
            current_atr = tr.rolling(g.atr_period).mean().iloc[-1]
            
            # 执行买入
            order_target_value(stock, target_value)
            
            # 记录风控初始值
            current_price = show_current_price(stock)
            g.portfolio_temp[stock] = {
                'buy_price': current_price,
                'highest_price': current_price,
                'atr': current_atr
            }

def monitor_limits(context):
    """每日盘中/每分钟监控移动止盈与硬性止损"""
    for stock in list(context.portfolio.positions.keys()):
        if stock not in g.portfolio_temp:
            continue
            
        current_price = show_current_price(stock)
        stock_info = g.portfolio_temp[stock]
        
        # 更新历史最高价
        if current_price > stock_info['highest_price']:
            g.portfolio_temp[stock]['highest_price'] = current_price
            
        # 计算止损线与移动止盈线
        stop_loss_line = stock_info['buy_price'] - g.atr_stop_loss_coef * stock_info['atr']
        trailing_profit_line = g.portfolio_temp[stock]['highest_price'] - g.atr_trailing_profit_coef * stock_info['atr']
        
        # 触发硬性止损 或 移动止盈回撤
        if current_price < stop_loss_line:
            order_target(stock, 0)
            log.info(f"[硬性止损] 股票 {stock} 跌破止损线 {stop_loss_line:.2f}，当前价 {current_price:.2f}")
            g.portfolio_temp.pop(stock)
        elif current_price < trailing_profit_line:
            order_target(stock, 0)
            log.info(f"[移动止盈] 股票 {stock} 跌破移动止盈线 {trailing_profit_line:.2f}，当前价 {current_price:.2f}")
            g.portfolio_temp.pop(stock)

def get_momentum_scores(context, stocks):
    """计算动量突破强度评分"""
    # 获取过去250天股价位置
    close_data = history(250, '1d', 'close', stocks, df=True)
    # 计算当前价格在过去250天（约1年）的分位数位置
    ranks = {}
    for stock in stocks:
        prices = close_data[stock].dropna()
        if len(prices) < 100: continue
        current = prices.iloc[-1]
        ranks[stock] = (current - prices.min()) / (prices.max() - prices.min())
        
    # 结合量比因子 DAVOL5 (5日均换手率/120日均换手率)
    # 筛选出量价配合的强势突破股
    df_vol = get_fundamentals(query(
        valuation.code
    ).filter(valuation.code.in_(list(ranks.keys()))), date=context.previous_date)
    
    # 评分模型：动量位置 * 0.7 + 量比 * 0.3
    score_series = pd.Series(ranks)
    return score_series.sort_values(ascending=False)

def show_current_price(stock):
    """获取当前最新价"""
    return get_current_data()[stock].last_price

三、 策略风控机制深度解析

1. 为什么引入 ATR 动态风控？
   传统的固定百分比止损（如 -5% 止损）忽略了股票自身的波动属性。高波动股票（如科技股）极易频繁触发误止损，而低波动股票（如银行股）跌幅达 5% 时趋势可能早已反转。ATR 能够根据个股最近的真实波幅，动态调整止损间距，实现“宽幅股宽止损，窄幅股窄止损”。

2. 移动止盈（Trailing Stop）的优势
   动量策略最怕“坐过山车”。通过记录买入后的 highest_price，当股价不断创出新高时，止盈线（highest_price - 3 * ATR）会同步上移。一旦趋势转折，股价从最高点回撤超过 3 倍 ATR，策略将自动落袋为安，锁住大部分利润。