如何在 PTrade 平台用 Python 完整复现经典的海龟交易法则？

这是一个基于**海龟交易法则（Turtle Trading Rules）**的PTrade策略实现。

策略核心逻辑说明

海龟交易法则是一个著名的趋势跟随策略，主要包含以下几个核心部分：

1. 市场选择：选择流动性高的标的（本代码以单只股票为例，可扩展为股票池）。
2. 仓位管理（N值）：使用ATR（平均真实波幅）来衡量市场波动性，定义为N。
   • 头寸规模（Unit）：每次开仓的风险控制在总资金的1%。计算公式：买入数量 = (总资金 * 0.01) / N。
3. 入场信号（唐奇安通道）：
   • 系统1（短期）：价格突破过去20天的最高价，买入1个Unit。
4. 加仓规则（金字塔式）：
   • 当价格在上次买入价格的基础上上涨 0.5 * N 时，加仓1个Unit。
   • 最大持仓限制为4个Unit。
5. 止损规则：
   • 当价格比最后一次买入价格下跌 2 * N 时，清仓止损。
6. 离场规则：
   • 当价格跌破过去10天的最低价时，止盈/离场。

PTrade 代码实现

以下代码适配 PTrade 股票回测环境（做多逻辑），兼容 Python 3.5 语法。

import numpy as np
import pandas as pd

def initialize(context):
    """
    策略初始化函数
    """
    # 1. 设置标的：这里以恒生电子为例，实际使用可修改或设置为股票池
    g.security = '600570.SS'
    set_universe(g.security)
    
    # 2. 海龟法则参数设置
    g.n_window = 20        # ATR计算周期 & 入场突破周期 (20日)
    g.exit_window = 10     # 离场周期 (10日)
    g.risk_ratio = 0.01    # 每次开仓风险占总资金比例 (1%)
    g.max_units = 4        # 最大允许持仓单元数
    
    # 3. 交易状态记录变量
    g.unit_count = 0       # 当前持仓单元数量
    g.last_entry_price = 0 # 上一次买入价格
    g.atr = 0              # 当前N值(ATR)

def get_atr(security, n_window):
    """
    计算ATR (N值)
    """
    # 获取历史数据：最高、最低、收盘。多取1天用于计算前一日收盘
    # 注意：为了计算准确，建议count设置大一些，这里取 n_window + 1
    h = get_history(n_window + 1, '1d', ['high', 'low', 'close'], security, fq='pre', include=False)
    
    if len(h) < n_window + 1:
        return 0
    
    high = h['high'].values
    low = h['low'].values
    close = h['close'].values
    
    # 计算真实波幅 TR
    # TR = Max(H-L, abs(H-PreC), abs(L-PreC))
    tr_list = []
    for i in range(1, len(close)):
        hl = high[i] - low[i]
        h_pc = abs(high[i] - close[i-1])
        l_pc = abs(low[i] - close[i-1])
        tr = max(hl, h_pc, l_pc)
        tr_list.append(tr)
    
    # 计算ATR (简单平均，也可以使用EMA平滑)
    if len(tr_list) == 0:
        return 0
    atr = np.mean(tr_list)
    return atr

