create BaseAnalysis class

freqtrade/optimize/base_analysis.py

@@ -0,0 +1,104 @@
+import logging
+import shutil
+from copy import deepcopy
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from pandas import DataFrame
+
+from freqtrade.configuration import TimeRange
+from freqtrade.data.history import get_timerange
+from freqtrade.exchange import timeframe_to_minutes
+from freqtrade.loggers.set_log_levels import (reduce_verbosity_for_bias_tester,
+                                              restore_verbosity_for_bias_tester)
+from freqtrade.optimize.backtesting import Backtesting
+
+
+logger = logging.getLogger(__name__)
+
+
+class VarHolder:
+    timerange: TimeRange
+    data: DataFrame
+    indicators: Dict[str, DataFrame]
+    result: DataFrame
+    compared: DataFrame
+    from_dt: datetime
+    to_dt: datetime
+    compared_dt: datetime
+    timeframe: str
+    startup_candle: int
+
+
+class BaseAnalysis:
+
+    def __init__(self, config: Dict[str, Any], strategy_obj: Dict):
+        self.failed_bias_check = True
+        self.full_varHolder = VarHolder()
+        self._fee = None
+
+        # pull variables the scope of the lookahead_analysis-instance
+        self.local_config = deepcopy(config)
+        self.local_config['strategy'] = strategy_obj['name']
+        self.strategy_obj = strategy_obj
+
+    @staticmethod
+    def dt_to_timestamp(dt: datetime):
+        timestamp = int(dt.replace(tzinfo=timezone.utc).timestamp())
+        return timestamp
+
+    def prepare_data(self, varholder: VarHolder, pairs_to_load: List[DataFrame]):
+
+        if 'freqai' in self.local_config and 'identifier' in self.local_config['freqai']:
+            # purge previous data if the freqai model is defined
+            # (to be sure nothing is carried over from older backtests)
+            path_to_current_identifier = (
+                Path(f"{self.local_config['user_data_dir']}/models/"
+                     f"{self.local_config['freqai']['identifier']}").resolve())
+            # remove folder and its contents
+            if Path.exists(path_to_current_identifier):
+                shutil.rmtree(path_to_current_identifier)
+
+        prepare_data_config = deepcopy(self.local_config)
+        prepare_data_config['timerange'] = (str(self.dt_to_timestamp(varholder.from_dt)) + "-" +
+                                            str(self.dt_to_timestamp(varholder.to_dt)))
+        prepare_data_config['exchange']['pair_whitelist'] = pairs_to_load
+
+        if self._fee is not None:
+            # Don't re-calculate fee per pair, as fee might differ per pair.
+            prepare_data_config['fee'] = self._fee
+
+        backtesting = Backtesting(prepare_data_config, self.exchange)
+        backtesting._set_strategy(backtesting.strategylist[0])
+
+        varholder.data, varholder.timerange = backtesting.load_bt_data()
+        backtesting.load_bt_data_detail()
+        varholder.timeframe = backtesting.timeframe
+
+        varholder.indicators = backtesting.strategy.advise_all_indicators(varholder.data)
+
+    def fill_full_varholder(self):
+        self.full_varHolder = VarHolder()
+
+        # define datetime in human-readable format
+        parsed_timerange = TimeRange.parse_timerange(self.local_config['timerange'])
+
+        if parsed_timerange.startdt is None:
+            self.full_varHolder.from_dt = datetime.fromtimestamp(0, tz=timezone.utc)
+        else:
+            self.full_varHolder.from_dt = parsed_timerange.startdt
+
+        if parsed_timerange.stopdt is None:
+            self.full_varHolder.to_dt = datetime.utcnow()
+        else:
+            self.full_varHolder.to_dt = parsed_timerange.stopdt
+
+        self.prepare_data(self.full_varHolder, self.local_config['pairs'])
+
+    def start(self) -> None:
+
+        # first make a single backtest
+        self.fill_full_varholder()
+
+        reduce_verbosity_for_bias_tester()

freqtrade/optimize/recursive_analysis.py

@@ -12,45 +12,22 @@ from freqtrade.exchange import timeframe_to_minutes
 from freqtrade.loggers.set_log_levels import (reduce_verbosity_for_bias_tester,
                                               restore_verbosity_for_bias_tester)
 from freqtrade.optimize.backtesting import Backtesting
+from freqtrade.optimize.base_analysis import BaseAnalysis, VarHolder
 
 
 logger = logging.getLogger(__name__)
 
 
