Source code for optuna.terminator.improvement.evaluator

from __future__ import annotations

import abc
from typing import Dict
from typing import List
from typing import TYPE_CHECKING

import numpy as np

from optuna._experimental import experimental_class
from optuna.distributions import BaseDistribution
from optuna.samplers._lazy_random_state import LazyRandomState
from optuna.search_space import intersection_search_space
from import StudyDirection
from optuna.trial import FrozenTrial
from optuna.trial import TrialState

    from optuna._gp import acqf
    from optuna._gp import gp
    from optuna._gp import optim_sample
    from optuna._gp import prior
    from optuna._gp import search_space
    from optuna._imports import _LazyImport

    gp = _LazyImport("")
    optim_sample = _LazyImport("optuna._gp.optim_sample")
    acqf = _LazyImport("optuna._gp.acqf")
    prior = _LazyImport("optuna._gp.prior")
    search_space = _LazyImport("optuna._gp.search_space")


def _get_beta(n_params: int, n_trials: int, delta: float = 0.1) -> float:
    # TODO(nabenabe0928): Check the original implementation to verify.
    # Especially, |D| seems to be the domain size, but not the dimension based on Theorem 1.
    beta = 2 * np.log(n_params * n_trials**2 * np.pi**2 / 6 / delta)

    # The following div is according to the original paper: "We then further scale it down
    # by a factor of 5 as defined in the experiments in
    # `Srinivas et al. (2010) <>`_"
    beta /= 5

    return beta

