用TB编写商品期货趋势型策略（下）

小象实战讲义 · 进阶量化交易实战：迭代式的量化策略研发 本讲义仅用于编程与量化方法教学，不构成任何投资建议。

单次满仓在趋势行情中无法实现利润最大化。本节深入剖析海龟交易法则中的金字塔加仓机制（Pyramid Position Sizing）与动态移动止损在 TB 与 Python 中的等价实现。本讲义原课以 TB/MC 演示，此处为 Python 等价实现。

💡 核心导读

• 海龟金字塔加仓与止损核心体系：
  • 初始建仓（Initial Unit）：价格突破 20 日高点开仓 1 个 Unit；
  • 金字塔加仓（Pyramid Adding）：价格每顺势上涨 $0.5 \times ATR$，追加 1 个 Unit，最多加仓至 4 个 Units；
  • 动态止损同步提升（Sync Stop-loss）：每完成一次加仓，所有已持仓 Units 的止损线统一上移至最新加仓价之下 $2 \times ATR$。

┌─────────────────────────────────────────────────────────────┐
│                   海龟金字塔加仓与止损同步上移拓扑图        │
└──────────────────────────────┬──────────────────────────────┘
                               │
 ┌─────────────────────────────┼─────────────────────────────┐
 ▼                             ▼                             ▼
┌─────────────────────────┐   ┌─────────────────────────┐   ┌─────────────────────────┐
│ 初始突破: 1 Unit        │   │ 加仓 2/3/4 Units        │   │ 止损线整体上移          │
├─────────────────────────┤   ├─────────────────────────┤   ├─────────────────────────┤
│ • 价格 > 20日最高价     │──►│ • 每涨 0.5 * ATR 加 1手 │──►│• 统一移至最新买价 - 2ATR│
│ • 止损 = 买入价 - 2*ATR │   │ • 最大持仓 4 Units 上限 │   │ • 保护全部已持仓浮盈    │
└─────────────────────────┘   └─────────────────────────┘   └─────────────────────────┘

一、海龟金字塔加仓数学状态机推导

声明：原课以 TB/MC 演示，此处为 Python 等价实现。

设第 $k$ 次加仓价格为 $P_k$，$k \in \{1, 2, 3, 4\}$。加仓触发条件与最新统一止损线为：

$$P_k = P_{k-1} + 0.5 \times ATR, \quad \text{StopPrice}_k = P_k - 2.0 \times ATR$$

import pandas as pd
import numpy as np

class TurtlePyramidStrategy:
    """
    海龟金字塔加仓状态机（Python 等价实现，合成教学样例，实盘需考虑滑点、流动性及强制平仓风险）
    """
    def __init__(self, max_units: int = 4):
        self.max_units = max_units
        self.units = 0
        self.entry_prices = []
        self.stop_price = 0.0
        
    def on_bar(self, price: float, upper_20: float, atr: float) -> list:
        actions = []
        # 1. 初始突破
        if self.units == 0 and price > upper_20:
            self.units = 1
            self.entry_prices.append(price)
            self.stop_price = price - 2.0 * atr
            actions.append(('BUY_INITIAL', price, self.stop_price))
        # 2. 金字塔加仓
        elif 0 < self.units < self.max_units and price >= self.entry_prices[-1] + 0.5 * atr:
            self.units += 1
            self.entry_prices.append(price)
            self.stop_price = price - 2.0 * atr
            actions.append(('BUY_ADD', price, self.stop_price))
        # 3. 动态止损触发
        elif self.units > 0 and price < self.stop_price:
            actions.append(('SELL_ALL', price, 0))
            self.units = 0
            self.entry_prices.clear()
            self.stop_price = 0.0
            
        return actions

# 合成数据测试状态机
rng = np.random.default_rng(seed=42)
# 模拟价格序列：先上涨触发加仓，后下跌触发止损
prices = 4000.0 + np.cumsum(rng.normal(loc=5, scale=15, size=30))
atr = 20.0
upper_20 = 4050.0

