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

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

TradeBlazer（TB）是国内商品期货量化交易的经典平台。本节深入剖析 TB 经典突破策略（Dual Thrust / 海龟通道）的核心逻辑框架，并用现代化 Python 3.12 给出完全等价、无未来函数的工业级实现。本讲义原课以 TB/MC 演示，此处为 Python 等价实现。

💡 核心导读

• TB 经典突破策略 Dual Thrust 核心数学逻辑：
  • 波幅范围（Range）：$\text{Range} = \max(HH - LC, HC - LL)$，综合过去 $N$ 日高低价与收盘价极差；
  • 动态触发线（Upper / Lower Triggers）： $$Upper = \text{Open}_t + K_1 \times \text{Range}, \quad Lower = \text{Open}_t - K_2 \times \text{Range}$$
  • 非对称系数设计：通过设定 $K_1 \ne K_2$ 调节多空偏好，适应不同品种多空不对称性。

┌─────────────────────────────────────────────────────────────┐
│                 TB 经典 Dual Thrust 策略逻辑拓扑图          │
└──────────────────────────────┬──────────────────────────────┘
                               │
 ┌─────────────────────────────┼─────────────────────────────┐
 ▼                             ▼                             ▼
┌─────────────────────────┐   ┌─────────────────────────┐   ┌─────────────────────────┐
│ 当日开盘价基准 (Open)   │   │ 计算前 N 日波幅 (Range) │   │ 突破触发与开平仓        │
├─────────────────────────┤   ├─────────────────────────┤   ├─────────────────────────┤
│ • 以当日 Open 为中轴    │   │ • Range = Max(HH-LC, ..)│   │ • 突破 Upper: 开多/平空 │
│ • 消除隔夜跳空影响      │   │ • $K_1, K_2$ 参数调节   │   │ • 跌破 Lower: 开空/平多 │
└─────────────────────────┘   └─────────────────────────┘   └─────────────────────────┘

一、TB 原始代码解析与 Python 等价算法实现

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

TB 源码使用 Open + K1 * Range 生成日内突破信号。在 Python 向量化中，我们必须将过去 $N$ 日的极差严格进行 shift(1)，避免引入当日价格：

$$\text{HH} = \max_{1 \le i \le N} High_{t-i}, \quad \text{LC} = \min_{1 \le i \le N} Close_{t-i}$$

import pandas as pd
import numpy as np

def calculate_dual_thrust_signals(df: pd.DataFrame, n_days: int = 5, k1: float = 0.5, k2: float = 0.5) -> pd.DataFrame:
    """
    TB Dual Thrust 策略 Python 工业级等价实现（无未来函数）
    注意：合成数据回测不代表实盘，存在过拟合与幸存者偏差风险。
    """
    res = df.copy()
    hh = res['high'].rolling(n_days).max().shift(1)
    hc = res['close'].rolling(n_days).max().shift(1)
    ll = res['low'].rolling(n_days).min().shift(1)
    lc = res['close'].rolling(n_days).min().shift(1)
    
    range_val = np.maximum(hh - lc, hc - ll)
    
    res['upper_bound'] = res['open'] + k1 * range_val
    res['lower_bound'] = res['open'] - k2 * range_val
    
    res['signal'] = 0
    res.loc[res['close'] > res['upper_bound'], 'signal'] = 1
    res.loc[res['close'] < res['lower_bound'], 'signal'] = -1
    return res

# 合成期货日K数据（模拟螺纹钢主力合约，合约乘数10吨/手，保证金比例10%）
rng = np.random.default_rng(42)
base_price = 4000.0
price_changes = rng.normal(1.5, 8.0, 30)
prices = base_price + np.cumsum(price_changes)
opens = prices + rng.uniform(-12, 12, 30)
df_demo = pd.DataFrame({
    'open': opens,
    'high': np.maximum(opens, prices) + rng.uniform(2, 8, 30),
    'low': np.minimum(opens, prices) - rng.uniform(2, 8, 30),
    'close': prices
})

# 计算信号
result = calculate_dual_thrust_signals(df_demo, n_days=5, k1=0.5, k2=0.5)
print("===== Dual Thrust 信号生成（期货合成数据） =====")
print(f"数据周期: {len(result)} 个交易日")
print(f"上轨触发次数: {(result['signal'] == 1).sum()}")
print(f"下轨触发次数: {(result['signal'] == -1).sum()}")
print(f"无信号次数: {(result['signal'] == 0).sum()}")
print("前5行信号预览:")
print(result[['open', 'high', 'low', 'close', 'upper_bound', 'lower_bound', 'signal']].head())