def handle_data(context, data):
    """
    按分钟或日线运行的策略主函数
    """
    security = g.security
    
    # 1. 确保有数据
    if security not in data:
        return
    
    # 获取当前价格
    current_price = data[security]['close']
    
    # 2. 计算 N 值 (ATR)
    g.atr = get_atr(security, g.n_window)
    if g.atr == 0:
        return # 数据不足，跳过
        
    # 3. 获取唐奇安通道数据 (突破价格和离场价格)
    # 获取过去 N 天的历史数据用于判断突破
    hist = get_history(g.n_window, '1d', ['high', 'low'], security, fq='pre', include=False)
    if len(hist) < g.n_window:
        return
        
    # 唐奇安通道上轨 (过去20天最高价) - 入场信号
    donchian_high = hist['high'].max()
    # 唐奇安通道下轨 (过去10天最低价) - 离场信号 (注意切片取最后10天)
    donchian_low = hist['low'][-g.exit_window:].min()
    
    # 4. 获取当前持仓信息
    position = get_position(security)
    current_amount = position.amount
    
    # 如果当前没有持仓，重置计数器
    if current_amount == 0:
        g.unit_count = 0
        g.last_entry_price = 0
    
    # 5. 计算一个 Unit 对应的股数
    # 公式：Unit = (总资产 * 1%) / N
    total_value = context.portfolio.portfolio_value
    if g.atr > 0:
        unit_shares = int((total_value * g.risk_ratio) / g.atr)
        # 向下取整到100股
        unit_shares = int(unit_shares / 100) * 100
    else:
        unit_shares = 0

    # ---------------- 交易逻辑开始 ----------------
    
    # 情况 A: 持仓为 0，检查入场信号
    if g.unit_count == 0:
        # 突破20日高点，买入
        if current_price > donchian_high and unit_shares >= 100:
            log.info("触发入场信号: 当前价 %.2f > 20日高点 %.2f, ATR=%.2f" % (current_price, donchian_high, g.atr))
            order(security, unit_shares)
            g.unit_count = 1
            g.last_entry_price = current_price
            
    # 情况 B: 已有持仓，检查 加仓、止损、止盈
    else:
        # B1. 止损检查：价格下跌 2N
        stop_loss_price = g.last_entry_price - 2 * g.atr
        if current_price < stop_loss_price:
            log.info("触发止损: 当前价 %.2f < 止损价 %.2f (入场价-2N)" % (current_price, stop_loss_price))
            order_target(security, 0) # 清仓
            g.unit_count = 0
            return # 止损后本周期不再操作

        # B2. 离场检查：跌破10日低点 (止盈或普通离场)
        if current_price < donchian_low:
            log.info("触发离场: 当前价 %.2f < 10日低点 %.2f" % (current_price, donchian_low))
            order_target(security, 0) # 清仓
            g.unit_count = 0
            return # 离场后本周期不再操作
            
        # B3. 加仓检查：价格上涨 0.5N
        # 且持仓未达到最大限制
        if g.unit_count < g.max_units:
            add_entry_price = g.last_entry_price + 0.5 * g.atr
            if current_price > add_entry_price and unit_shares >= 100:
                # 检查现金是否足够
                if context.portfolio.cash >= unit_shares * current_price:
                    log.info("触发加仓: 当前价 %.2f > 上次入场+0.5N (%.2f)" % (current_price, add_entry_price))
                    order(security, unit_shares)
                    g.unit_count += 1
                    g.last_entry_price = current_price # 更新最后一次入场价用于计算下一次加仓和止损
                else:
                    log.info("资金不足，无法加仓")

代码详细解析

1. initialize (初始化):

   • 定义了全局变量 g，包括标的股票、N值窗口（20天）、离场窗口（10天）、风险比例（1%）和最大加仓次数（4次）。
   • 初始化了用于记录交易状态的变量：g.unit_count（当前持有几个单位）、g.last_entry_price（上一次成交价）。

2. get_atr (计算波动率):

   • 这是海龟法则的核心。它获取过去21天的最高、最低、收盘价。
   • 计算每日的 TR (True Range)：取 最高-最低、|最高-昨收|、|最低-昨收| 中的最大值。
   • 取 TR 的平均值作为 ATR (N值)。

3. handle_data (交易逻辑):

   • 计算 Unit 大小：根据账户总资产的1%除以ATR，计算出每次应该买多少股。这保证了无论市场波动大小，每次交易承担的风险金额是固定的。
   • 入场 (System 1)：如果当前没有持仓，且价格突破了过去20天的最高价 (donchian_high)，买入 1 个 Unit。
   • 加仓 (Pyramiding)：如果已有持仓，且价格比上一次买入价上涨了 0.5 * N，并且持仓未满4个 Unit，继续买入 1 个 Unit。
   • 止损 (Stop Loss)：海龟法则极其重视风控。如果价格跌破 最后一次买入价 - 2 * N，立即清仓止损。
   • 离场 (Exit)：如果价格跌破过去10天的最低价 (donchian_low)，视为趋势结束，清仓离场。

注意事项

• 做空逻辑：经典海龟法则包含做空（跌破20日低点做空）。由于A股普通账户只能做多，上述代码仅实现了做多逻辑。如果您的账户支持融券，可以使用 marginsec_open 替换 order 并反转逻辑来实现做空。
• 数据频率：此策略通常在日线级别运行。如果在分钟级别回测，handle_data 会每分钟运行，但逻辑中使用的 get_history 依然是获取 '1d' (日线) 数据，这是符合海龟法则定义的（基于日间趋势）。
• 资金管理：代码中包含了资金检查，防止在加仓时因资金不足导致报错。

Q&A

Q: 为什么计算ATR时要取 n_window + 1 天的数据？
A: 计算真实波幅（TR）需要用到“前一日收盘价”。为了计算完整的20天TR，我们需要获取21天的历史数据，这样第2天到第21天才能每一天都算出TR值。

Q: 如何调整策略的灵敏度？
A: 修改 initialize 中的 g.n_window。将其改为 10 或 55 可以改变突破的判定周期（55日突破即海龟法则中的系统2）。

Q: 为什么加仓后止损价是基于“最后一次买入价”计算的？
A: 这是海龟法则的特点。随着价格上涨并加仓，止损线也随之上移（Trailing Stop），这样可以保护已有的利润，同时限制新增头寸的风险。