15749127397

Committed 14 Jun 2025 07:51PM UTC coverage: 94.23% (-0.2%) from 94.383%

Build # 15749127397

Build Type

push

github

Committed by

web-flow

Commit Message

Merge branch 'freqtrade:develop' into develop

Run Details

26 of 29 new or added lines in 12 files covered. (89.66%)

332 existing lines in 27 files now uncovered.

22146 of 23502 relevant lines covered (94.23%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.89

/freqtrade/optimize/hyperopt/hyperopt_optimizer.py

"""
This module contains the hyperopt optimizer class, which needs to be pickled
and will be sent to the hyperopt worker processes.
"""

import logging
import sys
import warnings
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import optuna
from joblib import delayed, dump, load, wrap_non_picklable_objects
from joblib.externals import cloudpickle
from optuna.exceptions import ExperimentalWarning
from optuna.terminator import BestValueStagnationEvaluator, Terminator
from pandas import DataFrame

from freqtrade.constants import DATETIME_PRINT_FORMAT, Config
from freqtrade.data.converter import trim_dataframes
from freqtrade.data.history import get_timerange
from freqtrade.data.metrics import calculate_market_change
from freqtrade.enums import HyperoptState
from freqtrade.exceptions import OperationalException
from freqtrade.ft_types import BacktestContentType
from freqtrade.misc import deep_merge_dicts, round_dict
from freqtrade.optimize.backtesting import Backtesting

# Import IHyperOptLoss to allow unpickling classes from these modules
from freqtrade.optimize.hyperopt.hyperopt_auto import HyperOptAuto
from freqtrade.optimize.hyperopt_loss.hyperopt_loss_interface import IHyperOptLoss
from freqtrade.optimize.hyperopt_tools import HyperoptStateContainer, HyperoptTools
from freqtrade.optimize.optimize_reports import generate_strategy_stats
from freqtrade.optimize.space import (
    DimensionProtocol,
    SKDecimal,
    ft_CategoricalDistribution,
    ft_FloatDistribution,
    ft_IntDistribution,
)
from freqtrade.resolvers.hyperopt_resolver import HyperOptLossResolver
from freqtrade.util.dry_run_wallet import get_dry_run_wallet


logger = logging.getLogger(__name__)

INITIAL_POINTS = 30

MAX_LOSS = 100000  # just a big enough number to be bad result in loss optimization

optuna_samplers_dict = {
    "TPESampler": optuna.samplers.TPESampler,
    "GPSampler": optuna.samplers.GPSampler,
    "CmaEsSampler": optuna.samplers.CmaEsSampler,
    "NSGAIISampler": optuna.samplers.NSGAIISampler,
    "NSGAIIISampler": optuna.samplers.NSGAIIISampler,
    "QMCSampler": optuna.samplers.QMCSampler,
}


class HyperOptimizer:
    """
    HyperoptOptimizer class
    This class is sent to the hyperopt worker processes.
    """

    def __init__(self, config: Config, data_pickle_file: Path) -> None:
        self.buy_space: list[DimensionProtocol] = []
        self.sell_space: list[DimensionProtocol] = []
        self.protection_space: list[DimensionProtocol] = []
        self.roi_space: list[DimensionProtocol] = []
        self.stoploss_space: list[DimensionProtocol] = []
        self.trailing_space: list[DimensionProtocol] = []
        self.max_open_trades_space: list[DimensionProtocol] = []
        self.dimensions: list[DimensionProtocol] = []
        self.o_dimensions: dict = {}

        self.config = config
        self.min_date: datetime
        self.max_date: datetime

        self.backtesting = Backtesting(self.config)
        self.pairlist = self.backtesting.pairlists.whitelist
        self.custom_hyperopt: HyperOptAuto
        self.analyze_per_epoch = self.config.get("analyze_per_epoch", False)

        if not self.config.get("hyperopt"):
            self.custom_hyperopt = HyperOptAuto(self.config)
        else:
            raise OperationalException(
                "Using separate Hyperopt files has been removed in 2021.9. Please convert "
                "your existing Hyperopt file to the new Hyperoptable strategy interface"
            )

        self.backtesting._set_strategy(self.backtesting.strategylist[0])
        self.custom_hyperopt.strategy = self.backtesting.strategy

        self.hyperopt_pickle_magic(self.backtesting.strategy.__class__.__bases__)
        self.custom_hyperoptloss: IHyperOptLoss = HyperOptLossResolver.load_hyperoptloss(
            self.config
        )
        self.calculate_loss = self.custom_hyperoptloss.hyperopt_loss_function

        self.data_pickle_file = data_pickle_file

        self.market_change = 0.0

        self.es_epochs = config.get("early_stop", 0)
        if self.es_epochs > 0 and self.es_epochs < 0.2 * config.get("epochs", 0):
            logger.warning(f"Early stop epochs {self.es_epochs} lower than 20% of total epochs")

        if HyperoptTools.has_space(self.config, "sell"):
            # Make sure use_exit_signal is enabled
            self.config["use_exit_signal"] = True

    def prepare_hyperopt(self) -> None:
        # Initialize spaces ...
        self.init_spaces()

        self.prepare_hyperopt_data()

        # We don't need exchange instance anymore while running hyperopt
        self.backtesting.exchange.close()
        self.backtesting.exchange._api = None
        self.backtesting.exchange._api_async = None
        self.backtesting.exchange.loop = None  # type: ignore
        self.backtesting.exchange._loop_lock = None  # type: ignore
        self.backtesting.exchange._cache_lock = None  # type: ignore
        # self.backtesting.exchange = None  # type: ignore
        self.backtesting.pairlists = None  # type: ignore

    def get_strategy_name(self) -> str:
        return self.backtesting.strategy.get_strategy_name()

    def hyperopt_pickle_magic(self, bases: tuple[type, ...]) -> None:
        """
        Hyperopt magic to allow strategy inheritance across files.
        For this to properly work, we need to register the module of the imported class
        to pickle as value.
        """
        for modules in bases:
            if modules.__name__ != "IStrategy":
                if mod := sys.modules.get(modules.__module__):
                    cloudpickle.register_pickle_by_value(mod)
                self.hyperopt_pickle_magic(modules.__bases__)

    def _get_params_details(self, params: dict) -> dict:
        """
        Return the params for each space
        """
        result: dict = {}

        if HyperoptTools.has_space(self.config, "buy"):
            result["buy"] = round_dict({p.name: params.get(p.name) for p in self.buy_space}, 13)
        if HyperoptTools.has_space(self.config, "sell"):
            result["sell"] = round_dict({p.name: params.get(p.name) for p in self.sell_space}, 13)
        if HyperoptTools.has_space(self.config, "protection"):
            result["protection"] = round_dict(
                {p.name: params.get(p.name) for p in self.protection_space}, 13
            )
        if HyperoptTools.has_space(self.config, "roi"):
            result["roi"] = round_dict(
                {str(k): v for k, v in self.custom_hyperopt.generate_roi_table(params).items()}, 13
            )
        if HyperoptTools.has_space(self.config, "stoploss"):
            result["stoploss"] = round_dict(
                {p.name: params.get(p.name) for p in self.stoploss_space}, 13
            )
        if HyperoptTools.has_space(self.config, "trailing"):
            result["trailing"] = round_dict(
                self.custom_hyperopt.generate_trailing_params(params), 13
            )
        if HyperoptTools.has_space(self.config, "trades"):
            result["max_open_trades"] = round_dict(
                {
                    "max_open_trades": (
                        self.backtesting.strategy.max_open_trades
                        if self.backtesting.strategy.max_open_trades != float("inf")
                        else -1
                    )
                },
                13,
            )

        return result

    def _get_no_optimize_details(self) -> dict[str, Any]:
        """
        Get non-optimized parameters
        """
        result: dict[str, Any] = {}
        strategy = self.backtesting.strategy
        if not HyperoptTools.has_space(self.config, "roi"):
            result["roi"] = {str(k): v for k, v in strategy.minimal_roi.items()}
        if not HyperoptTools.has_space(self.config, "stoploss"):
            result["stoploss"] = {"stoploss": strategy.stoploss}
        if not HyperoptTools.has_space(self.config, "trailing"):
            result["trailing"] = {
                "trailing_stop": strategy.trailing_stop,
                "trailing_stop_positive": strategy.trailing_stop_positive,
                "trailing_stop_positive_offset": strategy.trailing_stop_positive_offset,
                "trailing_only_offset_is_reached": strategy.trailing_only_offset_is_reached,
            }
        if not HyperoptTools.has_space(self.config, "trades"):
            result["max_open_trades"] = {"max_open_trades": strategy.max_open_trades}
        return result

    def init_spaces(self):
        """
        Assign the dimensions in the hyperoptimization space.
        """
        if HyperoptTools.has_space(self.config, "protection"):
            # Protections can only be optimized when using the Parameter interface
            logger.debug("Hyperopt has 'protection' space")
            # Enable Protections if protection space is selected.
            self.config["enable_protections"] = True
            self.backtesting.enable_protections = True
            self.protection_space = self.custom_hyperopt.protection_space()

        if HyperoptTools.has_space(self.config, "buy"):
            logger.debug("Hyperopt has 'buy' space")
            self.buy_space = self.custom_hyperopt.buy_indicator_space()

        if HyperoptTools.has_space(self.config, "sell"):
            logger.debug("Hyperopt has 'sell' space")
            self.sell_space = self.custom_hyperopt.sell_indicator_space()

        if HyperoptTools.has_space(self.config, "roi"):
            logger.debug("Hyperopt has 'roi' space")
            self.roi_space = self.custom_hyperopt.roi_space()

        if HyperoptTools.has_space(self.config, "stoploss"):
            logger.debug("Hyperopt has 'stoploss' space")
            self.stoploss_space = self.custom_hyperopt.stoploss_space()

        if HyperoptTools.has_space(self.config, "trailing"):
            logger.debug("Hyperopt has 'trailing' space")
            self.trailing_space = self.custom_hyperopt.trailing_space()

        if HyperoptTools.has_space(self.config, "trades"):
            logger.debug("Hyperopt has 'trades' space")
            self.max_open_trades_space = self.custom_hyperopt.max_open_trades_space()

        self.dimensions = (
            self.buy_space
            + self.sell_space
            + self.protection_space
            + self.roi_space
            + self.stoploss_space
            + self.trailing_space
            + self.max_open_trades_space
        )

    def assign_params(self, params_dict: dict[str, Any], category: str) -> None:
        """
        Assign hyperoptable parameters
        """
        for attr_name, attr in self.backtesting.strategy.enumerate_parameters(category):
            if attr.optimize:
                # noinspection PyProtectedMember
                attr.value = params_dict[attr_name]

    @delayed
    @wrap_non_picklable_objects
    def generate_optimizer_wrapped(self, params_dict: dict[str, Any]) -> dict[str, Any]:
        return self.generate_optimizer(params_dict)

    def generate_optimizer(self, params_dict: dict[str, Any]) -> dict[str, Any]:
        """
        Used Optimize function.
        Called once per epoch to optimize whatever is configured.
        Keep this function as optimized as possible!
        """
        HyperoptStateContainer.set_state(HyperoptState.OPTIMIZE)
        backtest_start_time = datetime.now(timezone.utc)

        # Apply parameters
        if HyperoptTools.has_space(self.config, "buy"):
            self.assign_params(params_dict, "buy")

        if HyperoptTools.has_space(self.config, "sell"):
            self.assign_params(params_dict, "sell")

        if HyperoptTools.has_space(self.config, "protection"):
            self.assign_params(params_dict, "protection")

        if HyperoptTools.has_space(self.config, "roi"):
            self.backtesting.strategy.minimal_roi = self.custom_hyperopt.generate_roi_table(
                params_dict
            )

        if HyperoptTools.has_space(self.config, "stoploss"):
            self.backtesting.strategy.stoploss = params_dict["stoploss"]

        if HyperoptTools.has_space(self.config, "trailing"):
            d = self.custom_hyperopt.generate_trailing_params(params_dict)
            self.backtesting.strategy.trailing_stop = d["trailing_stop"]
            self.backtesting.strategy.trailing_stop_positive = d["trailing_stop_positive"]
            self.backtesting.strategy.trailing_stop_positive_offset = d[
                "trailing_stop_positive_offset"
            ]
            self.backtesting.strategy.trailing_only_offset_is_reached = d[
                "trailing_only_offset_is_reached"
            ]

        if HyperoptTools.has_space(self.config, "trades"):
            if self.config["stake_amount"] == "unlimited" and (
                params_dict["max_open_trades"] == -1 or params_dict["max_open_trades"] == 0
            ):
                # Ignore unlimited max open trades if stake amount is unlimited
                params_dict.update({"max_open_trades": self.config["max_open_trades"]})

            updated_max_open_trades = (
                int(params_dict["max_open_trades"])
                if (params_dict["max_open_trades"] != -1 and params_dict["max_open_trades"] != 0)
                else float("inf")
            )

            self.config.update({"max_open_trades": updated_max_open_trades})

            self.backtesting.strategy.max_open_trades = updated_max_open_trades

        with self.data_pickle_file.open("rb") as f:
            processed = load(f, mmap_mode="r")
        if self.analyze_per_epoch:
            # Data is not yet analyzed, rerun populate_indicators.
            processed = self.advise_and_trim(processed)

        bt_results = self.backtesting.backtest(
            processed=processed, start_date=self.min_date, end_date=self.max_date
        )
        backtest_end_time = datetime.now(timezone.utc)
        bt_results.update(
            {
                "backtest_start_time": int(backtest_start_time.timestamp()),
                "backtest_end_time": int(backtest_end_time.timestamp()),
            }
        )
        result = self._get_results_dict(
            bt_results, self.min_date, self.max_date, params_dict, processed=processed
        )
        return result

    def _get_results_dict(
        self,
        backtesting_results: BacktestContentType,
        min_date: datetime,
        max_date: datetime,
        params_dict: dict[str, Any],
        processed: dict[str, DataFrame],
    ) -> dict[str, Any]:
        params_details = self._get_params_details(params_dict)

        strat_stats = generate_strategy_stats(
            self.pairlist,
            self.backtesting.strategy.get_strategy_name(),
            backtesting_results,
            min_date,
            max_date,
            market_change=self.market_change,
            is_hyperopt=True,
        )
        results_explanation = HyperoptTools.format_results_explanation_string(
            strat_stats, self.config["stake_currency"]
        )

        not_optimized = self.backtesting.strategy.get_no_optimize_params()
        not_optimized = deep_merge_dicts(not_optimized, self._get_no_optimize_details())

        trade_count = strat_stats["total_trades"]
        total_profit = strat_stats["profit_total"]

        # If this evaluation contains too short amount of trades to be
        # interesting -- consider it as 'bad' (assigned max. loss value)
        # in order to cast this hyperspace point away from optimization
        # path. We do not want to optimize 'hodl' strategies.
        loss: float = MAX_LOSS
        if trade_count >= self.config["hyperopt_min_trades"]:
            loss = self.calculate_loss(
                results=backtesting_results["results"],
                trade_count=trade_count,
                min_date=min_date,
                max_date=max_date,
                config=self.config,
                processed=processed,
                backtest_stats=strat_stats,
                starting_balance=get_dry_run_wallet(self.config),
            )
        return {
            "loss": loss,
            "params_dict": params_dict,
            "params_details": params_details,
            "params_not_optimized": not_optimized,
            "results_metrics": strat_stats,
            "results_explanation": results_explanation,
            "total_profit": total_profit,
        }

    def convert_dimensions_to_optuna_space(self, s_dimensions: list[DimensionProtocol]) -> dict:
        o_dimensions: dict[str, optuna.distributions.BaseDistribution] = {}
        for original_dim in s_dimensions:
            if isinstance(
                original_dim,
                ft_CategoricalDistribution | ft_IntDistribution | ft_FloatDistribution | SKDecimal,
            ):
                o_dimensions[original_dim.name] = original_dim
            else:
                raise OperationalException(
                    f"Unknown search space {original_dim.name} - {original_dim} / \
                        {type(original_dim)}"
                )
        return o_dimensions

    def get_optimizer(
        self,
        random_state: int,
    ):
        o_sampler = self.custom_hyperopt.generate_estimator(
            dimensions=self.dimensions, random_state=random_state
        )
        self.o_dimensions = self.convert_dimensions_to_optuna_space(self.dimensions)

        if isinstance(o_sampler, str):
            if o_sampler not in optuna_samplers_dict.keys():
                raise OperationalException(f"Optuna Sampler {o_sampler} not supported.")
            with warnings.catch_warnings():
                warnings.filterwarnings(action="ignore", category=ExperimentalWarning)
                if o_sampler in ["NSGAIIISampler", "NSGAIISampler"]:
                    sampler = optuna_samplers_dict[o_sampler](
                        seed=random_state, population_size=INITIAL_POINTS
                    )
                elif o_sampler in ["GPSampler", "TPESampler", "CmaEsSampler"]:
                    sampler = optuna_samplers_dict[o_sampler](
                        seed=random_state, n_startup_trials=INITIAL_POINTS
                    )
                else:
                    sampler = optuna_samplers_dict[o_sampler](seed=random_state)
        else:
            sampler = o_sampler

        if self.es_epochs > 0:
            with warnings.catch_warnings():
                warnings.filterwarnings(action="ignore", category=ExperimentalWarning)
                self.es_terminator = Terminator(BestValueStagnationEvaluator(self.es_epochs))

        logger.info(f"Using optuna sampler {o_sampler}.")
        return optuna.create_study(sampler=sampler, direction="minimize")

    def advise_and_trim(self, data: dict[str, DataFrame]) -> dict[str, DataFrame]:
        preprocessed = self.backtesting.strategy.advise_all_indicators(data)

        # Trim startup period from analyzed dataframe to get correct dates for output.
        # This is only used to keep track of min/max date after trimming.
        # The result is NOT returned from this method, actual trimming happens in backtesting.
        trimmed = trim_dataframes(preprocessed, self.timerange, self.backtesting.required_startup)
        self.min_date, self.max_date = get_timerange(trimmed)
        if not self.market_change:
            self.market_change = calculate_market_change(trimmed, "close")

        # Real trimming will happen as part of backtesting.
        return preprocessed

    def prepare_hyperopt_data(self) -> None:
        HyperoptStateContainer.set_state(HyperoptState.DATALOAD)
        data, self.timerange = self.backtesting.load_bt_data()
        logger.info("Dataload complete. Calculating indicators")

        if not self.analyze_per_epoch:
            HyperoptStateContainer.set_state(HyperoptState.INDICATORS)

            preprocessed = self.advise_and_trim(data)

            logger.info(
                f"Hyperopting with data from "
                f"{self.min_date.strftime(DATETIME_PRINT_FORMAT)} "
                f"up to {self.max_date.strftime(DATETIME_PRINT_FORMAT)} "
                f"({(self.max_date - self.min_date).days} days).."
            )
            # Store non-trimmed data - will be trimmed after signal generation.
            dump(preprocessed, self.data_pickle_file)
        else:
            dump(data, self.data_pickle_file)

1	"""
2	This module contains the hyperopt optimizer class, which needs to be pickled
3	and will be sent to the hyperopt worker processes.
4	"""
5
6	import logging	1✔
7	import sys	1✔
8	import warnings	1✔
9	from datetime import datetime, timezone	1✔
10	from pathlib import Path	1✔
11	from typing import Any	1✔
12
13	import optuna	1✔
14	from joblib import delayed, dump, load, wrap_non_picklable_objects	1✔
15	from joblib.externals import cloudpickle	1✔
16	from optuna.exceptions import ExperimentalWarning	1✔
17	from optuna.terminator import BestValueStagnationEvaluator, Terminator	1✔
18	from pandas import DataFrame	1✔
19
20	from freqtrade.constants import DATETIME_PRINT_FORMAT, Config	1✔
21	from freqtrade.data.converter import trim_dataframes	1✔
22	from freqtrade.data.history import get_timerange	1✔
23	from freqtrade.data.metrics import calculate_market_change	1✔
24	from freqtrade.enums import HyperoptState	1✔
25	from freqtrade.exceptions import OperationalException	1✔
26	from freqtrade.ft_types import BacktestContentType	1✔
27	from freqtrade.misc import deep_merge_dicts, round_dict	1✔
28	from freqtrade.optimize.backtesting import Backtesting	1✔
29
30	# Import IHyperOptLoss to allow unpickling classes from these modules
31	from freqtrade.optimize.hyperopt.hyperopt_auto import HyperOptAuto	1✔
32	from freqtrade.optimize.hyperopt_loss.hyperopt_loss_interface import IHyperOptLoss	1✔
33	from freqtrade.optimize.hyperopt_tools import HyperoptStateContainer, HyperoptTools	1✔
34	from freqtrade.optimize.optimize_reports import generate_strategy_stats	1✔
35	from freqtrade.optimize.space import (	1✔
36	DimensionProtocol,
37	SKDecimal,
38	ft_CategoricalDistribution,
39	ft_FloatDistribution,
40	ft_IntDistribution,
41	)
42	from freqtrade.resolvers.hyperopt_resolver import HyperOptLossResolver	1✔
43	from freqtrade.util.dry_run_wallet import get_dry_run_wallet	1✔
44
45
46	logger = logging.getLogger(__name__)	1✔
47
48	INITIAL_POINTS = 30	1✔
49
50	MAX_LOSS = 100000 # just a big enough number to be bad result in loss optimization	1✔
51
52	optuna_samplers_dict = {	1✔
53	"TPESampler": optuna.samplers.TPESampler,
54	"GPSampler": optuna.samplers.GPSampler,
55	"CmaEsSampler": optuna.samplers.CmaEsSampler,
56	"NSGAIISampler": optuna.samplers.NSGAIISampler,
57	"NSGAIIISampler": optuna.samplers.NSGAIIISampler,
58	"QMCSampler": optuna.samplers.QMCSampler,
59	}
60
61
62	class HyperOptimizer:	1✔
63	"""
64	HyperoptOptimizer class
65	This class is sent to the hyperopt worker processes.
66	"""
67
68	def __init__(self, config: Config, data_pickle_file: Path) -> None:	1✔
69	self.buy_space: list[DimensionProtocol] = []	1✔
70	self.sell_space: list[DimensionProtocol] = []	1✔
71	self.protection_space: list[DimensionProtocol] = []	1✔
72	self.roi_space: list[DimensionProtocol] = []	1✔
73	self.stoploss_space: list[DimensionProtocol] = []	1✔
74	self.trailing_space: list[DimensionProtocol] = []	1✔
75	self.max_open_trades_space: list[DimensionProtocol] = []	1✔
76	self.dimensions: list[DimensionProtocol] = []	1✔
77	self.o_dimensions: dict = {}	1✔
78
79	self.config = config	1✔
80	self.min_date: datetime	1✔
81	self.max_date: datetime	1✔
82
83	self.backtesting = Backtesting(self.config)	1✔
84	self.pairlist = self.backtesting.pairlists.whitelist	1✔
85	self.custom_hyperopt: HyperOptAuto	1✔
86	self.analyze_per_epoch = self.config.get("analyze_per_epoch", False)	1✔
87
88	if not self.config.get("hyperopt"):	1✔
89	self.custom_hyperopt = HyperOptAuto(self.config)	1✔
90	else:
UNCOV 91	raise OperationalException(	×
92	"Using separate Hyperopt files has been removed in 2021.9. Please convert "
93	"your existing Hyperopt file to the new Hyperoptable strategy interface"
94	)
95
96	self.backtesting._set_strategy(self.backtesting.strategylist[0])	1✔
97	self.custom_hyperopt.strategy = self.backtesting.strategy	1✔
98
99	self.hyperopt_pickle_magic(self.backtesting.strategy.__class__.__bases__)	1✔
100	self.custom_hyperoptloss: IHyperOptLoss = HyperOptLossResolver.load_hyperoptloss(	1✔
101	self.config
102	)
103	self.calculate_loss = self.custom_hyperoptloss.hyperopt_loss_function	1✔
104
105	self.data_pickle_file = data_pickle_file	1✔
106
107	self.market_change = 0.0	1✔
108
109	self.es_epochs = config.get("early_stop", 0)	1✔
110	if self.es_epochs > 0 and self.es_epochs < 0.2 * config.get("epochs", 0):	1✔
UNCOV 111	logger.warning(f"Early stop epochs {self.es_epochs} lower than 20% of total epochs")	×
112
113	if HyperoptTools.has_space(self.config, "sell"):	1✔
114	# Make sure use_exit_signal is enabled
115	self.config["use_exit_signal"] = True	1✔
116
117	def prepare_hyperopt(self) -> None:	1✔
118	# Initialize spaces ...
119	self.init_spaces()	1✔
120
121	self.prepare_hyperopt_data()	1✔
122
123	# We don't need exchange instance anymore while running hyperopt
124	self.backtesting.exchange.close()	1✔
125	self.backtesting.exchange._api = None	1✔
126	self.backtesting.exchange._api_async = None	1✔
127	self.backtesting.exchange.loop = None # type: ignore	1✔
128	self.backtesting.exchange._loop_lock = None # type: ignore	1✔
129	self.backtesting.exchange._cache_lock = None # type: ignore	1✔
130	# self.backtesting.exchange = None # type: ignore
131	self.backtesting.pairlists = None # type: ignore	1✔
132
133	def get_strategy_name(self) -> str:	1✔
134	return self.backtesting.strategy.get_strategy_name()	1✔
135
136	def hyperopt_pickle_magic(self, bases: tuple[type, ...]) -> None:	1✔
137	"""
138	Hyperopt magic to allow strategy inheritance across files.
139	For this to properly work, we need to register the module of the imported class
140	to pickle as value.
141	"""
142	for modules in bases:	1✔
143	if modules.__name__ != "IStrategy":	1✔
144	if mod := sys.modules.get(modules.__module__):	1✔
145	cloudpickle.register_pickle_by_value(mod)	1✔
146	self.hyperopt_pickle_magic(modules.__bases__)	1✔
147
148	def _get_params_details(self, params: dict) -> dict:	1✔
149	"""
150	Return the params for each space
151	"""
152	result: dict = {}	1✔
153
154	if HyperoptTools.has_space(self.config, "buy"):	1✔
155	result["buy"] = round_dict({p.name: params.get(p.name) for p in self.buy_space}, 13)	1✔
156	if HyperoptTools.has_space(self.config, "sell"):	1✔
157	result["sell"] = round_dict({p.name: params.get(p.name) for p in self.sell_space}, 13)	1✔
158	if HyperoptTools.has_space(self.config, "protection"):	1✔
159	result["protection"] = round_dict(	1✔
160	{p.name: params.get(p.name) for p in self.protection_space}, 13
161	)
162	if HyperoptTools.has_space(self.config, "roi"):	1✔
163	result["roi"] = round_dict(	1✔
164	{str(k): v for k, v in self.custom_hyperopt.generate_roi_table(params).items()}, 13
165	)
166	if HyperoptTools.has_space(self.config, "stoploss"):	1✔
167	result["stoploss"] = round_dict(	1✔
168	{p.name: params.get(p.name) for p in self.stoploss_space}, 13
169	)
170	if HyperoptTools.has_space(self.config, "trailing"):	1✔
171	result["trailing"] = round_dict(	1✔
172	self.custom_hyperopt.generate_trailing_params(params), 13
173	)
174	if HyperoptTools.has_space(self.config, "trades"):	1✔
175	result["max_open_trades"] = round_dict(	1✔
176	{
177	"max_open_trades": (
178	self.backtesting.strategy.max_open_trades
179	if self.backtesting.strategy.max_open_trades != float("inf")
180	else -1
181	)
182	},
183	13,
184	)
185
186	return result	1✔
187
188	def _get_no_optimize_details(self) -> dict[str, Any]:	1✔
189	"""
190	Get non-optimized parameters
191	"""
192	result: dict[str, Any] = {}	1✔
193	strategy = self.backtesting.strategy	1✔
194	if not HyperoptTools.has_space(self.config, "roi"):	1✔
195	result["roi"] = {str(k): v for k, v in strategy.minimal_roi.items()}	1✔
196	if not HyperoptTools.has_space(self.config, "stoploss"):	1✔
197	result["stoploss"] = {"stoploss": strategy.stoploss}	1✔
198	if not HyperoptTools.has_space(self.config, "trailing"):	1✔
199	result["trailing"] = {	1✔
200	"trailing_stop": strategy.trailing_stop,
201	"trailing_stop_positive": strategy.trailing_stop_positive,
202	"trailing_stop_positive_offset": strategy.trailing_stop_positive_offset,
203	"trailing_only_offset_is_reached": strategy.trailing_only_offset_is_reached,
204	}
205	if not HyperoptTools.has_space(self.config, "trades"):	1✔
206	result["max_open_trades"] = {"max_open_trades": strategy.max_open_trades}	1✔
207	return result	1✔
208
209	def init_spaces(self):	1✔
210	"""
211	Assign the dimensions in the hyperoptimization space.
212	"""
213	if HyperoptTools.has_space(self.config, "protection"):	1✔
214	# Protections can only be optimized when using the Parameter interface
215	logger.debug("Hyperopt has 'protection' space")	1✔
216	# Enable Protections if protection space is selected.
217	self.config["enable_protections"] = True	1✔
218	self.backtesting.enable_protections = True	1✔
219	self.protection_space = self.custom_hyperopt.protection_space()	1✔
220
221	if HyperoptTools.has_space(self.config, "buy"):	1✔
222	logger.debug("Hyperopt has 'buy' space")	1✔
223	self.buy_space = self.custom_hyperopt.buy_indicator_space()	1✔
224
225	if HyperoptTools.has_space(self.config, "sell"):	1✔
226	logger.debug("Hyperopt has 'sell' space")	1✔
227	self.sell_space = self.custom_hyperopt.sell_indicator_space()	1✔
228
229	if HyperoptTools.has_space(self.config, "roi"):	1✔
230	logger.debug("Hyperopt has 'roi' space")	1✔
231	self.roi_space = self.custom_hyperopt.roi_space()	1✔
232
233	if HyperoptTools.has_space(self.config, "stoploss"):	1✔
234	logger.debug("Hyperopt has 'stoploss' space")	1✔
235	self.stoploss_space = self.custom_hyperopt.stoploss_space()	1✔
236
237	if HyperoptTools.has_space(self.config, "trailing"):	1✔
238	logger.debug("Hyperopt has 'trailing' space")	1✔
239	self.trailing_space = self.custom_hyperopt.trailing_space()	1✔
240
241	if HyperoptTools.has_space(self.config, "trades"):	1✔
242	logger.debug("Hyperopt has 'trades' space")	1✔
243	self.max_open_trades_space = self.custom_hyperopt.max_open_trades_space()	1✔
244
245	self.dimensions = (	1✔
246	self.buy_space
247	+ self.sell_space
248	+ self.protection_space
249	+ self.roi_space
250	+ self.stoploss_space
251	+ self.trailing_space
252	+ self.max_open_trades_space
253	)
254
255	def assign_params(self, params_dict: dict[str, Any], category: str) -> None:	1✔
256	"""
257	Assign hyperoptable parameters
258	"""
259	for attr_name, attr in self.backtesting.strategy.enumerate_parameters(category):	1✔
260	if attr.optimize:	1✔
261	# noinspection PyProtectedMember
262	attr.value = params_dict[attr_name]	1✔
263
264	@delayed	1✔
265	@wrap_non_picklable_objects	1✔
266	def generate_optimizer_wrapped(self, params_dict: dict[str, Any]) -> dict[str, Any]:	1✔
267	return self.generate_optimizer(params_dict)	1✔
268
269	def generate_optimizer(self, params_dict: dict[str, Any]) -> dict[str, Any]:	1✔
270	"""
271	Used Optimize function.
272	Called once per epoch to optimize whatever is configured.
273	Keep this function as optimized as possible!
274	"""
275	HyperoptStateContainer.set_state(HyperoptState.OPTIMIZE)	1✔
276	backtest_start_time = datetime.now(timezone.utc)	1✔
277
278	# Apply parameters
279	if HyperoptTools.has_space(self.config, "buy"):	1✔
280	self.assign_params(params_dict, "buy")	1✔
281
282	if HyperoptTools.has_space(self.config, "sell"):	1✔
283	self.assign_params(params_dict, "sell")	1✔
284
285	if HyperoptTools.has_space(self.config, "protection"):	1✔
286	self.assign_params(params_dict, "protection")	1✔
287
288	if HyperoptTools.has_space(self.config, "roi"):	1✔
289	self.backtesting.strategy.minimal_roi = self.custom_hyperopt.generate_roi_table(	1✔
290	params_dict
291	)
292
293	if HyperoptTools.has_space(self.config, "stoploss"):	1✔
294	self.backtesting.strategy.stoploss = params_dict["stoploss"]	1✔
295
296	if HyperoptTools.has_space(self.config, "trailing"):	1✔
297	d = self.custom_hyperopt.generate_trailing_params(params_dict)	1✔
298	self.backtesting.strategy.trailing_stop = d["trailing_stop"]	1✔
299	self.backtesting.strategy.trailing_stop_positive = d["trailing_stop_positive"]	1✔
300	self.backtesting.strategy.trailing_stop_positive_offset = d[	1✔
301	"trailing_stop_positive_offset"
302	]
303	self.backtesting.strategy.trailing_only_offset_is_reached = d[	1✔
304	"trailing_only_offset_is_reached"
305	]
306
307	if HyperoptTools.has_space(self.config, "trades"):	1✔
308	if self.config["stake_amount"] == "unlimited" and (	1✔
309	params_dict["max_open_trades"] == -1 or params_dict["max_open_trades"] == 0
310	):
311	# Ignore unlimited max open trades if stake amount is unlimited
UNCOV 312	params_dict.update({"max_open_trades": self.config["max_open_trades"]})	×
313
314	updated_max_open_trades = (	1✔
315	int(params_dict["max_open_trades"])
316	if (params_dict["max_open_trades"] != -1 and params_dict["max_open_trades"] != 0)
317	else float("inf")
318	)
319
320	self.config.update({"max_open_trades": updated_max_open_trades})	1✔
321
322	self.backtesting.strategy.max_open_trades = updated_max_open_trades	1✔
323
324	with self.data_pickle_file.open("rb") as f:	1✔
325	processed = load(f, mmap_mode="r")	1✔
326	if self.analyze_per_epoch:	1✔
327	# Data is not yet analyzed, rerun populate_indicators.
UNCOV 328	processed = self.advise_and_trim(processed)	×
329
330	bt_results = self.backtesting.backtest(	1✔
331	processed=processed, start_date=self.min_date, end_date=self.max_date
332	)
333	backtest_end_time = datetime.now(timezone.utc)	1✔
334	bt_results.update(	1✔
335	{
336	"backtest_start_time": int(backtest_start_time.timestamp()),
337	"backtest_end_time": int(backtest_end_time.timestamp()),
338	}
339	)
340	result = self._get_results_dict(	1✔
341	bt_results, self.min_date, self.max_date, params_dict, processed=processed
342	)
343	return result	1✔
344
345	def _get_results_dict(	1✔
346	self,
347	backtesting_results: BacktestContentType,
348	min_date: datetime,
349	max_date: datetime,
350	params_dict: dict[str, Any],
351	processed: dict[str, DataFrame],
352	) -> dict[str, Any]:
353	params_details = self._get_params_details(params_dict)	1✔
354
355	strat_stats = generate_strategy_stats(	1✔
356	self.pairlist,
357	self.backtesting.strategy.get_strategy_name(),
358	backtesting_results,
359	min_date,
360	max_date,
361	market_change=self.market_change,
362	is_hyperopt=True,
363	)
364	results_explanation = HyperoptTools.format_results_explanation_string(	1✔
365	strat_stats, self.config["stake_currency"]
366	)
367
368	not_optimized = self.backtesting.strategy.get_no_optimize_params()	1✔
369	not_optimized = deep_merge_dicts(not_optimized, self._get_no_optimize_details())	1✔
370
371	trade_count = strat_stats["total_trades"]	1✔
372	total_profit = strat_stats["profit_total"]	1✔
373
374	# If this evaluation contains too short amount of trades to be
375	# interesting -- consider it as 'bad' (assigned max. loss value)
376	# in order to cast this hyperspace point away from optimization
377	# path. We do not want to optimize 'hodl' strategies.
378	loss: float = MAX_LOSS	1✔
379	if trade_count >= self.config["hyperopt_min_trades"]:	1✔
380	loss = self.calculate_loss(	1✔
381	results=backtesting_results["results"],
382	trade_count=trade_count,
383	min_date=min_date,
384	max_date=max_date,
385	config=self.config,
386	processed=processed,
387	backtest_stats=strat_stats,
388	starting_balance=get_dry_run_wallet(self.config),
389	)
390	return {	1✔
391	"loss": loss,
392	"params_dict": params_dict,
393	"params_details": params_details,
394	"params_not_optimized": not_optimized,
395	"results_metrics": strat_stats,
396	"results_explanation": results_explanation,
397	"total_profit": total_profit,
398	}
399
400	def convert_dimensions_to_optuna_space(self, s_dimensions: list[DimensionProtocol]) -> dict:	1✔
401	o_dimensions: dict[str, optuna.distributions.BaseDistribution] = {}	1✔
402	for original_dim in s_dimensions:	1✔
403	if isinstance(	1✔
404	original_dim,
405	ft_CategoricalDistribution \| ft_IntDistribution \| ft_FloatDistribution \| SKDecimal,
406	):
407	o_dimensions[original_dim.name] = original_dim	1✔
408	else:
UNCOV 409	raise OperationalException(	×
410	f"Unknown search space {original_dim.name} - {original_dim} / \
411	{type(original_dim)}"
412	)
413	return o_dimensions	1✔
414
415	def get_optimizer(	1✔
416	self,
417	random_state: int,
418	):
419	o_sampler = self.custom_hyperopt.generate_estimator(	1✔
420	dimensions=self.dimensions, random_state=random_state
421	)
422	self.o_dimensions = self.convert_dimensions_to_optuna_space(self.dimensions)	1✔
423
424	if isinstance(o_sampler, str):	1✔
425	if o_sampler not in optuna_samplers_dict.keys():	1✔
426	raise OperationalException(f"Optuna Sampler {o_sampler} not supported.")	1✔
427	with warnings.catch_warnings():	1✔
428	warnings.filterwarnings(action="ignore", category=ExperimentalWarning)	1✔
429	if o_sampler in ["NSGAIIISampler", "NSGAIISampler"]:	1✔
430	sampler = optuna_samplers_dict[o_sampler](	1✔
431	seed=random_state, population_size=INITIAL_POINTS
432	)
UNCOV 433	elif o_sampler in ["GPSampler", "TPESampler", "CmaEsSampler"]:	×
UNCOV 434	sampler = optuna_samplers_dict[o_sampler](	×
435	seed=random_state, n_startup_trials=INITIAL_POINTS
436	)
437	else:
UNCOV 438	sampler = optuna_samplers_dict[o_sampler](seed=random_state)	×
439	else:
UNCOV 440	sampler = o_sampler	×
441
442	if self.es_epochs > 0:	1✔
UNCOV 443	with warnings.catch_warnings():	×
UNCOV 444	warnings.filterwarnings(action="ignore", category=ExperimentalWarning)	×
UNCOV 445	self.es_terminator = Terminator(BestValueStagnationEvaluator(self.es_epochs))	×
446
447	logger.info(f"Using optuna sampler {o_sampler}.")	1✔
448	return optuna.create_study(sampler=sampler, direction="minimize")	1✔
449
450	def advise_and_trim(self, data: dict[str, DataFrame]) -> dict[str, DataFrame]:	1✔
451	preprocessed = self.backtesting.strategy.advise_all_indicators(data)	1✔
452
453	# Trim startup period from analyzed dataframe to get correct dates for output.
454	# This is only used to keep track of min/max date after trimming.
455	# The result is NOT returned from this method, actual trimming happens in backtesting.
456	trimmed = trim_dataframes(preprocessed, self.timerange, self.backtesting.required_startup)	1✔
457	self.min_date, self.max_date = get_timerange(trimmed)	1✔
458	if not self.market_change:	1✔
459	self.market_change = calculate_market_change(trimmed, "close")	1✔
460
461	# Real trimming will happen as part of backtesting.
462	return preprocessed	1✔
463
464	def prepare_hyperopt_data(self) -> None:	1✔
465	HyperoptStateContainer.set_state(HyperoptState.DATALOAD)	1✔
466	data, self.timerange = self.backtesting.load_bt_data()	1✔
467	logger.info("Dataload complete. Calculating indicators")	1✔
468
469	if not self.analyze_per_epoch:	1✔
470	HyperoptStateContainer.set_state(HyperoptState.INDICATORS)	1✔
471
472	preprocessed = self.advise_and_trim(data)	1✔
473
474	logger.info(	1✔
475	f"Hyperopting with data from "
476	f"{self.min_date.strftime(DATETIME_PRINT_FORMAT)} "
477	f"up to {self.max_date.strftime(DATETIME_PRINT_FORMAT)} "
478	f"({(self.max_date - self.min_date).days} days).."
479	)
480	# Store non-trimmed data - will be trimmed after signal generation.
481	dump(preprocessed, self.data_pickle_file)	1✔
482	else:
483	dump(data, self.data_pickle_file)	1✔

freqtrade / freqtrade / 15749127397

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous