PTrade 平台如何用 Python 实现 ATR 动态追踪止损？（附源码）

这是一个基于 PTrade 平台的量化交易策略，实现了 ATR（平均真实波幅）动态追踪止损 的逻辑。

策略原理

1. 入场信号（示例）：为了演示止损逻辑，这里使用简单的双均线策略作为入场信号（5日均线上穿20日均线，即金叉买入）。
2. ATR 动态追踪止损（核心）：
   • 计算 ATR：使用 talib 库计算 14 日 ATR 指标。
   • 记录最高价：持仓期间，记录自开仓以来的最高价格（Highest_High）。
   • 计算止损线：止损价 = 历史最高价 - (系数 * ATR)。
   • 触发止损：当当前价格跌破止损价时，触发卖出信号。
   • 动态调整：随着股价上涨，Highest_High 会更新，止损线也会随之上移，从而锁定利润；当股价下跌时，止损线保持不变，直到价格触及止损线离场。

策略代码

import talib
import numpy as np

def initialize(context):
    """
    初始化函数，设置参数和股票池
    """
    # 设置要操作的股票，这里以贵州茅台为例
    g.security = ['600519.SS']
    set_universe(g.security)
    
    # 策略参数设置
    g.atr_period = 14       # ATR计算周期
    g.atr_multiplier = 2.0  # ATR倍数，用于计算止损距离
    g.ma_short = 5          # 短期均线周期
    g.ma_long = 20          # 长期均线周期
    
    # 用于记录持仓股票的持仓期间最高价，格式：{stock_code: highest_price}
    g.highest_prices = {} 

def before_trading_start(context, data):
    """
    盘前处理
    """
    pass

def handle_data(context, data):
    """
    盘中每分钟或每天运行的逻辑
    """
    # 获取当前回测/交易的股票列表
    security_list = g.security
    
    # 获取历史数据，取足够长的数据以确保指标计算准确
    # count=100 保证有足够的数据计算 MA20 和 ATR14
    hist_data = get_history(100, frequency='1d', field=['high', 'low', 'close'], security_list=security_list, fq='pre')
    
    for stock in security_list:
        # 检查该股票是否停牌，如果停牌则跳过
        if stock in data and data[stock].is_open == 0:
            continue
            
        # 获取该股票的 K 线数据
        # 注意：get_history 返回的数据结构在单只和多只股票时可能不同
        # 这里使用 query 确保获取到的是 DataFrame
        if len(security_list) == 1:
            df = hist_data
        else:
            df = hist_data.query('code == "%s"' % stock)
        
        # 提取 numpy 数组用于 talib 计算
        high_prices = df['high'].values
        low_prices = df['low'].values
        close_prices = df['close'].values
        
        # 确保数据长度足够
        if len(close_prices) < 30:
            continue
            
        # --- 1. 计算技术指标 ---
        # 计算 ATR
        atr = talib.ATR(high_prices, low_prices, close_prices, timeperiod=g.atr_period)
        current_atr = atr[-1] # 获取最新的 ATR 值
        
        # 计算均线 (用于入场)
        ma_short = talib.MA(close_prices, timeperiod=g.ma_short)
        ma_long = talib.MA(close_prices, timeperiod=g.ma_long)
        
        # 获取当前价格
        current_price = data[stock]['close']
        
        # 获取当前持仓
        position = get_position(stock)
        
        # --- 2. 交易逻辑 ---
        
        # A. 如果当前持有仓位 -> 执行 ATR 追踪止损逻辑
        if position.amount > 0:
            # 初始化或更新持仓期间的最高价
            # 如果该股票不在记录中，说明是刚买入，初始化为买入成本或当前价
            if stock not in g.highest_prices:
                g.highest_prices[stock] = current_price
            else:
                # 如果当前价格创新高，则更新最高价
                if current_price > g.highest_prices[stock]:
                    g.highest_prices[stock] = current_price
            
            # 计算动态止损价
            # 止损价 = 持仓期最高价 - (ATR系数 * 当前ATR)
            stop_loss_price = g.highest_prices[stock] - (g.atr_multiplier * current_atr)
            
            # 打印调试日志 (可选)
            # log.info("%s 持仓中, 最高价: %.2f, ATR: %.2f, 止损价: %.2f, 当前价: %.2f" % (stock, g.highest_prices[stock], current_atr, stop_loss_price, current_price))
            
            # 判断是否触发止损
            if current_price < stop_loss_price:
                log.info("触发ATR追踪止损: %s, 卖出价格: %.2f" % (stock, current_price))
                order_target(stock, 0) # 清仓
                
                # 卖出后，清除最高价记录
                if stock in g.highest_prices:
                    del g.highest_prices[stock]
        
        # B. 如果当前空仓 -> 执行均线入场逻辑
        else:
            # 简单的金叉判断：短期均线上穿长期均线
            # 前一天短均线 < 长均线，且 今天短均线 > 长均线
            if ma_short[-2] < ma_long[-2] and ma_short[-1] > ma_long[-1]:
                log.info("均线金叉买入: %s, 价格: %.2f" % (stock, current_price))
                # 全仓买入 (这里简单使用可用资金买入)
                order_value(stock, context.portfolio.cash)
                
                # 买入后，立即初始化最高价记录，防止下一分钟/下一天逻辑出错
                g.highest_prices[stock] = current_price

