如何使用Python實現Dual Thrust 區間突破策略

# 如何使用Python實現Dual Thrust區間突破策略

## 目錄
1. [策略概述](#策略概述)
2. [數學原理](#數學原理)
3. [數據準備](#數據準備)
4. [核心代碼實現](#核心代碼實現)
5. [回測與優化](#回測與優化)
6. [實盤注意事項](#實盤注意事項)
7. [完整代碼示例](#完整代碼示例)

---

## 策略概述
Dual Thrust是由Michael Chalek在1980年代開發的經典趨勢跟蹤策略，其核心思想是通過動態計算價格通道，在突破上軌時做多，突破下軌時做空。該策略具有以下特點：

- **多空對稱**：通過參數k1/k2控制多空觸發閾值
- **自適應波動**：基于歷史價格波動范圍計算交易區間
- **趨勢跟蹤**：適合趨勢明顯的市場環境

![Dual Thrust示意圖](https://example.com/dual_thrust.png)

---

## 數學原理
### 關鍵指標計算
1. **價格區間計算**：

HH = 最高價的N日最高 HC = 收盤價的N日最高 LC = 收盤價的N日最低 LL = 最低價的N日最低

2. **波動范圍計算**：

Range = Max(HH - LC, HC - LL)

3. **上下軌計算**：

上軌 = 開盤價 + k1 * Range 下軌 = 開盤價 - k2 * Range

   （典型參數：N=5, k1=0.5, k2=0.5）

### 交易信號
| 條件 | 操作 |
|------|------|
| 價格 > 上軌 | 開多/平空 |
| 價格 < 下軌 | 開空/平多 |

---

## 數據準備
使用`pandas`處理OHLC數據：

```python
import pandas as pd
import yfinance as yf

# 獲取數據
def fetch_data(symbol, start, end):
    df = yf.download(symbol, start, end)
    df = df[['Open', 'High', 'Low', 'Close']]
    return df.dropna()

# 示例數據
data = fetch_data('AAPL', '2020-01-01', '2023-12-31')

數據預處理：

def preprocess_data(df, n=5):
    df['HH'] = df['High'].rolling(n).max()  # N日最高價
    df['HC'] = df['Close'].rolling(n).max()  # N日收盤最高
    df['LC'] = df['Close'].rolling(n).min()  # N日收盤最低
    df['LL'] = df['Low'].rolling(n).min()   # N日最低價
    return df.dropna()

核心代碼實現

1. 計算交易區間

def calculate_range(df, k1=0.5, k2=0.5):
    df['Range'] = df[['HH','HC','LC','LL']].apply(
        lambda x: max(x['HH']-x['LC'], x['HC']-x['LL']), axis=1)
    df['Upper'] = df['Open'] + k1 * df['Range']
    df['Lower'] = df['Open'] - k2 * df['Range']
    return df

2. 生成交易信號

def generate_signals(df):
    df['Signal'] = 0
    df.loc[df['Close'] > df['Upper'], 'Signal'] = 1    # 做多信號
    df.loc[df['Close'] < df['Lower'], 'Signal'] = -1   # 做空信號
    return df

3. 策略邏輯流程圖

graph TD
    A[獲取歷史數據] --> B[計算HH/HC/LC/LL]
    B --> C[計算Range和上下軌]
    C --> D[生成交易信號]
    D --> E[執行回測]

回測與優化

回測框架實現

import numpy as np

def backtest(df, initial_capital=100000):
    df['Position'] = df['Signal'].shift(1)  # 次日執行
    df['Return'] = np.log(df['Close']/df['Close'].shift(1))
    df['Strategy_Return'] = df['Position'] * df['Return']
    df['Cum_Return'] = df['Strategy_Return'].cumsum()
    df['Drawdown'] = df['Cum_Return'] - df['Cum_Return'].cummax()
    return df

參數優化網格

優化方法：

from itertools import product

def parameter_optimization(data, n_range, k_range):
    results = []
    for n, k1, k2 in product(n_range, k_range, k_range):
        temp_df = preprocess_data(data.copy(), n)
        temp_df = calculate_range(temp_df, k1, k2)
        temp_df = generate_signals(temp_df)
        temp_df = backtest(temp_df)
        sharpe = temp_df['Strategy_Return'].mean() / temp_df['Strategy_Return'].std()
        results.append((n, k1, k2, sharpe))
    return pd.DataFrame(results, columns=['N','k1','k2','Sharpe'])

實盤注意事項

1. 滑點處理：

   def apply_slippage(signal, slippage=0.001):
      return signal * (1 - slippage)
   

2. 資金管理建議：

   • 單筆交易風險不超過2%
   • 動態調整倉位大小

3. 失效場景：

   • 震蕩行情中會出現連續虧損
   • 重大新聞事件導致價格跳空

完整代碼示例

# 完整實現
class DualThrust:
    def __init__(self, n=5, k1=0.5, k2=0.5):
        self.n = n
        self.k1 = k1
        self.k2 = k2
    
    def fit(self, data):
        df = data.copy()
        # 計算指標
        df = preprocess_data(df, self.n)
        df = calculate_range(df, self.k1, self.k2)
        # 生成信號
        df = generate_signals(df)
        # 回測
        df = backtest(df)
        return df

# 使用示例
strategy = DualThrust(n=5, k1=0.5, k2=0.5)
results = strategy.fit(data)
print(results[['Close','Upper','Lower','Signal']].tail())

參考文獻

1. Chalek, M. (1988). “Trading Systems: Methods and Algorithms”
2. 《量化交易之路》- 阿布
3. QuantConnect官方文檔

”`

注：實際使用時需要： 1. 替換示例圖片URL 2. 根據實際數據調整參數 3. 添加更詳細的風險控制邏輯 4. 補充mermaid圖表渲染支持（如需要）