turtle = TurtlePyramidStrategy(max_units=4)
all_actions = []

for i, p in enumerate(prices):
    acts = turtle.on_bar(p, upper_20, atr)
    if acts:
        all_actions.extend(acts)
    # 打印关键状态变化
    if acts or i == 0 or i == len(prices)-1:
        print(f"Bar {i:2d}: Price={p:7.2f}, Units={turtle.units}, Stop={turtle.stop_price:.2f}, Actions={acts}")

print("\n===== 海龟金字塔状态机执行摘要 =====")
print(f"总动作次数: {len(all_actions)}")
print(f"最终持仓单位数: {turtle.units}")
print(f"入场价格列表: {turtle.entry_prices}")

Bar  0: Price=4009.57, Units=0, Stop=0.00, Actions=[]
Bar 13: Price=4062.42, Units=1, Stop=4022.42, Actions=[('BUY_INITIAL', np.float64(4062.4212309614345), np.float64(4022.4212309614345))]
Bar 14: Price=4074.43, Units=2, Stop=4034.43, Actions=[('BUY_ADD', np.float64(4074.433871095215), np.float64(4034.433871095215))]
Bar 18: Price=4085.87, Units=3, Stop=4045.87, Actions=[('BUY_ADD', np.float64(4085.8692614203783), np.float64(4045.8692614203783))]
Bar 22: Price=4110.47, Units=4, Stop=4070.47, Actions=[('BUY_ADD', np.float64(4110.471614216457), np.float64(4070.471614216457))]
Bar 29: Price=4157.57, Units=4, Stop=4070.47, Actions=[]

===== 海龟金字塔状态机执行摘要 =====
总动作次数: 4
最终持仓单位数: 4
入场价格列表: [np.float64(4062.4212309614345), np.float64(4074.433871095215), np.float64(4085.8692614203783), np.float64(4110.471614216457)]

二、加仓风险敞口与资本保护验证

头寸由 1 个 Unit 加到 4 个 Units 后，尽管止损线随加仓不断抬升，反转触损时各 Unit 的亏损并不相同：最后加仓的 Unit 4 亏损 $2.0 \times ATR$，Unit 3 / 2 / 1 依次亏损 $1.5 / 1.0 / 0.5 \times ATR$。因此满仓 4 Units 最大亏损为 $5 \times ATR \times \text{unit\_size}$（各 Unit 亏损之和），是初始 Unit 风险的 2.5 倍——加仓在放大利润的同时也放大了极端风险敞口，资金与心理预期都必须按此口径准备。

import numpy as np

def verify_turtle_risk_bound(atr: float, unit_size: int) -> float:
    """
    验证 4 个 Unit 满仓时的最大理论亏损绝对值（合成教学样例，实盘亏损可能因跳空、流动性而超出理论值）
    """
    # Unit 4: 亏损 2.0 ATR; Unit 3: 亏损 1.5 ATR; Unit 2: 亏损 1.0 ATR; Unit 1: 亏损 0.5 ATR
    # 总亏损 = (2.0 + 1.5 + 1.0 + 0.5) * ATR * unit_size = 5.0 * ATR * unit_size
    max_risk = 5.0 * atr * unit_size # 满仓 4 Units 最大亏损 5×ATR×unit_size
    return max_risk

# 合成参数测试
test_atr = 25.0
test_unit_size = 10
risk = verify_turtle_risk_bound(test_atr, test_unit_size)

print("===== 海龟策略满仓风险边界验证 =====")
print(f"ATR 值: {test_atr}")
print(f"每 Unit 合约数量: {test_unit_size}")
print(f"满仓 (4 Units) 最大理论亏损绝对值: {risk:.2f}")
print(f"相当于 {risk / (test_atr * test_unit_size):.1f} 倍 ATR×UnitSize")
print("注：理论计算未考虑滑点与跳空风险。")

===== 海龟策略满仓风险边界验证 =====
ATR 值: 25.0
每 Unit 合约数量: 10
满仓 (4 Units) 最大理论亏损绝对值: 1250.00
相当于 5.0 倍 ATR×UnitSize
注：理论计算未考虑滑点与跳空风险。

