mirror of
https://github.com/freqtrade/freqtrade.git
synced 2024-09-20 17:41:12 +00:00
104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
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()
|