如何在 PTrade 策略中编写和调用自定义的技术指标？

在PTrade量化交易平台中，虽然系统提供了MACD、KDJ等常见指标的直接调用接口，但ATR（平均真实波幅）指标通常需要用户自行编写函数来计算。

ATR是一个衡量市场波动率的指标，常用于止损计算或仓位管理。

以下是关于如何在PTrade中编写ATR计算函数并在策略中使用的完整指南。

1. ATR 计算原理

ATR的计算步骤如下：

1. 计算真实波幅（TR）：取以下三者的最大值：
   • 当前最高价 - 当前最低价
   • |当前最高价 - 前一日收盘价|
   • |当前最低价 - 前一日收盘价|
2. 计算ATR：对TR进行N周期的移动平均（通常使用简单移动平均SMA）。

2. 策略代码实现

以下是一个完整的策略示例。该策略包含了一个名为 calculate_atr 的自定义函数，并在 handle_data 中调用该函数来根据波动率动态调整止损点。

import numpy as np
import pandas as pd

def initialize(context):
    """
    初始化函数，设置股票池和全局变量
    """
    # 设置我们要操作的股票，以恒生电子为例
    g.security = '600570.SS'
    set_universe(g.security)
    
    # ATR计算周期
    g.atr_period = 14
    # 移动平均线周期
    g.ma_period = 20
    
    # 记录策略运行天数，用于确保有足够数据计算指标
    g.days = 0

def calculate_atr(security, period, frequency='1d'):
    """
    自定义ATR指标计算函数
    
    参数:
    security: 标的代码
    period: ATR计算周期
    frequency: 数据频率，默认为日线
    
    返回:
    最新的ATR值 (float)
    """
    # 获取历史数据
    # 为了确保计算准确，获取的数据量应大于 period，建议取 period * 2 或固定 100 条
    # 需要 high, low, close 字段
    count = 100
    h = get_history(count, frequency, ['high', 'low', 'close'], security, fq='pre', include=False)
    
    # 如果数据不足，返回 None
    if len(h) < period + 1:
        return None
    
    # 将数据转换为 DataFrame 方便计算
    # 注意：PTrade的get_history返回格式根据版本可能不同，这里统一处理为DataFrame
    if isinstance(h, dict):
        # 如果是字典(多股查询或特定版本)，提取对应股票的DataFrame
        df = h[security]
    else:
        df = h
        
    # 确保 df 是 DataFrame
    if not isinstance(df, pd.DataFrame):
        df = pd.DataFrame(df)

    # 计算前一日收盘价
    df['pre_close'] = df['close'].shift(1)
    
    # 计算 TR 的三个组成部分
    # 1. 当前最高 - 当前最低
    df['hl'] = df['high'] - df['low']
    # 2. |当前最高 - 前收|
    df['hc'] = (df['high'] - df['pre_close']).abs()
    # 3. |当前最低 - 前收|
    df['lc'] = (df['low'] - df['pre_close']).abs()
    
    # 计算 TR (True Range): 取三者最大值
    df['tr'] = df[['hl', 'hc', 'lc']].max(axis=1)
    
    # 计算 ATR (TR 的简单移动平均)
    # 注意：也可以使用加权移动平均(EWM)，这里使用标准的简单移动平均(Rolling Mean)
    df['atr'] = df['tr'].rolling(window=period).mean()
    
    # 返回最新一个时间点的 ATR 值
    current_atr = df['atr'].iloc[-1]
    
    return current_atr

def handle_data(context, data):
    """
    盘中运行函数
    """
    g.days += 1
    # 确保有足够的数据进行计算
    if g.days < g.ma_period + 2:
        return

    security = g.security
    
    # 1. 调用自定义函数计算 ATR
    atr_value = calculate_atr(security, g.atr_period)
    
    # 如果计算失败（数据不足），则跳过
    if atr_value is None or np.isnan(atr_value):
        log.info("ATR计算数据不足")
        return
        
    # 2. 获取均线数据作为趋势判断
    hist = get_history(g.ma_period, '1d', 'close', security, fq='pre', include=False)
    ma_value = hist['close'].mean()
    
    # 获取当前价格
    current_price = data[security]['close']
    
    # 获取当前持仓
    position = get_position(security)
    
    # 3. 交易逻辑示例
    # 策略逻辑：
    # - 当价格高于MA + 1倍ATR时，视为突破，买入
    # - 当价格低于MA - 1倍ATR时，视为跌破，卖出
    
    upper_band = ma_value + 1.0 * atr_value
    lower_band = ma_value - 1.0 * atr_value
    
    # 打印日志方便调试
    log.info("当前价格: %.2f, MA: %.2f, ATR: %.2f, 上轨: %.2f, 下轨: %.2f" % (
        current_price, ma_value, atr_value, upper_band, lower_band))
    
    # 买入逻辑
    if current_price > upper_band and position.amount == 0:
        log.info("价格突破上轨，买入")
        # 全仓买入
        order_value(security, context.portfolio.cash)
        
    # 卖出逻辑
    elif current_price < lower_band and position.amount > 0:
        log.info("价格跌破下轨，卖出")
        # 清仓
        order_target(security, 0)

3. 代码关键点解析

1. 数据获取 (get_history):

   • 在 calculate_atr 函数中，我们调用 get_history 获取了100条数据。虽然ATR周期是14，但计算TR需要前一日收盘价（会损失第一条数据），且移动平均需要足够的前置数据来填充窗口，因此获取的数据量必须远大于ATR周期。
   • 使用了 fq='pre'（前复权），这在计算涉及价格连续性的指标时非常重要，可以避免分红派息造成的缺口影响波动率计算。

2. Pandas 向量化计算:

   • 代码没有使用 for 循环来遍历每一天计算TR，而是利用 pandas 的列运算（Vectorization）。
   • df['close'].shift(1) 用于快速获取整列的前一日收盘价。
   • df[['hl', 'hc', 'lc']].max(axis=1) 用于按行取三个分量中的最大值。
   • 这种写法在Python中效率最高，也最简洁。

3. 兼容性处理:

   • 代码中加入了 if isinstance(h, dict): 的判断。这是因为在PTrade的不同版本或当 security 参数传入列表时，get_history 的返回结构可能是字典。这种写法增强了函数的健壮性。

4. 策略应用:

   • 示例策略展示了ATR最经典的用法之一：**肯特纳通道（Keltner Channel）**的变种。即利用均线加减ATR倍数构建通道，价格突破通道上轨做多，跌破下轨做空。

Q&A

Q: 为什么不直接使用 TA-Lib 库中的 ATR 函数？
A: PTrade 环境确实支持 TA-Lib 库。如果你习惯使用第三方库，也可以通过 import talib 后调用 talib.ATR 来计算。但手写 Pandas 版本的优势在于不依赖外部库的版本，且逻辑透明，方便你根据需要修改 TR 的定义（例如某些变种ATR算法）。

Q: get_history 中的 include=False 是什么意思？
A: include=False 表示获取的历史数据不包含当前正在进行的这一根K线（例如盘中运行时的当日K线）。在计算指标生成交易信号时，通常使用已完成的历史K线（昨天收盘）来避免“未来函数”或盘中信号闪烁的问题。如果你的策略需要基于盘中实时波动计算ATR，可以将其改为 True。

Q: 如何利用 ATR 进行动态止损？
A: 你可以在开仓时记录开仓价格 cost，然后设定止损价为 stop_price = cost - 2 * atr_value。在 handle_data 中，如果 current_price < stop_price 则触发止损。由于 ATR 会随市场波动变化，你也可以每个周期更新 ATR 值来移动止损线（追踪止损）。