JoinQuant 平台如何计算 ATR 指标并实现移动止损（Trailing Stop）？

在聚宽（JoinQuant）平台上，利用 ATR（平均真实波幅）指标进行移动止损（Trailing Stop）是一种非常经典的风险管理策略。

核心逻辑

1. 计算 ATR：衡量当前市场的波动率。
2. 确定止损价：
   • 多头策略：止损价 = 历史最高价（持仓期间） - 系数 × ATR
   • 随着价格上涨，最高价更新，止损价跟随上移。
   • 如果价格下跌，最高价不变，止损价不变（只升不降）。
3. 触发止损：当当前价格跌破止损价时，执行卖出。

策略代码实现

以下是一个完整的策略示例。该策略使用 talib 库计算 ATR，当价格突破 20 日均线买入，并使用 ATR 进行移动止损。

# -*- coding: utf-8 -*-
import jqdata
import talib
import numpy as np

def initialize(context):
    """
    初始化函数
    """
    # 设定基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 过滤掉order系列API产生的比error级别低的log
    log.set_level('order', 'error')
    
    # --- 策略参数设置 ---
    g.security = '000001.XSHE'  # 操作标的：平安银行
    g.atr_period = 14           # ATR计算周期
    g.atr_multiplier = 2.0      # ATR倍数，用于设置止损宽度
    g.ma_period = 20            # 入场均线周期
    
    # --- 全局变量 ---
    g.highest_price = 0         # 持仓期间的最高价
    
    # 每天开盘时运行
    run_daily(market_open, time='every_bar')

def market_open(context):
    """
    每日交易逻辑
    """
    security = g.security
    
    # 1. 获取历史数据 (多取一些数据以确保计算指标时长度足够)
    # 需要获取 high, low, close 用于计算 ATR
    data_len = g.ma_period + g.atr_period + 10
    h_data = attribute_history(security, data_len, '1d', ['high', 'low', 'close'])
    
    # 提取为 numpy 数组，供 talib 使用
    high_arr = h_data['high'].values
    low_arr = h_data['low'].values
    close_arr = h_data['close'].values
    
    # 2. 计算技术指标
    # 计算 ATR
    atr_values = talib.ATR(high_arr, low_arr, close_arr, timeperiod=g.atr_period)
    current_atr = atr_values[-1]
    
    # 计算 MA (用于简单的入场逻辑)
    ma = close_arr[-g.ma_period:].mean()
    
    # 获取当前价格
    current_price = close_arr[-1]
    
    # 获取当前持仓
    position = context.portfolio.positions[security]
    
    # 3. 交易逻辑
    
    # --- 止损与卖出逻辑 (如果有持仓) ---
    if position.total_amount > 0:
        # 更新持仓期间的最高价
        if current_price > g.highest_price:
            g.highest_price = current_price
        
        # 计算当前的移动止损价
        stop_loss_price = g.highest_price - (current_atr * g.atr_multiplier)
        
        # 记录止损价以便在回测图中查看
        record(StopPrice=stop_loss_price)
        
        # 检查是否跌破止损价
        if current_price < stop_loss_price:
            log.info("触发ATR移动止损: 当前价:%.2f < 止损价:%.2f (最高价:%.2f, ATR:%.2f)" % 
                     (current_price, stop_loss_price, g.highest_price, current_atr))
            order_target(security, 0)
            # 重置最高价
            g.highest_price = 0
            return # 卖出后今日不再进行买入判断

    # --- 买入逻辑 (如果没有持仓) ---
    else:
        # 简单的突破均线买入
        if current_price > ma:
            log.info("价格突破均线，买入: %.2f" % current_price)
            order_value(security, context.portfolio.available_cash)
            # 买入后，初始化最高价为当前价格
            g.highest_price = current_price
            
    # 绘制曲线
    record(Price=current_price)
    record(MA=ma)

代码关键点解析

1. 数据获取 (attribute_history):

   • ATR 计算需要最高价（High）、最低价（Low）和收盘价（Close）。
   • 我们获取的数据长度要比 ATR 周期长，因为 talib 计算初期需要一些预热数据。

2. 计算 ATR (talib.ATR):

   • talib 是 Python 中最高效的技术指标库，聚宽环境已内置。
   • talib.ATR(high, low, close, timeperiod=14) 返回的是一个数组，我们取最后一个值 [-1] 作为当前的 ATR。

3. 移动止损逻辑:

   • g.highest_price: 这是一个全局变量，用于记录从买入那一刻开始，直到现在的最高价格。
   • 更新机制: 只要持有股票，如果 current_price > g.highest_price，就更新 g.highest_price。
   • 止损线计算: stop_loss_price = g.highest_price - (current_atr * g.atr_multiplier)。
   • 系数调整: g.atr_multiplier 通常设置为 2.0 到 3.0。系数越小，止损越紧（容易被震荡洗出）；系数越大，止损越松（回撤可能较大）。

4. 买入初始化:

   • 在执行买入操作 (order_value) 后，必须立即将 g.highest_price 初始化为当前价格，否则止损逻辑会出错。

常见问题 (Q&A)

Q1: 为什么使用 ATR 而不是固定百分比（如跌 10%）止损？
A: ATR 是基于波动率的。当市场波动剧烈时，ATR 变大，止损间距自动变宽，防止被正常的市场噪音震出局；当市场平稳时，ATR 变小，止损变紧，能更好地锁定利润。它比固定百分比更具适应性。

Q2: 如何在分钟级别回测中使用此策略？
A: 代码逻辑基本通用。只需将 attribute_history 的频率参数从 '1d' 改为 '1m'，并将 run_daily 改为 run_daily(market_open, time='every_bar') 即可实现分钟级的移动止损。

Q3: talib 库在聚宽中需要安装吗？
A: 不需要。聚宽的回测和研究环境默认已经预装了 talib 库，直接 import talib 即可使用。

Q4: 止损价是盘中触发还是收盘触发？
A: 上述代码写在 market_open 中（按天回测通常在开盘运行），使用的是前一天的收盘数据或当天的开盘数据（取决于具体设置）。如果是按分钟回测，则是盘中每分钟触发。如果希望在日线级别盘中实时触发，可以使用 handle_data 并获取分钟数据进行判断。