[docs] @experimental_class("3.2.0") class BaseImprovementEvaluator(metaclass=abc.ABCMeta): """Base class for improvement evaluators.""" @abc.abstractmethod def evaluate( self, trials: List[FrozenTrial], study_direction: StudyDirection, ) -> float: pass
[docs] @experimental_class("3.2.0") class RegretBoundEvaluator(BaseImprovementEvaluator): """An error evaluator for upper bound on the regret with high-probability confidence. This evaluator evaluates the regret of current best solution, which defined as the difference between the objective value of the best solution and of the global optimum. To be specific, this evaluator calculates the upper bound on the regret based on the fact that empirical estimator of the objective function is bounded by lower and upper confidence bounds with high probability under the Gaussian process model assumption. Args: gp: A Gaussian process model on which evaluation base. If not specified, the default Gaussian process model is used. top_trials_ratio: A ratio of top trials to be considered when estimating the regret. Default to 0.5. min_n_trials: A minimum number of complete trials to estimate the regret. Default to 20. min_lcb_n_additional_samples: A minimum number of additional samples to estimate the lower confidence bound. Default to 2000. For further information about this evaluator, please refer to the following paper: - `Automatic Termination for Hyperparameter Optimization <>`_ """ # NOQA: E501 def __init__( self, top_trials_ratio: float = DEFAULT_TOP_TRIALS_RATIO, min_n_trials: int = DEFAULT_MIN_N_TRIALS, seed: int | None = None, ) -> None: self._top_trials_ratio = top_trials_ratio self._min_n_trials = min_n_trials self._log_prior = prior.default_log_prior self._minimum_noise = prior.DEFAULT_MINIMUM_NOISE_VAR self._optimize_n_samples = 2048 self._rng = LazyRandomState(seed) def _get_top_n( self, normalized_params: np.ndarray, values: np.ndarray ) -> tuple[np.ndarray, np.ndarray]: assert len(normalized_params) == len(values) n_trials = len(normalized_params) top_n = np.clip(int(n_trials * self._top_trials_ratio), self._min_n_trials, n_trials) top_n_val = np.partition(values, n_trials - top_n)[n_trials - top_n] top_n_mask = values >= top_n_val return normalized_params[top_n_mask], values[top_n_mask] def _compute_standardized_regret_bound( self, kernel_params: gp.KernelParamsTensor, gp_search_space: search_space.SearchSpace, normalized_top_n_params: np.ndarray, standarized_top_n_values: np.ndarray, ) -> float: """ In the original paper, f(x) was intended to be minimized, but here we would like to maximize f(x). Hence, the following changes happen: 1. min(ucb) over top trials becomes max(lcb) over top trials, and 2. min(lcb) over the search space becomes max(ucb) over the search space, and 3. Regret bound becomes max(ucb) over the search space minus max(lcb) over top trials. """ n_trials, n_params = normalized_top_n_params.shape # calculate max_ucb beta = _get_beta(n_params, n_trials) ucb_acqf_params = acqf.create_acqf_params( acqf_type=acqf.AcquisitionFunctionType.UCB, kernel_params=kernel_params, search_space=gp_search_space, X=normalized_top_n_params, Y=standarized_top_n_values, beta=beta, ) # UCB over the search space. (Original: LCB over the search space. See Change 1 above.) standardized_ucb_value = max( acqf.eval_acqf_no_grad(ucb_acqf_params, normalized_top_n_params).max(), optim_sample.optimize_acqf_sample( ucb_acqf_params, n_samples=self._optimize_n_samples, rng=self._rng.rng )[1], ) # calculate min_lcb lcb_acqf_params = acqf.create_acqf_params( acqf_type=acqf.AcquisitionFunctionType.LCB, kernel_params=kernel_params, search_space=gp_search_space, X=normalized_top_n_params, Y=standarized_top_n_values, beta=beta, ) # LCB over the top trials. (Original: UCB over the top trials. See Change 2 above.) standardized_lcb_value = np.max( acqf.eval_acqf_no_grad(lcb_acqf_params, normalized_top_n_params) ) # max(UCB) - max(LCB). (Original: min(UCB) - min(LCB). See Change 3 above.) return standardized_ucb_value - standardized_lcb_value # standardized regret bound def evaluate( self, trials: List[FrozenTrial], study_direction: StudyDirection, ) -> float: optuna_search_space = intersection_search_space(trials) self._validate_input(trials, optuna_search_space) complete_trials = [t for t in trials if t.state == TrialState.COMPLETE] # _gp module assumes that optimization direction is maximization sign = -1 if study_direction == StudyDirection.MINIMIZE else 1 values = np.array([t.value for t in complete_trials]) * sign gp_search_space, normalized_params = search_space.get_search_space_and_normalized_params( complete_trials, optuna_search_space ) normalized_top_n_params, top_n_values = self._get_top_n(normalized_params, values) top_n_values_mean = top_n_values.mean() top_n_values_std = max(1e-10, top_n_values.std()) standarized_top_n_values = (top_n_values - top_n_values_mean) / top_n_values_std kernel_params = gp.fit_kernel_params( X=normalized_top_n_params, Y=standarized_top_n_values, is_categorical=(gp_search_space.scale_types == search_space.ScaleType.CATEGORICAL), log_prior=self._log_prior, minimum_noise=self._minimum_noise, # TODO(contramundum53): Add option to specify this. deterministic_objective=False, # TODO(y0z): Add `kernel_params_cache` to speedup. initial_kernel_params=None, ) standardized_regret_bound = self._compute_standardized_regret_bound( kernel_params, gp_search_space, normalized_top_n_params, standarized_top_n_values, ) return standardized_regret_bound * top_n_values_std # regret bound @classmethod def _validate_input( cls, trials: List[FrozenTrial], search_space: Dict[str, BaseDistribution] ) -> None: if len([t for t in trials if t.state == TrialState.COMPLETE]) == 0: raise ValueError( "Because no trial has been completed yet, the regret bound cannot be evaluated." ) if len(search_space) == 0: raise ValueError( "The intersection search space is empty. This condition is not supported by " f"{cls.__name__}." )
[docs] @experimental_class("3.4.0") class BestValueStagnationEvaluator(BaseImprovementEvaluator): """Evaluates the stagnation period of the best value in an optimization process. This class is initialized with a maximum stagnation period (`max_stagnation_trials`) and is designed to evaluate the remaining trials before reaching this maximum period of allowed stagnation. If this remaining trials reach zero, the trial terminates. Therefore, the default error evaluator is instantiated by StaticErrorEvaluator(const=0). Args: max_stagnation_trials: The maximum number of trials allowed for stagnation. """ def __init__( self, max_stagnation_trials: int = 30, ) -> None: if max_stagnation_trials < 0: raise ValueError("The maximum number of stagnant trials must not be negative.") self._max_stagnation_trials = max_stagnation_trials def evaluate( self, trials: List[FrozenTrial], study_direction: StudyDirection, ) -> float: self._validate_input(trials) is_maximize_direction = True if (study_direction == StudyDirection.MAXIMIZE) else False trials = [t for t in trials if t.state == TrialState.COMPLETE] current_step = len(trials) - 1 best_step = 0 for i, trial in enumerate(trials): best_value = trials[best_step].value current_value = trial.value assert best_value is not None assert current_value is not None if is_maximize_direction and (best_value < current_value): best_step = i elif (not is_maximize_direction) and (best_value > current_value): best_step = i return self._max_stagnation_trials - (current_step - best_step) @classmethod def _validate_input( cls, trials: List[FrozenTrial], ) -> None: if len([t for t in trials if t.state == TrialState.COMPLETE]) == 0: raise ValueError( "Because no trial has been completed yet, the improvement cannot be evaluated." )