===== Dual Thrust 信号生成（期货合成数据） =====
数据周期: 30 个交易日
上轨触发次数: 1
下轨触发次数: 0
无信号次数: 29
前5行信号预览:
          open         high          low  ...  upper_bound  lower_bound  signal
0  4009.812028  4015.822446  4001.023864  ...          NaN          NaN       0
1  4008.338097  4013.164675  3990.939942  ...          NaN          NaN       0
2  4000.441282  4010.012890  3995.764344  ...          NaN          NaN       0
3  4010.537024  4020.235984  4006.250897  ...          NaN          NaN       0
4  3998.807049  4005.346019  3994.997977  ...          NaN          NaN       0

[5 rows x 7 columns]

二、开平仓状态机与反手交易执行

Dual Thrust 属于典型的连续反手策略（Always-in-the-market）。当持有多头时触碰下轨，立即执行“平多并开空”操作。

import pandas as pd
import numpy as np

def execute_dual_thrust_backtest(df_signals: pd.DataFrame) -> pd.Series:
    """
    执行反手交易回测（期货合成数据，未考虑手续费与滑点，存在前视偏差风险）
    """
    df = df_signals.copy()
    df['pos'] = df['signal'].replace(0, np.nan).ffill().shift(1).fillna(0)
    df['ret'] = df['close'].pct_change().fillna(0)
    df['strat_ret'] = df['pos'] * df['ret']
    return (1.0 + df['strat_ret']).cumprod()

# 构造包含信号的合成数据
rng = np.random.default_rng(123)
n_bars = 50
base = 100.0
price_walk = base + np.cumsum(rng.normal(0.1, 2.0, n_bars))
df_test = pd.DataFrame({
    'open': price_walk + rng.uniform(-1, 1, n_bars),
    'high': price_walk + rng.uniform(1, 3, n_bars),
    'low': price_walk - rng.uniform(1, 3, n_bars),
    'close': price_walk,
    'signal': 0
})
# 手动设置一些信号点
signal_idx = [5, 15, 25, 35, 45]
df_test.loc[signal_idx, 'signal'] = rng.choice([-1, 1], size=len(signal_idx))

# 执行回测
equity_curve = execute_dual_thrust_backtest(df_test)
print("===== 反手交易回测（合成数据） =====")
print(f"回测区间: {len(equity_curve)} 根K线")
print(f"初始净值: {equity_curve.iloc[0]:.4f}")
print(f"最终净值: {equity_curve.iloc[-1]:.4f}")
print(f"期间最大回撤（近似）: {(equity_curve.cummax() - equity_curve).max():.4f}")
print("净值曲线前10个值:")
print(equity_curve.head(10).round(4))

===== 反手交易回测（合成数据） =====
回测区间: 50 根K线
初始净值: 1.0000
最终净值: 0.9638
期间最大回撤（近似）: 0.1096
净值曲线前10个值:
0    1.0000
1    1.0000
2    1.0000
3    1.0000
4    1.0000
5    1.0000
6    1.0113
7    0.9996
8    1.0048
9    1.0101
Name: strat_ret, dtype: float64

📝 动手练一练

编写 Python 仿真测试，在模拟商品期货 K 线上运行 Dual Thrust 策略并验证信号。

⚠️ 失败案例警示（本节反面教材）：本节仿真最初的参数组合为「日波动 $\sigma=20$ + $K_1=0.4, K_2=0.6$」，5 日 Range 中位数约 57 点，触发线被推到开盘价 $\pm 23 \sim 34$ 点之外，而收盘价相对开盘价的偏离被限制在 $\pm 5$ 点以内——通道过宽导致 100 根 K 线全程 0 信号、净值恒为 1.0000。更危险的是，旧断言只校验 equity > 0，这种「零交易回测」照样打印 PASS，属于典型的静默失效（Silent Failure）：回测什么也没验证，却给了人策略可行的错觉。修正方案：收窄模拟波动（$\sigma=8$）、放宽开盘跳空幅度（$\pm 12$）并调低 $K$ 值（0.3 / 0.4），让触发线进入收盘价可达区间；同时新增信号数断言，从机制上防止回归：

import pandas as pd
import numpy as np

