15748876185

Committed 26 May 2025 05:06AM UTC coverage: 94.322% (+0.1%) from 94.226%

Build # 15748876185

Build Type

push

github

Committed by

xmatthias

Commit Message

fix: capture ws edge-case on reconnect

Run Details

0 of 1 new or added line in 1 file covered. (0.0%)

198 existing lines in 14 files now uncovered.

22411 of 23760 relevant lines covered (94.32%)

0.94 hits per line

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

96.09

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


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

freqtrade / freqtrade / 15748876185

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