基本完毕

strategies/bollinger.py

@@ -1,57 +1,39 @@
-import MetaTrader5 as mt5
 import pandas as pd
-from utils import get_rates, close_all, send_order
-from logger import logger
+from .base_strategy import BaseStrategy
+from config import STRATEGY_CONFIG
 
-class Strategy:
-    def __init__(self):
-        self.symbol = "XAUUSD"
-        self.timeframe = mt5.TIMEFRAME_M1
-        self.bollinger_period = 20
-        self.bollinger_std_dev = 2
+class BollingerStrategy(BaseStrategy):
+    def __init__(self, data_provider, symbol, timeframe, period=None, std_dev=None):
+        super().__init__(data_provider, symbol, timeframe)
+        # 从配置中获取参数，如果传入参数则使用传入的参数
+        config = STRATEGY_CONFIG.get('bollinger', {})
+        self.period = period if period is not None else config.get('period', 20)
+        self.std_dev = std_dev if std_dev is not None else config.get('std_dev', 2.0)
 
     def _calculate_indicators(self, df):
-        """
-        计算布林带指标
-        """
-        mean = df['close'].rolling(self.bollinger_period).mean()
-        std = df['close'].rolling(self.bollinger_period).std()
-        df['upper_band'] = mean + self.bollinger_std_dev * std
-        df['lower_band'] = mean - self.bollinger_std_dev * std
+        mean = df['close'].rolling(self.period).mean()
+        std = df['close'].rolling(self.period).std()
+        df['upper_band'] = mean + self.std_dev * std
+        df['lower_band'] = mean - self.std_dev * std
         return df
 
     def generate_signal(self):
-        """
-        布林带策略实盘：
-        当价格跌破下轨买入，涨破上轨卖出。
-        """
-        rates = get_rates(self.symbol, self.timeframe, self.bollinger_period + 30)
-        if rates is None or len(rates) < self.bollinger_period:
+        rates = self.data_provider.get_historical_data(self.symbol, self.timeframe, self.period + 5)
+        if rates is None or len(rates) < self.period:
             return 0
         df = pd.DataFrame(rates)
         df = self._calculate_indicators(df)
 
-        if df['close'].iloc[-2] < df['lower_band'].iloc[-2]:
-            logger.info(f"价格跌破下轨，产生买入信号: {self.symbol}")
+        if df['close'].iloc[-1] < df['lower_band'].iloc[-1]:
             return 1
-        elif df['close'].iloc[-2] > df['upper_band'].iloc[-2]:
-            logger.info(f"价格涨破上轨，产生卖出信号: {self.symbol}")
+        elif df['close'].iloc[-1] > df['upper_band'].iloc[-1]:
             return -1
         return 0
 
     def run_backtest(self, df):
-        """
-        布林带策略回测：
-        价格突破下轨买入，突破上轨卖出
-        返回信号序列：1买入，-1卖出，0无操作
-        """
         df = df.copy()
         df = self._calculate_indicators(df)
-
         signals = pd.Series(0, index=df.index)
-        for i in range(self.bollinger_period, len(df)):
-            if df['close'].iloc[i-1] < df['lower_band'].iloc[i-1]:
-                signals.iat[i] = 1
-            elif df['close'].iloc[i-1] > df['upper_band'].iloc[i-1]:
-                signals.iat[i] = -1
-        return signals
+        signals[df['close'] < df['lower_band']] = 1
+        signals[df['close'] > df['upper_band']] = -1
+        return signals