def calculate_dual_thrust_signals(df: pd.DataFrame, n_days: int = 5, k1: float = 0.5, k2: float = 0.5) -> pd.DataFrame:
    """Dual Thrust 信号生成（本块内完整定义）"""
    res = df.copy()
    hh = res['high'].rolling(n_days).max().shift(1)
    hc = res['close'].rolling(n_days).max().shift(1)
    ll = res['low'].rolling(n_days).min().shift(1)
    lc = res['close'].rolling(n_days).min().shift(1)
    range_val = np.maximum(hh - lc, hc - ll)
    res['upper_bound'] = res['open'] + k1 * range_val
    res['lower_bound'] = res['open'] - k2 * range_val
    res['signal'] = 0
    res.loc[res['close'] > res['upper_bound'], 'signal'] = 1
    res.loc[res['close'] < res['lower_bound'], 'signal'] = -1
    return res

def execute_dual_thrust_backtest(df_signals: pd.DataFrame) -> pd.Series:
    """反手交易回测（本块内完整定义）"""
    df = df_signals.copy()
    df['pos'] = df['signal'].replace(0, np.nan).ffill().shift(1).fillna(0)
    df['ret'] = df['close'].pct_change().fillna(0)
    df['strat_ret'] = df['pos'] * df['ret']
    return (1.0 + df['strat_ret']).cumprod()

def run_dual_thrust_test():
    """完整策略测试流程（合成数据，历史回测不代表未来收益）"""
    rng = np.random.default_rng(42)
    prices = 4000.0 + np.cumsum(rng.normal(1.5, 8.0, 100))
    opens = prices + rng.uniform(-12, 12, 100)
    df = pd.DataFrame({
        'open': opens,
        'high': np.maximum(opens, prices) + rng.uniform(2, 8, 100),
        'low': np.minimum(opens, prices) - rng.uniform(2, 8, 100),
        'close': prices
    })
    
    res = calculate_dual_thrust_signals(df, n_days=5, k1=0.3, k2=0.4)
    equity = execute_dual_thrust_backtest(res)
    
    print("================ TB Dual Thrust 策略 Python 等价实现 ================")
    print(f"有效回测 Bar 数  : {len(df)}")
    print(f"上轨触发买入次数 : {(res['signal'] == 1).sum()} 次")
    print(f"下轨触发卖出次数 : {(res['signal'] == -1).sum()} 次")
    print(f"策略最终累计净值 : {equity.iloc[-1]:.4f}")
    print(f"基准（买入持有）净值 : {(df['close'].iloc[-1] / df['close'].iloc[0]):.4f}")
    # 设计型断言：确保通道宽度设置合理，有信号产生
    assert (res['signal'] != 0).sum() > 0, "通道过宽导致全程无信号，回测静默失效！"
    assert equity.iloc[-1] > 0, "回测净值计算异常！"
    print("✅ TB Dual Thrust 等价复现与回测验证 PASS")
    print("====================================================================")

run_dual_thrust_test()

运行结果输出：

================ TB Dual Thrust 策略 Python 等价实现 ================
有效回测 Bar 数  : 100
上轨触发买入次数 : 16 次
下轨触发卖出次数 : 4 次
策略最终累计净值 : 1.0151
基准（买入持有）净值 : 1.0264
✅ TB Dual Thrust 等价复现与回测验证 PASS
====================================================================

本章小结

1. TB 经典策略完美映射：采用 Python 向量化严格对齐 Dual Thrust 算法，消除对商业封闭平台的依赖。
2. 非对称多空参数设计：通过 $K_1$ 与 $K_2$ 调节多空灵敏度，适应期货各品种独特的波动结构。
3. 开盘价锚定规避跳空：以当日 Open 叠加历史 Range 为上下轨，有效降低隔夜跳空造成的滑点冲击。
4. 警惕零交易的静默失效：触发通道相对价格可达区间过宽时，回测会全程无信号、净值恒为 1.0，而只检查净值的断言照样 PASS；必须先断言信号次数大于 0，回测结论才有意义。

📋 行动清单 (Action Checklist)

• 运行 Dual Thrust 测试脚本，测试修改 $K_1, K_2$ 对策略胜率与交易频次的影响；
• 为 Dual Thrust 增加 ATR 波动率止损出场条件；
• 预习下一节，掌握用 TB/Python 编写带金字塔加仓的高阶趋势策略。

—— 小象教研组