HPO Optimizers API¶

def search(self) -> dict[str, Any]:
    """
    Actual function performing hyperparameter optimization for the selected
    objective function.

    Returns
    -------
    dict[str, Any]
        The best set of parameters with achieved score.

    Raises
    ------
    OptimizationError
        When there were no successful evaluations for given constraints.
    """
    self.evaluate_initial_random_nodes()

    iterations_limit = self._get_iterations_limit()

    for _ in range(iterations_limit):
        self._run_iteration()

    successful_evaluations = [evaluation for evaluation in self.evaluations if evaluation["score"] is not None]
    if not successful_evaluations:
        raise OptimizationError("Number of evaluations has reached limit. All iterations have failed.")

    # Sort in ascending order and take the last element (highest score).
    # This assumes we're maximizing the score.
    best_config_with_score = sorted(successful_evaluations, key=lambda d: d["score"])[-1]

    return best_config_with_score

def evaluate_initial_random_nodes(self) -> None:
    """
    Perform evaluation of randomly chosen n nodes from the solutions space.
    Evaluations are performed until desired number of successful evaluations
    is reached or maximum number of evaluations is reached.

    When the optimizer has been warm-started with known observations,
    already-successful evaluations count toward the n_random_nodes target
    and already-evaluated combinations are excluded from candidates.

    The selection is stratified: combinations that introduce at least one new
    unique value for any categorical (string-valued) parameter are moved to the
    front of the queue before the random fill. This guarantees that every
    distinct categorical value (e.g. ``search_mode="vector"`` vs
    ``search_mode="hybrid"``) is evaluated at least once before GAM training
    begins, regardless of how skewed the raw search space is.

    A warning is logged when ``n_random_nodes`` is smaller than the estimated
    minimum required to guarantee full categorical coverage.
    """
    successful_evaluations = sum(1 for e in self.evaluations if e["score"] is not None)

    if successful_evaluations >= self.settings.n_random_nodes:
        logger.info(
            "Skipping random evaluation phase: %d known successful evaluations >= n_random_nodes (%d).",
            successful_evaluations,
            self.settings.n_random_nodes,
        )
        return

    if len(self.evaluations) >= self.max_iterations:
        return

    combinations_local = [c for c in copy(self._search_space.combinations) if c not in self._evaluated_combinations]
    random.Random(self.settings.random_state).shuffle(combinations_local)

    # Values already covered by successful warm-start observations so
    # stratification does not waste early slots on redundant coverage.
    already_covered: dict[str, set[str]] = {}
    for eval_entry in self.evaluations:
        if eval_entry.get("score") is not None:
            for col, val in eval_entry.items():
                if col != "score" and isinstance(val, str):
                    already_covered.setdefault(col, set()).add(val)

    min_needed = self._min_n_random_nodes_for_coverage(combinations_local, already_covered)
    if min_needed > self.settings.n_random_nodes:
        logger.warning(
            "n_random_nodes=%d may be too small to guarantee full categorical coverage "
            "(estimated minimum: %d). Consider increasing n_random_nodes.",
            self.settings.n_random_nodes,
            min_needed,
        )

    combinations_local = self._get_stratified_combinations(combinations_local, already_covered)

    gen = (x for x in combinations_local)

    while successful_evaluations < self.settings.n_random_nodes:
        params = next(gen)
        score = self._objective_function(params=params)
        if score is not None:
            successful_evaluations += 1
        self._evaluated_combinations.append(params)
        params_with_score = params | {"score": score}
        self.evaluations.append(params_with_score)

        if len(self.evaluations) == self.max_iterations:
            break

def search(self) -> dict[str, Any]:
    """
    Actual function performing hyperparameter optimization for the selected
    objective function.

    Returns
    -------
    dict[str, Any]
        The best set of parameters with achieved score.

    Raises
    ------
    OptimizationError
        When there were no successful evaluations for given constraints.
    """
    combinations = list(self._search_space.combinations)
    random.Random(self.settings.random_state).shuffle(combinations)

    for idx in range(self.settings.max_evals):
        score = self._objective_function(combinations[idx])
        self._evaluated_combinations.append(combinations[idx] | {"score": score})

    successful_evaluations = [ev for ev in self._evaluated_combinations if ev["score"] is not None]

    if not successful_evaluations:
        raise OptimizationError("Number of evaluations has reached limit. All iterations have failed.")

    best_config_with_score = sorted(successful_evaluations, key=lambda d: d["score"])[-1]

    return best_config_with_score