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.exchange: Optional[Any] = None 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()