PTrade 教程：除了内置指标，如何编写自己的技术指标函数？

在PTrade量化交易平台中，虽然系统提供了许多内置的技术指标函数（如 get_MACD, get_KDJ 等），但在实际策略开发中，您经常需要编写自定义的指标逻辑。

创建和使用自定义技术指标的核心流程分为三步：

1. 获取数据：使用 get_history 获取足够的历史K线数据。
2. 计算指标：利用 Python 的 pandas 或 numpy 库对数据进行数学运算。
3. 生成信号：根据计算出的指标值与当前价格对比，生成买卖信号。

以下是详细的步骤说明和一个完整的策略示例。

核心步骤说明

1. 获取历史数据

使用 get_history 接口。注意：获取的数据长度（count）必须大于您计算指标所需的周期。例如，计算20日均线，建议至少获取25-30条数据，以防数据缺失或计算偏差。

# 获取过去30天的收盘价
history_data = get_history(30, '1d', 'close', security_list=g.security)

2. 编写计算逻辑

PTrade 环境支持 pandas 和 numpy，这是处理时间序列数据的最佳工具。您可以利用 rolling（滚动窗口）函数来计算均值、标准差等。

# 示例：计算布林带
# 计算20日均线 (中轨)
mid_line = history_data['close'].rolling(window=20).mean()
# 计算20日标准差
std_dev = history_data['close'].rolling(window=20).std()
# 计算上轨和下轨
upper_line = mid_line + 2 * std_dev
lower_line = mid_line - 2 * std_dev

3. 整合到策略中

将上述逻辑封装成函数，或者直接写在 handle_data 中，结合 order 函数进行交易。

完整策略代码示例

下面是一个完整的策略示例。该策略实现了一个自定义的布林带（Bollinger Bands）策略。

• 逻辑：当价格突破下轨时买入（看多反转），当价格突破上轨时卖出（看空回调）。
• 自定义部分：我们没有使用库函数，而是通过 pandas 手动计算了中轨、上轨和下轨。

import numpy as np
import pandas as pd

def initialize(context):
    """
    初始化函数，设置股票池和全局变量
    """
    # 设置我们要操作的股票，这里以 600570.SS (恒生电子) 为例
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 设定布林带的参数
    g.n_period = 20  # 均线周期
    g.k_std = 2      # 标准差倍数

def calculate_custom_bollinger(history_df, n, k):
    """
    自定义技术指标计算函数：布林带
    输入：
        history_df: 包含 'close' 列的 DataFrame
        n: 周期
        k: 标准差倍数
    输出：
        upper: 上轨值 (float)
        middle: 中轨值 (float)
        lower: 下轨值 (float)
    """
    # 提取收盘价序列
    close_series = history_df['close']
    
    # 利用 pandas 的 rolling 函数计算移动平均线 (中轨)
    middle_series = close_series.rolling(window=n).mean()
    
    # 计算标准差
    std_series = close_series.rolling(window=n).std()
    
    # 计算上轨和下轨
    upper_series = middle_series + k * std_series
    lower_series = middle_series - k * std_series
    
    # 返回最新的一个数据点（即当前周期的指标值）
    # 使用 iloc[-1] 获取序列中最后一个值
    return upper_series.iloc[-1], middle_series.iloc[-1], lower_series.iloc[-1]

def handle_data(context, data):
    """
    盘中运行函数，每个周期执行一次
    """
    security = g.security
    
    # 1. 获取历史数据
    # 我们需要计算20日均线，为了安全起见，获取过去 30 天的数据
    # include=True 表示包含当前这根未走完的K线（如果是日线回测，通常指当日收盘价，实盘为最新价）
    # 注意：如果是在盘中实时计算，建议 include=False 使用昨日收盘价计算指标，避免信号闪烁；
    # 或者 include=True 使用最新价计算。这里演示使用 include=True。
    count = g.n_period + 5
    history_data = get_history(count, '1d', 'close', security, include=True)
    
    # 检查数据长度是否足够，不够则不计算，防止报错
    if len(history_data) < g.n_period:
        log.info("历史数据不足，跳过计算")
        return

    # 2. 调用自定义函数计算指标
    upper_band, mid_band, lower_band = calculate_custom_bollinger(history_data, g.n_period, g.k_std)
    
    # 获取当前最新价格
    current_price = data[security]['close']
    
    # 获取当前持仓
    position = get_position(security)
    curr_amount = position.amount
    
    # 打印日志方便调试
    log.info("当前价格: %.2f, 上轨: %.2f, 下轨: %.2f" % (current_price, upper_band, lower_band))
    
    # 3. 交易逻辑
    # 策略逻辑：价格跌破下轨，视为超卖，买入
    if current_price < lower_band:
        if curr_amount == 0:
            # 全仓买入 (使用所有可用资金)
            cash = context.portfolio.cash
            if cash > 0:
                order_value(security, cash)
                log.info("价格跌破下轨，买入 %s" % security)
    
    # 策略逻辑：价格突破上轨，视为超买，卖出
    elif current_price > upper_band:
        if curr_amount > 0:
            # 卖出所有持仓
            order_target(security, 0)
            log.info("价格突破上轨，卖出 %s" % security)

关键点解析

1. 数据获取 (get_history):

   • count 参数非常重要。如果您需要计算 N 日的指标，count 必须大于等于 N。
   • include 参数决定了是否包含当前正在进行的这根K线。在回测中，如果 include=True，使用的是当天的收盘价（未来函数风险较低，因为回测是在收盘后结算）；在实盘中，include=True 意味着使用最新成交价，指标值会随价格跳动而变化。

2. Pandas 的使用:

   • PTrade 的 get_history 返回的是 pandas.DataFrame (当查询单只股票时) 或 pandas.Panel (旧版多只股票) 或 MultiIndex DataFrame (新版多只股票)。
   • 使用 df['column'].rolling(window=N).mean() 是计算移动平均最高效的方法，避免了使用 Python for 循环，计算速度更快。

3. Python 版本兼容性:

   • 代码中使用了 % 格式化字符串 (例如 "%.2f" % value) 而不是 f-string，这是为了确保代码在 PTrade 可能使用的旧版本 Python (如 Python 3.5) 中也能正常运行。

4. 模块化:

   • 将计算逻辑提取为 calculate_custom_bollinger 函数。这样做的好处是代码结构清晰，且方便在 handle_data 之外的地方（如 before_trading_start）复用。

常见问题 (Q&A)

Q: 我可以使用 TA-Lib 库吗？
A: 是的，PTrade 环境通常预装了 TA-Lib 库。如果您不想自己写公式，可以直接 import talib 并传入 numpy 数组进行计算。但上述示例展示的是如何完全“自定义”逻辑，这对于非标准指标非常有用。

Q: 如何计算多只股票的自定义指标？
A: get_history 支持传入股票列表。返回的数据结构会略有不同（通常是 MultiIndex DataFrame）。您可以使用 for 循环遍历股票代码，或者利用 Pandas 的 groupby 功能进行分组计算。

Q: 为什么我的指标计算结果是 NaN？
A: 通常是因为获取的历史数据长度不足。例如计算20日均线，数据的前19个点因为没有足够的前置数据，计算结果都会是 NaN。请确保 get_history 的 count 参数足够大。

Q: 可以在盘前 (before_trading_start) 计算指标吗？
A: 可以。对于日线级别的策略，通常建议在盘前计算好昨日的指标值，盘中仅做信号比对。这样可以减少盘中的计算压力。