[Strategy] Add Dual Thrust Strategy

Author: KarhouTam

src/quanteval/strategies/__init__.py

@@ -3,9 +3,11 @@
 from quanteval.strategies.dual_ma import DualMAStrategy
 from quanteval.strategies.bollinger_mean_reversion import BollingerMeanReversionStrategy
 from quanteval.strategies.buy_hold import BuyAndHoldStrategy
+from quanteval.strategies.dual_thrust import DualThrustStrategy
 
 __all__ = [
     'DualMAStrategy',
     'BollingerMeanReversionStrategy',
     'BuyAndHoldStrategy',
+    'DualThrustStrategy',
 ]

src/quanteval/strategies/dual_thrust.py

@@ -0,0 +1,75 @@
+"""
+Dual Thrust Strategy -
+Strategy Logic:
+    - Buy: When price breaks above the upper threshold (based on previous range)
+    - Sell: When price breaks below the lower threshold (based on previous range)
+策略逻辑：
+    - 买入：当价格突破上轨（基于前一天的区间）
+    - 卖出：当价格突破下轨（基于前一天的区间）
+"""
+
+import pandas as pd
+import numpy as np
+from quanteval.core.strategy import Strategy
+
+
+class DualThrustStrategy(Strategy):
+    """
+    Dual Thrust Strategy - A breakout strategy based on price range.
+
+    双重突破策略，基于价格区间的突破策略。
+
+    Strategy Logic:
+        - Buy: When price breaks above the upper threshold (based on previous range)
+        - Sell: When price breaks below the lower threshold (based on previous range)
+    策略逻辑：
+        - 买入：当价格突破上轨（基于前一天的区间）
+        - 卖出：当价格突破下轨（基于前一天的区间）
+    Args:
+        k1: 上轨系数 (Upper threshold multiplier, default 0.5)
+        k2: 下轨系数 (Lower threshold multiplier, default 0.5)
+
+    Example:
+        >>> strategy = DualThrustStrategy(k1=0.5, k2=0.5)
+        >>> bt = Backtester(strategy, data)
+        >>> results = bt.run()
+    """
+
+    def __init__(self, k1: float = 0.5, k2: float = 0.5, window: int = 5):
+        super().__init__(
+            name=f'DualThrust(k1={k1},k2={k2}, N={window})', k1=k1, k2=k2, window=window
+        )
+
+    def generate_signals(self, data: pd.DataFrame) -> pd.Series:
+        """
+        生成交易信号
+
+        Generate trading signals based on Dual Thrust logic.
+
+        Returns:
+            Series with values:
+                1: Long position (price > upper threshold)
+                -1: Short position (price < lower threshold)
+                0: No position
+        """
+        k1 = self.params['k1']
+        k2 = self.params['k2']
+        n = self.params['window']
+
+        # Calculate previous day's range
+        hh = data['High'].shift(1).rolling(window=n).max()
+        hc = data['Close'].shift(1).rolling(window=n).max()
+        ll = data['Low'].shift(1).rolling(window=n).min()
+        lc = data['Close'].shift(1).rolling(window=n).min()
+        diff = np.maximum(hh - lc, hc - ll)
+
+        # Calculate thresholds
+        upper_threshold = data['Open'] + k1 * diff
+        lower_threshold = data['Open'] - k2 * diff
+
+        # Generate signals
+        signals = pd.Series(0, index=data.index)
+        signals[data['High'] > upper_threshold] = 1  # Buy signal
+        signals[data['Low'] < lower_threshold] = 0  # Sell signal
+
+        return signals

tests/test_strategies.py

@@ -1,6 +1,11 @@
 import pandas as pd
 
-from quanteval import BollingerMeanReversionStrategy, BuyAndHoldStrategy, DualMAStrategy
+from quanteval.strategies import (
+    BollingerMeanReversionStrategy,
+    BuyAndHoldStrategy,
+    DualMAStrategy,
+    DualThrustStrategy,
+)
 
 
 def test_buy_and_hold_always_long(sample_market_data: pd.DataFrame) -> None:
@@ -20,3 +25,9 @@ def test_bollinger_strategy_returns_position_series(sample_market_data: pd.DataF
     )
     assert signal.index.equals(sample_market_data.index)
     assert set(signal.unique()).issubset({0.0, 1.0})
+
+
+def test_dual_thrust_returns_binary_signal(sample_market_data: pd.DataFrame) -> None:
+    signal = DualThrustStrategy(k1=0.5, k2=0.5, window=5).generate_signals(sample_market_data)
+    assert isinstance(signal, pd.Series)
+    assert set(signal.dropna().unique()).issubset({0, 1})