📝 动手练一练

编写 Python 测试脚本，在单边大牛市行情中模拟金字塔加仓并检验 4 个 Units 满仓执行：

import pandas as pd
import numpy as np

class TurtlePyramidStrategy:
    """
    海龟金字塔加仓状态机（Python 等价实现，合成教学样例，实盘需考虑滑点、流动性及强制平仓风险）
    """
    def __init__(self, max_units: int = 4):
        self.max_units = max_units
        self.units = 0
        self.entry_prices = []
        self.stop_price = 0.0
        
    def on_bar(self, price: float, upper_20: float, atr: float) -> list:
        actions = []
        # 1. 初始突破
        if self.units == 0 and price > upper_20:
            self.units = 1
            self.entry_prices.append(price)
            self.stop_price = price - 2.0 * atr
            actions.append(('BUY_INITIAL', price, self.stop_price))
        # 2. 金字塔加仓
        elif 0 < self.units < self.max_units and price >= self.entry_prices[-1] + 0.5 * atr:
            self.units += 1
            self.entry_prices.append(price)
            self.stop_price = price - 2.0 * atr
            actions.append(('BUY_ADD', price, self.stop_price))
        # 3. 动态止损触发
        elif self.units > 0 and price < self.stop_price:
            actions.append(('SELL_ALL', price, 0))
            self.units = 0
            self.entry_prices.clear()
            self.stop_price = 0.0
            
        return actions

def run_turtle_pyramid_test():
    rng = np.random.default_rng(seed=42)
    # 模拟单边上涨 100 根 K 线
    prices = 4000.0 + np.linspace(0, 500, 100)
    atr = 20.0
    upper_20 = 4050.0
    
    turtle = TurtlePyramidStrategy(max_units=4)
    all_actions = []
    
    for p in prices:
        acts = turtle.on_bar(p, upper_20, atr)
        if acts:
            all_actions.extend(acts)
            
    print("================ 海龟金字塔加仓状态机测试 ================")
    print(f"策略执行动作总数: {len(all_actions)} 次")
    for act in all_actions:
        print(f"  - 执行动作: {act[0]:<12} | 触发价格: {act[1]:.2f} | 动态止损线: {act[2]:.2f}")
    assert turtle.units == 4, "金字塔加仓未能加满 4 个 Units！"
    print("✅ 海龟金字塔加仓与止损同步机制验证 PASS")
    print("========================================================")

run_turtle_pyramid_test()

运行结果输出：

================ 海龟金字塔加仓状态机测试 ================
策略执行动作总数: 4 次
  - 执行动作: BUY_INITIAL  | 触发价格: 4050.51 | 动态止损线: 4010.51
  - 执行动作: BUY_ADD      | 触发价格: 4060.61 | 动态止损线: 4020.61
  - 执行动作: BUY_ADD      | 触发价格: 4070.71 | 动态止损线: 4030.71
  - 执行动作: BUY_ADD      | 触发价格: 4080.81 | 动态止损线: 4040.81
✅ 海龟金字塔加仓与止损同步机制验证 PASS
========================================================

本章小结

1. 顺势浮盈加仓扩大战果：只在已有仓位产生浮盈时才允许追加头寸，严禁在亏损时逆势摊平。
2. 止损线同步抬升控险：每加仓一次，所有持仓止损线统一上移，使整体组合风险敞口始终有明确可量化的上限（满仓 4 Units 为 $5 \times ATR \times \text{unit\_size}$）。
3. 设定严格持仓 Units 上限：单品种最高 4 Units，全市场总仓位不超过 12 Units，防范黑天鹅集中爆发。

📋 行动清单 (Action Checklist)

• 运行金字塔加仓测试代码，观察在剧烈反转行情下的平仓点位；
• 编写跟踪止盈在加仓策略中的集成实现；
• 预习下一节，掌握商品期货趋势策略的参数优化与评价思路。

—— 小象教研组