def after_trading_end(context, data):
    """
    盘后处理
    """
    pass

代码关键点解析

1. import talib: PTrade 环境内置了 TA-Lib 库，这是计算 ATR 最标准、最高效的方法。
2. g.highest_prices 字典:
   • 这是实现“追踪”功能的关键。我们需要记住自买入以来，价格达到过的最高点。
   • 当持有股票时，如果 current_price > g.highest_prices[stock]，我们就更新这个最高价。
   • 当卖出股票后，我们需要 del 删除这个记录，以免影响下一次交易。
3. 止损公式:
   • stop_loss_price = g.highest_prices[stock] - (g.atr_multiplier * current_atr)
   • 这个公式体现了“动态”特性。随着股价上涨，g.highest_prices 变大，stop_loss_price 也跟着变大（上移），从而保护浮盈。
   • g.atr_multiplier (ATR倍数) 通常设置为 2.0 到 3.0 之间。倍数越小，止损越紧，容易被震荡洗出；倍数越大，止损越松，回撤可能较大。
4. 数据获取:
   • 使用了 get_history(100, ...)。虽然 ATR 只需要 14 天数据，但为了计算结果的稳定性（ATR 和 MA 都需要一定的历史数据来平滑），取 100 条数据是比较稳妥的做法。

Q&A

Q1: 为什么使用 ATR 而不是固定百分比止损？
A: 固定百分比（如跌 5% 止损）忽略了市场的波动率。ATR 代表了资产的平均真实波幅。在市场波动剧烈时，ATR 变大，止损间距自动变宽，防止被正常波动震出；在市场平稳时，ATR 变小，止损间距变窄，能更敏锐地锁定利润。

Q2: 如何调整策略的灵敏度？
A: 修改 initialize 中的 g.atr_multiplier。

• 调大 (如 3.0)：适合长线趋势，能容忍更大的回撤，不易被洗盘。
• 调小 (如 1.5)：适合短线交易，反应快，但容易频繁止损。

Q3: 这个策略可以直接用于实盘吗？
A: 该代码是一个基础框架。在实盘前，建议您：

1. 添加风控：比如单只股票最大持仓比例、大盘择时过滤。
2. 优化入场：目前的双均线金叉比较基础，可以替换为您自己更有效的入场信号。
3. 资金管理：代码中使用 order_value(stock, context.portfolio.cash) 是全仓买入，实盘建议根据资金量分仓操作。