-class VarHolder:
+class RecursiveAnalysis(BaseAnalysis):
-    timerange: TimeRange
-    data: DataFrame
-    indicators: Dict[str, DataFrame]
-    from_dt: datetime
-    to_dt: datetime
-    timeframe: str
-    startup_candle: int
-
-
-class RecursiveAnalysis:
 
     def __init__(self, config: Dict[str, Any], strategy_obj: Dict):
-        self.failed_bias_check = True
+        super().__init__(config, strategy_obj)
-        self.full_varHolder = VarHolder()
         self.partial_varHolder_array: List[VarHolder] = []
         self.partial_varHolder_lookahead_array: List[VarHolder] = []
 
-        self.entry_varHolders: List[VarHolder] = []
-        self.exit_varHolders: List[VarHolder] = []
-        self.exchange: Optional[Any] = None
-
-        # pull variables the scope of the recursive_analysis-instance
-        self.local_config = deepcopy(config)
-        self.local_config['strategy'] = strategy_obj['name']
         self._startup_candle = config.get('startup_candle', [199, 399, 499, 999, 1999])
-        self.strategy_obj = strategy_obj
         self.dict_recursive: Dict[str, Any] = dict()
 
-    @staticmethod
-    def dt_to_timestamp(dt: datetime):
-        timestamp = int(dt.replace(tzinfo=timezone.utc).timestamp())
-        return timestamp
-
     # For recursive bias check
     # analyzes two data frames with processed indicators and shows differences between them.
     def analyze_indicators(self):
@@ -123,51 +100,6 @@ class RecursiveAnalysis:
         else:
             logger.info("No lookahead bias on indicators found. Stop the process.")
 
-    def prepare_data(self, varholder: VarHolder, pairs_to_load: List[DataFrame]):
-
-        if 'freqai' in self.local_config and 'identifier' in self.local_config['freqai']:
-            # purge previous data if the freqai model is defined
-            # (to be sure nothing is carried over from older backtests)
-            path_to_current_identifier = (
-                Path(f"{self.local_config['user_data_dir']}/models/"
-                     f"{self.local_config['freqai']['identifier']}").resolve())
-            # remove folder and its contents
-            if Path.exists(path_to_current_identifier):
-                shutil.rmtree(path_to_current_identifier)
-
-        prepare_data_config = deepcopy(self.local_config)
-        prepare_data_config['timerange'] = (str(self.dt_to_timestamp(varholder.from_dt)) + "-" +
-                                            str(self.dt_to_timestamp(varholder.to_dt)))
-        prepare_data_config['exchange']['pair_whitelist'] = pairs_to_load
-
-        backtesting = Backtesting(prepare_data_config, self.exchange)
-        self.exchange = backtesting.exchange
-        backtesting._set_strategy(backtesting.strategylist[0])
-
-        varholder.data, varholder.timerange = backtesting.load_bt_data()
-        backtesting.load_bt_data_detail()
-        varholder.timeframe = backtesting.timeframe
-
-        varholder.indicators = backtesting.strategy.advise_all_indicators(varholder.data)
-
-    def fill_full_varholder(self):
-        self.full_varHolder = VarHolder()
-
-        # define datetime in human-readable format
-        parsed_timerange = TimeRange.parse_timerange(self.local_config['timerange'])
-
-        if parsed_timerange.startdt is None:
-            self.full_varHolder.from_dt = datetime.fromtimestamp(0, tz=timezone.utc)
-        else:
-            self.full_varHolder.from_dt = parsed_timerange.startdt
-
-        if parsed_timerange.stopdt is None:
-            self.full_varHolder.to_dt = datetime.utcnow()
-        else:
-            self.full_varHolder.to_dt = parsed_timerange.stopdt
-
-        self.prepare_data(self.full_varHolder, self.local_config['pairs'])
-
     def fill_partial_varholder(self, start_date, startup_candle):
         partial_varHolder = VarHolder()
 
@@ -186,9 +118,6 @@ class RecursiveAnalysis:
 
         partial_varHolder.from_dt = self.full_varHolder.from_dt
         partial_varHolder.to_dt = end_date
-        # partial_varHolder.startup_candle = startup_candle
-
-        # self.local_config['startup_candle_count'] = startup_candle
 
         self.prepare_data(partial_varHolder, self.local_config['pairs'])
 
@@ -196,11 +125,8 @@ class RecursiveAnalysis:
 
     def start(self) -> None:
 
-        # first make a single backtest
+        super().start()
-        self.fill_full_varholder()
+        
-
-        reduce_verbosity_for_bias_tester()
-
         start_date_full = self.full_varHolder.from_dt
         end_date_full = self.full_varHolder.to_dt