15035830784

Committed 09 May 2025 07:53AM UTC coverage: 94.367% (-0.009%) from 94.376%

Build # 15035830784

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #11558 from viotemp1/optuna

switch hyperopt from scikit-optimize to  Optuna

Run Details

120 of 127 new or added lines in 8 files covered. (94.49%)

3 existing lines in 2 files now uncovered.

22333 of 23666 relevant lines covered (94.37%)

0.94 hits per line

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

97.75

/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 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__)


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

        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)
                sampler = optuna_samplers_dict[o_sampler](seed=random_state)
        else:
            sampler = o_sampler

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

freqtrade / freqtrade / 15035830784

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