BaseSearchCV

yohou.model_selection.search.BaseSearchCV

Bases: BaseForecaster, MetaEstimatorMixin

Abstract base class for hyperparameter search with cross-validation.

Warning: This class should not be used directly. Use derived classes GridSearchCV and RandomizedSearchCV instead.

Attributes

Name	Type	Description
cv_results_	dict of numpy (masked) ndarrays	A dict with keys as column headers and values as columns, that can be imported into a pandas DataFrame. For instance the below given table:: +-----------+------------+-------------------+---+-----------------+ |param_alpha|param_gamma |param_degree |...|rank_test_score | +===========+============+===================+===+=================+ | 1.0 | -- | -- |...| 2 | +-----------+------------+-------------------+---+-----------------+ | 10.0 | -- | -- |...| 1 | +-----------+------------+-------------------+---+-----------------+ | 100.0 | -- | -- |...| 3 | +-----------+------------+-------------------+---+-----------------+ | -- | 0.1 | 2 |...| 5 | +-----------+------------+-------------------+---+-----------------+ | -- | 0.2 | 3 |...| 4 | +-----------+------------+-------------------+---+-----------------+ will be represented by a cv_results_ dict of:: { 'param_alpha': masked_array(data=[1.0, 10.0, 100.0, --, --], mask=[False False False True True]...), 'param_gamma': masked_array(data=[--, --, --, 0.1, 0.2], mask=[ True True True False False]...), 'param_degree': masked_array(data=[--, --, --, 2, 3], mask=[ True True True False False]...), 'split0_test_score' : [0.80, 0.70, 0.60, 0.75, 0.85], 'split1_test_score' : [0.82, 0.75, 0.65, 0.72, 0.80], 'mean_test_score' : [0.81, 0.725, 0.625, 0.735, 0.825], 'std_test_score' : [0.01, 0.025, 0.025, 0.015, 0.025], 'rank_test_score' : [2, 3, 5, 4, 1], 'split0_train_score' : [0.85, 0.80, 0.75, 0.82, 0.90], 'split1_train_score' : [0.87, 0.82, 0.77, 0.84, 0.92], 'mean_train_score' : [0.86, 0.81, 0.76, 0.83, 0.91], 'std_train_score' : [0.01, 0.01, 0.01, 0.01, 0.01], 'mean_fit_time' : [0.73, 0.63, 0.43, 0.49, 0.55], 'std_fit_time' : [0.01, 0.02, 0.01, 0.01, 0.02], 'mean_score_time' : [0.01, 0.06, 0.04, 0.04, 0.05], 'std_score_time' : [0.00, 0.00, 0.00, 0.01, 0.00], 'params' : [{'alpha': 1.0}, {'alpha': 10.0}, ...], } NOTE: The key 'params' is used to store a list of parameter settings dicts for all the parameter candidates. The mean_fit_time, std_fit_time, mean_score_time and std_score_time are all in seconds. For multi-metric evaluation, the scores for all the scorers are available in the cv_results_ dict at the keys ending with that scorer's name ('_<scorer_name>') instead of '_score' shown above. ('split0_test_mae', 'mean_train_rmse' etc.)
best_forecaster_	BaseForecaster	Forecaster that was chosen by the search, i.e. forecaster which gave highest score (or smallest loss if specified) on the left out data. Not available if refit=False. See refit parameter for more information on allowed values.
best_score_	float	Mean cross-validated score of the best_forecaster_. Follows sklearn's sign convention: for lower_is_better scorers (e.g. MAE, RMSE) the value is negated so that higher always means better. The raw metric value is -best_score_. For multi-metric evaluation, this is present only if refit is specified. This attribute is not available if refit is a function.
best_params_	dict	Parameter setting that gave the best results on the hold out data. For multi-metric evaluation, this is present only if refit is specified.
best_index_	int	The index (of the cv_results_ arrays) which corresponds to the best candidate parameter setting. The dict at search.cv_results_['params'][search.best_index_] gives the parameter setting for the best model, that gives the highest mean score (search.best_score_). For multi-metric evaluation, this is present only if refit is specified.
scorer_	BaseScorer or dict	Scorer function(s) used on the held out data to choose the best parameters for the model. For multi-metric evaluation, this attribute holds the validated scoring dict which maps the scorer key to the scorer callable.
n_splits_	int	The number of cross-validation splits (folds/iterations).
refit_time_	float	Seconds used for refitting the best forecaster on the whole dataset. This is present only if refit is not False.
multimetric_	bool	Whether or not the scorers compute several metrics.
n_features_in_	int	Number of features seen during fit. Only defined if best_forecaster_ is defined (see the documentation for the refit parameter for more details) and that best_forecaster_ exposes n_features_in_ when fit.
feature_names_in_	ndarray of shape (``n_features_in_``,)	Names of features seen during fit. Only defined if best_forecaster_ is defined (see the documentation for the refit parameter for more details) and that best_forecaster_ exposes feature_names_in_ when fit.

See Also

• GridSearchCV : Exhaustive search over specified parameter values.
• RandomizedSearchCV : Randomized search over parameter distributions.

Source Code

View on GitHub

Show/Hide source

class BaseSearchCV(BaseForecaster, MetaEstimatorMixin, metaclass=ABCMeta):
    """Abstract base class for hyperparameter search with cross-validation.

    Warning: This class should not be used directly. Use derived classes
    ``GridSearchCV`` and ``RandomizedSearchCV`` instead.

    Attributes
    ----------
    cv_results_ : dict of numpy (masked) ndarrays
        A dict with keys as column headers and values as columns, that can be
        imported into a pandas ``DataFrame``.

        For instance the below given table::

            +-----------+------------+-------------------+---+-----------------+
            |param_alpha|param_gamma |param_degree       |...|rank_test_score  |
            +===========+============+===================+===+=================+
            |  1.0      |     --     |         --        |...|        2        |
            +-----------+------------+-------------------+---+-----------------+
            |  10.0     |     --     |         --        |...|        1        |
            +-----------+------------+-------------------+---+-----------------+
            |  100.0    |     --     |         --        |...|        3        |
            +-----------+------------+-------------------+---+-----------------+
            |   --      |    0.1     |         2         |...|        5        |
            +-----------+------------+-------------------+---+-----------------+
            |   --      |    0.2     |         3         |...|        4        |
            +-----------+------------+-------------------+---+-----------------+

        will be represented by a cv_results_ dict of::

            {
            'param_alpha': masked_array(data=[1.0, 10.0, 100.0, --, --],
                                        mask=[False False False  True  True]...),
            'param_gamma': masked_array(data=[--, --, --, 0.1, 0.2],
                                       mask=[ True  True  True False False]...),
            'param_degree': masked_array(data=[--, --, --, 2, 3],
                                         mask=[ True  True  True False False]...),
            'split0_test_score'  : [0.80, 0.70, 0.60, 0.75, 0.85],
            'split1_test_score'  : [0.82, 0.75, 0.65, 0.72, 0.80],
            'mean_test_score'    : [0.81, 0.725, 0.625, 0.735, 0.825],
            'std_test_score'     : [0.01, 0.025, 0.025, 0.015, 0.025],
            'rank_test_score'    : [2, 3, 5, 4, 1],
            'split0_train_score' : [0.85, 0.80, 0.75, 0.82, 0.90],
            'split1_train_score' : [0.87, 0.82, 0.77, 0.84, 0.92],
            'mean_train_score'   : [0.86, 0.81, 0.76, 0.83, 0.91],
            'std_train_score'    : [0.01, 0.01, 0.01, 0.01, 0.01],
            'mean_fit_time'      : [0.73, 0.63, 0.43, 0.49, 0.55],
            'std_fit_time'       : [0.01, 0.02, 0.01, 0.01, 0.02],
            'mean_score_time'    : [0.01, 0.06, 0.04, 0.04, 0.05],
            'std_score_time'     : [0.00, 0.00, 0.00, 0.01, 0.00],
            'params'             : [{'alpha': 1.0}, {'alpha': 10.0}, ...],
            }

        NOTE: The key ``'params'`` is used to store a list of parameter
        settings dicts for all the parameter candidates.

        The ``mean_fit_time``, ``std_fit_time``, ``mean_score_time`` and
        ``std_score_time`` are all in seconds.

        For multi-metric evaluation, the scores for all the scorers are
        available in the ``cv_results_`` dict at the keys ending with that
        scorer's name (``'_<scorer_name>'``) instead of ``'_score'`` shown
        above. (``'split0_test_mae'``, ``'mean_train_rmse'`` etc.)

    best_forecaster_ : BaseForecaster
        Forecaster that was chosen by the search, i.e. forecaster which gave
        highest score (or smallest loss if specified) on the left out data.
        Not available if ``refit=False``.

        See ``refit`` parameter for more information on allowed values.

    best_score_ : float
        Mean cross-validated score of the best_forecaster_.

        Follows sklearn's sign convention: for ``lower_is_better`` scorers
        (e.g. MAE, RMSE) the value is **negated** so that higher always
        means better.  The raw metric value is ``-best_score_``.

        For multi-metric evaluation, this is present only if ``refit`` is
        specified.

        This attribute is not available if ``refit`` is a function.

    best_params_ : dict
        Parameter setting that gave the best results on the hold out data.

        For multi-metric evaluation, this is present only if ``refit`` is
        specified.

    best_index_ : int
        The index (of the ``cv_results_`` arrays) which corresponds to the best
        candidate parameter setting.

        The dict at ``search.cv_results_['params'][search.best_index_]`` gives
        the parameter setting for the best model, that gives the highest
        mean score (``search.best_score_``).

        For multi-metric evaluation, this is present only if ``refit`` is
        specified.

    scorer_ : BaseScorer or dict
        Scorer function(s) used on the held out data to choose the best
        parameters for the model.

        For multi-metric evaluation, this attribute holds the validated
        ``scoring`` dict which maps the scorer key to the scorer callable.

    n_splits_ : int
        The number of cross-validation splits (folds/iterations).

    refit_time_ : float
        Seconds used for refitting the best forecaster on the whole dataset.

        This is present only if ``refit`` is not False.

    multimetric_ : bool
        Whether or not the scorers compute several metrics.

    n_features_in_ : int
        Number of features seen during ``fit``. Only defined if
        ``best_forecaster_`` is defined (see the documentation for the ``refit``
        parameter for more details) and that ``best_forecaster_`` exposes
        ``n_features_in_`` when fit.

    feature_names_in_ : ndarray of shape (``n_features_in_``,)
        Names of features seen during ``fit``. Only defined if
        ``best_forecaster_`` is defined (see the documentation for the ``refit``
        parameter for more details) and that ``best_forecaster_`` exposes
        ``feature_names_in_`` when fit.

    See Also
    --------
    - [`GridSearchCV`][yohou.model_selection.search.GridSearchCV] : Exhaustive search over specified parameter values.
    - [`RandomizedSearchCV`][yohou.model_selection.search.RandomizedSearchCV] : Randomized search over parameter distributions.

    """

    _parameter_constraints: dict = {
        "forecaster": [HasMethods(["fit", "predict"])],
        "scoring": [None, callable, dict],
        "n_jobs": [numbers.Integral, None],
        "refit": ["boolean", str, callable],
        "cv": [numbers.Integral, HasMethods(["split", "get_n_splits"]), None],
        "verbose": ["verbose"],
        "pre_dispatch": [numbers.Integral, str],
        "error_score": [StrOptions({"raise"}), numbers.Real],
        "return_train_score": ["boolean"],
    }

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Merge parameter constraints from all classes in the MRO."""
        super().__init_subclass__(**kwargs)
        merged: dict = {}
        for klass in reversed(cls.__mro__):
            own = klass.__dict__.get("_parameter_constraints")
            if own and isinstance(own, dict):
                merged.update(own)
        cls._parameter_constraints = merged

    @abstractmethod
    def __init__(
        self,
        forecaster,
        *,
        scoring=None,
        n_jobs=None,
        refit=True,
        cv=None,
        verbose=0,
        pre_dispatch="2*n_jobs",
        error_score=np.nan,
        return_train_score=False,
    ):
        self.forecaster = forecaster
        self.scoring = scoring
        self.n_jobs = n_jobs
        self.refit = refit
        self.cv = cv
        self.verbose = verbose
        self.pre_dispatch = pre_dispatch
        self.error_score = error_score
        self.return_train_score = return_train_score

    def __sklearn_tags__(self):
        """Get tags from best_forecaster_ if available."""
        if hasattr(self, "best_forecaster_"):
            return self.best_forecaster_.__sklearn_tags__()
        # Return tags from the base forecaster during initialization
        return self.forecaster.__sklearn_tags__()

    @property
    def n_features_in_(self):
        """Number of features seen during fit."""
        check_is_fitted(self)
        return self.best_forecaster_.n_features_in_

    @property
    def feature_names_in_(self):
        """Names of features seen during fit."""
        check_is_fitted(self)
        return self.best_forecaster_.feature_names_in_

    @property
    def interval_(self):
        """Time interval detected during fit."""
        check_is_fitted(self)
        return self.best_forecaster_.interval_

    @property
    def groups_(self):
        """Panel group names detected during fit."""
        check_is_fitted(self)
        return self.best_forecaster_.groups_

    def _check_refit_for_multimetric(self, scores):
        """Check that refit parameter is valid for multimetric scoring.

        Parameters
        ----------
        scores : dict
            Dictionary of scorer names to scores.

        Raises
        ------
        ValueError
            If refit is not valid for multimetric scoring.

        """
        multimetric_refit_msg = (
            "For multi-metric scoring, the parameter refit must be set to a "
            "scorer key or a callable to refit a forecaster with the best "
            "parameter setting on the whole data and make the best_* "
            "attributes available for that metric. If this is not needed, "
            f"refit should be set to False explicitly. {self.refit!r} was "
            "passed."
        )

        valid_refit_dict = isinstance(self.refit, str) and self.refit in scores

        if self.refit is not False and not valid_refit_dict and not callable(self.refit):
            raise ValueError(multimetric_refit_msg)

    @staticmethod
    def _select_best_index(refit, refit_metric, results):
        """Select the index of the best parameter combination.

        This method implements the logic for choosing which parameter setting
        is "best" based on the refit parameter:

        - If refit is callable: Call refit(results) which must return an integer
          index into the results arrays. This allows custom selection logic.
        - Otherwise: Use argmin on rank_test_{refit_metric}, which selects the
          parameter setting with the best (rank 1) mean test score for that metric.

        The ranking approach (argmin of rank) handles ties consistently and ensures
        that lower rank numbers (better performance) are selected.

        Parameters
        ----------
        refit : bool, str, or callable
            Refit parameter from constructor. If callable, should accept cv_results_
            dict and return integer index.
        refit_metric : str
            Name of the metric to use for refitting (e.g., 'score', 'mae', 'rmse').
            Used to construct the key 'rank_test_{refit_metric}' in results dict.
        results : dict
            Cross-validation results dictionary (cv_results_) containing
            'rank_test_{refit_metric}' and 'params' keys.

        Returns
        -------
        best_index : int
            Index of the best parameter combination. This index can be used to
            access results['params'][best_index] for the best parameters.

        Raises
        ------
        TypeError
            If callable refit returns non-integer.
        IndexError
            If callable refit returns out-of-range index.

        """
        if callable(refit):
            # If callable, refit is expected to return the index of the best
            # parameter set.
            best_index = refit(results)
            if not isinstance(best_index, numbers.Integral):
                raise TypeError("best_index_ returned is not an integer")
            if best_index < 0 or best_index >= len(results["params"]):
                raise IndexError("best_index_ index out of range")
        else:
            best_index = results[f"rank_test_{refit_metric}"].argmin()
        return best_index

    def _get_scorers(self):
        """Get the scorer(s) to be used.

        This is used in fit and get_metadata_routing.

        Returns
        -------
        scorers : BaseScorer or _MultimetricScorer
            Scorer to use.
        refit_metric : str
            Name of the metric to use for refitting.
        """
        refit_metric = "score"

        if self.scoring is None:
            raise ValueError("scoring parameter cannot be None")

        refit_metric = "score"  # Default for single metric

        if callable(self.scoring):
            if not isinstance(self.scoring, BaseScorer):
                raise ValueError("scoring must be an instance of BaseScorer or a dict of BaseScorer instances")
            scorers = self.scoring
            # Single metric, default name is "score"
        elif isinstance(self.scoring, dict):
            # Multi-metric scoring
            scorers_result = _check_scoring(self.forecaster, self.scoring)
            # _check_scoring returns the dict directly when given a dict

            scorers_dict = cast(dict[str, BaseScorer], scorers_result)
            self._check_refit_for_multimetric(self.scoring)
            refit_metric = self.refit if isinstance(self.refit, str) else "score"
            scorers = _MultimetricScorer(scorers=scorers_dict, raise_exc=(self.error_score == "raise"))
        else:
            raise ValueError("scoring must be an instance of BaseScorer or a dict of BaseScorer instances")

        return scorers, refit_metric

    def _get_routed_params_for_fit(self, params):
        """Get the parameters to be used for routing.

        This is a method instead of a snippet in fit since it's used twice,
        here in fit, and potentially in subclasses.

        Parameters
        ----------
        params : dict
            Parameters passed to fit.

        Returns
        -------
        routed_params : Bunch
            Routed parameters for forecaster, scorer, and splitter.
        """
        routed_params = process_routing(self, "fit", **params)
        return routed_params

    @abstractmethod
    def _run_search(self, evaluate_candidates):
        """Execute the search strategy.

        Subclasses implement specific search strategies (grid, random, etc.).

        Parameters
        ----------
        evaluate_candidates : callable
            This callback accepts:
                - a list of candidates, where each candidate is a dict of
                  parameter settings.
                - an optional `cv` parameter which can be used to e.g.
                  evaluate candidates on different dataset splits.
                - an optional `more_results` dict. Each key will be added to
                  the `cv_results_` attribute. Values should be lists of
                  length `n_candidates`.

            It returns a dict of all results so far, formatted like `cv_results_`.

        Examples
        --------
        GridSearchCV implements::

            def _run_search(self, evaluate_candidates):
                evaluate_candidates(ParameterGrid(self.param_grid))

        RandomizedSearchCV implements::

            def _run_search(self, evaluate_candidates):
                evaluate_candidates(
                    ParameterSampler(self.param_distributions, self.n_iter, random_state=self.random_state)
                )
        """
        raise NotImplementedError("_run_search not implemented.")

    def _format_results(self, candidate_params, n_splits, out, more_results=None):
        """Format the cv_results_ dictionary.

        This method aggregates results from parallel _fit_and_score calls into
        the cv_results_ dictionary structure with comprehensive statistics:

        - Per-split scores: split{i}_test_{scorer} for each fold
        - Aggregated scores: mean_test_{scorer}, std_test_{scorer}
        - Rankings: rank_test_{scorer} (1=best, higher=worse)
        - Timing: mean_fit_time, std_fit_time, mean_score_time, std_score_time
        - Parameters: param_{name} as masked arrays
        - Train scores: split{i}_train_{scorer}, mean_train_{scorer}, std_train_{scorer}
          (if return_train_score=True)

        For multi-metric scoring, scores for each metric are stored with keys
        ending in the scorer name (e.g., 'mean_test_mae', 'rank_test_rmse').

        The aggregation uses weighted averaging when test_sample_counts vary
        across folds, ensuring that folds with more samples contribute
        proportionally more to the mean score.

        Parameters
        ----------
        candidate_params : list of dict
            List of parameter dictionaries, one per candidate.
        n_splits : int
            Number of cross-validation splits.
        out : list of dict
            List of fit/score results from _fit_and_score, length n_candidates * n_splits.
            Each dict contains: test_scores, train_scores, fit_time, score_time, n_test_samples.
        more_results : dict, optional
            Additional results to include in cv_results_. Keys are column names,
            values are lists of length n_candidates.

        Returns
        -------
        results : dict
            Formatted cv_results_ dictionary with keys including:
            - 'params': List of parameter dicts
            - 'param_{name}': Masked array for each parameter
            - 'split{i}_test_{scorer}': Per-fold test scores
            - 'mean_test_{scorer}': Mean test score
            - 'std_test_{scorer}': Standard deviation of test scores
            - 'rank_test_{scorer}': Ranking (1=best)
            - 'mean_fit_time', 'std_fit_time': Fit timing statistics
            - 'mean_score_time', 'std_score_time': Score timing statistics
            - Train score keys (if return_train_score=True)

        """
        n_candidates = len(candidate_params)

        # _aggregate_score_dicts returns a dict with keys:
        # fit_time, score_time, test_scores, train_scores
        out = _aggregate_score_dicts(out)

        test_scores = _normalize_score_results(out["test_scores"])
        if self.return_train_score:
            train_scores = _normalize_score_results(out["train_scores"])

        results = {}

        def _store(key_name, array, weights=None, splits=False, rank=False):
            """Store scores/times to cv_results_."""
            # Scores for lower_is_better metrics are negated in _score() so
            # that higher numeric values always indicate better performance
            # (matching sklearn's sign convention).  rankdata(-values) then
            # correctly assigns rank 1 to the best candidate.
            array = np.array(array, dtype=np.float64).reshape(n_candidates, n_splits)
            if splits:
                for split_idx in range(n_splits):
                    results[f"split{split_idx}_{key_name}"] = array[:, split_idx]

            array_means = np.average(array, axis=1, weights=weights)
            results[f"mean_{key_name}"] = array_means

            if key_name.startswith(("train_", "test_")) and np.any(~np.isfinite(array_means)):
                warnings.warn(
                    f"One or more of the {key_name.split('_')[0]} scores are non-finite: {array_means}",
                    stacklevel=2,
                    category=UserWarning,
                )

            # Weighted std is not directly available in np
            # However, std = sqrt(mean(abs(x - mean(x))^2))
            # The weighted variance is then:
            # var = mean(w * (x - mean(x))^2)
            if weights is not None:
                array_stds = np.sqrt(np.average((array - array_means[:, np.newaxis]) ** 2, axis=1, weights=weights))
            else:
                array_stds = np.std(array, axis=1)
            results[f"std_{key_name}"] = array_stds

            if rank:
                # When the fit/scoring fails, array_means contains NaNs, we
                # will exclude them from the ranking process and consider them
                # as tied with the worst performers.
                if np.isnan(array_means).all():
                    # All fit/scoring routines failed
                    rank_result = np.ones_like(array_means, dtype=np.int32)
                else:
                    min_array_means = np.nanmin(array_means) - 1
                    array_means_ranked = np.nan_to_num(array_means, nan=min_array_means)
                    rank_result = rankdata(-array_means_ranked, method="min").astype(np.int32)
                results[f"rank_{key_name}"] = rank_result

        _store("fit_time", out["fit_time"])
        _store("score_time", out["score_time"])
        # Use one MaskedArray for all the test scores per scorer
        # because it's more memory efficient than storing multiple
        # MaskedArrays.
        test_sample_counts = np.array(out.get("n_test_samples", []), dtype=np.int32)
        if len(test_sample_counts) > 0:
            test_sample_counts = test_sample_counts.reshape(n_candidates, n_splits)

        # Store test scores - test_scores is a dict with scorer names as keys
        for scorer_name in test_scores:
            _store(
                f"test_{scorer_name}",
                test_scores[scorer_name],
                splits=True,
                rank=True,
                weights=test_sample_counts if len(test_sample_counts) > 0 else None,
            )

        if self.return_train_score:
            for scorer_name in train_scores:
                _store(f"train_{scorer_name}", train_scores[scorer_name], splits=True)

        # Store parameters
        results["params"] = candidate_params
        for param_name, param_values in _yield_masked_array_for_each_param(candidate_params):
            results[f"param_{param_name}"] = param_values

        if more_results is not None:
            for key, value in more_results.items():
                results[key] = value

        return results

    def get_metadata_routing(self):
        """Get metadata routing for this object.

        Returns
        -------
        routing : MetadataRouter
            A MetadataRouter encapsulating routing information.
        """
        router = MetadataRouter(owner=self)

        # Add forecaster routing
        router.add(
            forecaster=self.forecaster,
            method_mapping=MethodMapping()
            .add(caller="fit", callee="fit")
            .add(caller="predict", callee="predict")
            .add(caller="predict_interval", callee="predict_interval")
            .add(caller="predict_class_proba", callee="predict_class_proba")
            .add(caller="observe_predict", callee="observe_predict")
            .add(caller="observe_predict_interval", callee="observe_predict_interval")
            .add(caller="observe_predict_class_proba", callee="observe_predict_class_proba"),
        )

        # Add scorer routing
        scorers, _ = self._get_scorers()
        # Always add as "scorer" regardless of single or multi-metric
        # _MultimetricScorer handles routing internally
        router.add(
            scorer=scorers,
            method_mapping=MethodMapping().add(caller="fit", callee="score"),
        )

        # Add CV splitter routing (if applicable)
        router.add(
            splitter=self.cv,
            method_mapping=MethodMapping().add(caller="fit", callee="split"),
        )

        return router

    @_fit_context(prefer_skip_nested_validation=True)
    def fit(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        forecasting_horizon: int = 1,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> BaseSearchCV:
        """Run fit with all sets of parameters.

        Performs cross-validated hyperparameter search using the specified
        parameter grid/distributions. For each parameter combination, fits
        the forecaster on each training fold and evaluates on the test fold.
        Optionally refits the best forecaster on the entire dataset if
        ``refit=True``.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series with a ``"time"`` column (datetime) and one
            or more numeric value columns.
        X_actual : pl.DataFrame or None, default=None
            Actual feature observations with a ``"time"`` column aligned
            with ``y``. Processed by the feature transformer to produce
            lags, rolling statistics, and other derived features. If
            ``None``, only target-derived features are used.
        forecasting_horizon : int, default=1
            Number of time steps to forecast into the future.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        self
            The fitted search instance.

        """
        _raise_for_params(params, self, "fit")

        # Validate input data for search

        validate_search_data(y, X_actual)

        scorers, refit_metric = self._get_scorers()

        # Validate forecaster/scorer type compatibility (step 1)
        _validate_forecaster_scorer_compatibility(self.forecaster, scorers)

        # Determine prediction context from scorer response methods
        response_method = _resolve_response_method(scorers)
        collected_coverage_rates = _collect_coverage_rates(scorers) if response_method == "predict_interval" else None

        y, X_actual = indexable(y, X_actual)
        params = _check_method_params(y, params=params)

        routed_params = self._get_routed_params_for_fit(params)

        # Get CV splitter
        cv_orig = check_cv(self.cv, forecasting_horizon)
        n_splits = cv_orig.get_n_splits(y, X_actual, **routed_params.splitter.split)

        base_forecaster = clone(self.forecaster)

        parallel = Parallel(n_jobs=self.n_jobs, pre_dispatch=self.pre_dispatch)

        fit_and_score_kwargs = {
            "scorer": scorers,
            "fit_params": routed_params.forecaster.fit,
            "predict_func_params": getattr(routed_params.forecaster, response_method, {}),
            "score_params": routed_params.scorer.score,
            "return_train_score": self.return_train_score,
            "return_n_test_samples": True,
            "return_times": True,
            "return_parameters": False,
            "error_score": self.error_score,
            "verbose": self.verbose,
            "coverage_rates": collected_coverage_rates,
        }
        results = {}
        with parallel:
            all_candidate_params = []
            all_out = []
            all_more_results = defaultdict(list)

            def evaluate_candidates(candidate_params, cv=None, more_results=None):
                """Evaluate candidate parameter settings using cross-validation."""
                cv = cv or cv_orig
                candidate_params = list(candidate_params)
                n_candidates = len(candidate_params)

                if self.verbose > 0:
                    pass

                out = parallel(
                    delayed(_fit_and_score)(
                        clone(base_forecaster),
                        y,
                        X_actual,
                        forecasting_horizon,
                        X_future=X_future,
                        X_forecast=X_forecast,
                        train=train,
                        test=test,
                        parameters=parameters,
                        split_progress=(split_idx, n_splits),
                        candidate_progress=(cand_idx, n_candidates),
                        **fit_and_score_kwargs,
                    )
                    for cand_idx, parameters in enumerate(candidate_params)
                    for split_idx, (train, test) in enumerate(cv.split(y, X_actual, **routed_params.splitter.split))
                )

                if len(out) < 1:
                    raise ValueError("No fits were performed. Was the CV iterator empty? Were there no candidates?")
                elif len(out) != n_candidates * n_splits:
                    raise ValueError(
                        f"cv.split and cv.get_n_splits returned "
                        f"inconsistent results. Expected {n_candidates * n_splits} "
                        f"splits, got {len(out)}"
                    )

                _warn_or_raise_about_fit_failures(out, self.error_score)

                # For callable self.scoring, the return type is only known after
                # calling. If the return type is a dictionary, the error scores
                # can now be inserted with the correct key.
                if callable(self.scoring):
                    _insert_error_scores(out, self.error_score)

                all_candidate_params.extend(candidate_params)
                all_out.extend(out)

                if more_results is not None:
                    for key, value in more_results.items():
                        all_more_results[key].extend(value)

                nonlocal results
                results = self._format_results(all_candidate_params, n_splits, all_out, all_more_results)

                return results

            self._run_search(evaluate_candidates)

            # multimetric is determined here because in the case of a callable
            # self.scoring the return type is only known after calling
            first_test_score = all_out[0]["test_scores"]
            self.multimetric_ = isinstance(first_test_score, dict)

            # check refit_metric now for a callable scorer that is multimetric
            if callable(self.scoring) and self.multimetric_:
                self._check_refit_for_multimetric(first_test_score)
                refit_metric = self.refit

        # For multi-metric evaluation, store the best_index_, best_params_ and
        # best_score_ iff refit is one of the scorer names
        # In single metric evaluation, refit_metric is "score"
        if self.refit or not self.multimetric_:
            self.best_index_ = self._select_best_index(self.refit, refit_metric, results)
            if not callable(self.refit):
                # With a non-custom callable, we can select the best score
                # based on the best index
                self.best_score_ = results[f"mean_test_{refit_metric}"][self.best_index_]
            self.best_params_ = results["params"][self.best_index_]

        if self.refit:
            # Clone the forecaster and parameters for refitting
            self.best_forecaster_ = clone(base_forecaster).set_params(**clone(self.best_params_, safe=False))

            refit_start_time = time.time()

            refit_params = dict(routed_params.forecaster.fit.items())
            if collected_coverage_rates is not None:
                refit_params["coverage_rates"] = collected_coverage_rates
            self.best_forecaster_.fit(
                y,
                X_actual,
                forecasting_horizon,
                X_future=X_future,
                X_forecast=X_forecast,
                **refit_params,
            )
            refit_end_time = time.time()
            self.refit_time_ = refit_end_time - refit_start_time

            # Fit scorers on the training data if they have a fit method
            if isinstance(scorers, _MultimetricScorer) or hasattr(scorers, "fit"):
                scorers.fit(y, forecaster=self.best_forecaster_)

        # Store the scorer (not as dict for single metric evaluation)
        if isinstance(scorers, _MultimetricScorer):
            self.scorer_ = scorers._scorers
        else:
            self.scorer_ = scorers

        self.cv_results_ = results
        self.n_splits_ = n_splits

        return self

    @available_if(_search_forecaster_has("predict"))
    def predict(
        self,
        forecasting_horizon: int | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Generate point forecasts using the best forecaster.

        Parameters
        ----------
        forecasting_horizon : int or None, default=None
            Number of time steps to forecast into the future.  If ``None``,
            uses the horizon specified at fit time.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Point predictions with ``"vintage_time"``, ``"time"``, and one
            column per target variable.

        """
        check_is_fitted(self)
        _raise_for_params(params, self, "predict")
        return self.best_forecaster_.predict(
            forecasting_horizon=forecasting_horizon,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
            **params,
        )

    @available_if(_search_forecaster_has("interval"))
    def predict_interval(
        self,
        forecasting_horizon: int | None = None,
        coverage_rates: list[float] | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Generate interval forecasts using the best forecaster.

        Parameters
        ----------
        forecasting_horizon : int or None, default=None
            Number of time steps to forecast into the future.  If ``None``,
            uses the horizon specified at fit time.
        coverage_rates : list of float or None, default=None
            Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
            for 90 % and 95 % intervals).  If ``None``, defaults to the rates
            used at fit time.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Interval predictions with ``"vintage_time"``, ``"time"``, and
            lower/upper bound columns for each target at each coverage rate.

        """
        check_is_fitted(self)
        _raise_for_params(params, self, "predict_interval")
        return self.best_forecaster_.predict_interval(
            forecasting_horizon=forecasting_horizon,
            coverage_rates=coverage_rates,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
            **params,
        )

    @available_if(_search_forecaster_has("observe"))
    def observe(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
    ) -> BaseSearchCV:
        """Observe new data with the best forecaster without refitting.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series with a ``"time"`` column (datetime) and one
            or more numeric value columns.
        X_actual : pl.DataFrame or None, default=None
            New actual feature observations with a ``"time"`` column
            aligned with ``y``. Forwarded to the best forecaster.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns.

        Returns
        -------
        self
            The search object with updated observation buffers.

        """
        check_is_fitted(self)
        self.best_forecaster_.observe(y, X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
        return self

    @available_if(_search_forecaster_has("rewind"))
    def rewind(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
    ) -> BaseSearchCV:
        """Rewind the best forecaster observation buffers.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series with a ``"time"`` column (datetime) and one
            or more numeric value columns.
        X_actual : pl.DataFrame or None, default=None
            Actual feature observations to restore the observation state
            to. Must align with ``y``.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns.

        Returns
        -------
        self
            The search object with rewound observation buffers.

        """
        check_is_fitted(self)
        self.best_forecaster_.rewind(y, X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
        return self

    @available_if(_search_forecaster_has("observe_predict"))
    def observe_predict(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        forecasting_horizon: int | None = None,
        groups: list[str] | None = None,
        stride: int | None = None,
        predict_transformed: bool = False,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Observe new data and generate point forecasts.

        Equivalent to calling ``observe(y, X_actual)`` then ``predict()``.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series with a ``"time"`` column (datetime) and one
            or more numeric value columns.
        X_actual : pl.DataFrame or None, default=None
            Actual feature observations with a ``"time"`` column aligned
            with ``y``. Sliced and observed incrementally at each step
            of the rolling loop.
        forecasting_horizon : int or None, default=None
            Number of time steps to forecast into the future.  If ``None``,
            uses the horizon specified at fit time.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.
        stride : int or None, default=None
            Step size for rolling update-predict.  If ``None``, defaults to
            ``forecasting_horizon``.
        predict_transformed : bool, default=False
            If ``True``, return predictions in the transformed space without
            applying inverse target transformation.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Point predictions with ``"vintage_time"``, ``"time"``, and one
            column per target variable.

        """
        check_is_fitted(self)
        _raise_for_params(params, self, "observe_predict")
        return self.best_forecaster_.observe_predict(
            y,
            X_actual,
            forecasting_horizon,
            groups,
            stride,
            predict_transformed,
            X_future=X_future,
            X_forecast=X_forecast,
            **params,
        )

    @available_if(_search_forecaster_has("observe_predict_interval"))
    def observe_predict_interval(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        coverage_rates: list[float] | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Observe new data and generate interval forecasts.

        Equivalent to calling ``observe(y, X_actual)`` then
        ``predict_interval()``.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series with a ``"time"`` column (datetime) and one
            or more numeric value columns.
        X_actual : pl.DataFrame or None, default=None
            Actual feature observations with a ``"time"`` column aligned
            with ``y``. Sliced and observed incrementally at each step
            of the rolling loop.
        coverage_rates : list of float or None, default=None
            Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
            for 90 % and 95 % intervals).  If ``None``, defaults to the rates
            used at fit time.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Interval predictions with ``"vintage_time"``, ``"time"``, and
            lower/upper bound columns for each target at each coverage rate.

        """
        check_is_fitted(self)
        _raise_for_params(params, self, "observe_predict_interval")
        return self.best_forecaster_.observe_predict_interval(
            y,
            X_actual,
            coverage_rates=coverage_rates,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
            **params,
        )

    @available_if(_search_forecaster_has("class_proba"))
    def predict_class_proba(
        self,
        forecasting_horizon: int | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Generate class-probability forecasts using the best forecaster.

        Parameters
        ----------
        forecasting_horizon : int or None, default=None
            Number of time steps to forecast into the future.  If ``None``,
            uses the horizon specified at fit time.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Class-probability predictions with ``"vintage_time"``,
            ``"time"``, and probability columns for each target.

        """
        check_is_fitted(self)
        _raise_for_params(params, self, "predict_class_proba")
        return self.best_forecaster_.predict_class_proba(
            forecasting_horizon=forecasting_horizon,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
            **params,
        )

    @available_if(_search_forecaster_has("observe_predict_class_proba"))
    def observe_predict_class_proba(
        self,
        y: pl.DataFrame,
        X_actual: pl.DataFrame | None = None,
        groups: list[str] | None = None,
        X_future: pl.DataFrame | None = None,
        X_forecast: pl.DataFrame | None = None,
        **params,
    ) -> pl.DataFrame:
        """Observe new data and generate class-probability forecasts.

        Equivalent to calling ``observe(y, X_actual)`` then
        ``predict_class_proba()``.

        Parameters
        ----------
        y : pl.DataFrame
            Target time series with a ``"time"`` column (datetime) and one
            or more categorical value columns.
        X_actual : pl.DataFrame or None, default=None
            Actual feature observations with a ``"time"`` column aligned
            with ``y``. Sliced and observed incrementally at each step
            of the rolling loop.
        groups : list of str or None, default=None
            Panel group prefixes to operate on.  If ``None``, all groups
            are used.
        X_future : pl.DataFrame or None, default=None
            Known future features with a ``"time"`` column.
        X_forecast : pl.DataFrame or None, default=None
            External forecasts with ``"vintage_time"`` and ``"time"``
            columns.
        **params : dict
            Metadata to route to nested estimators.

        Returns
        -------
        pl.DataFrame
            Class-probability predictions with ``"vintage_time"``,
            ``"time"``, and probability columns for each target.

        """
        check_is_fitted(self)
        _raise_for_params(params, self, "observe_predict_class_proba")
        return self.best_forecaster_.observe_predict_class_proba(
            y,
            X_actual,
            groups=groups,
            X_future=X_future,
            X_forecast=X_forecast,
            **params,
        )

Methods

n_features_in_ property

Number of features seen during fit.

feature_names_in_ property

Names of features seen during fit.

interval_ property

Time interval detected during fit.

groups_ property

Panel group names detected during fit.

__init_subclass__(**kwargs)

Merge parameter constraints from all classes in the MRO.

Source Code

View on GitHub

Show/Hide source

def __init_subclass__(cls, **kwargs: Any) -> None:
    """Merge parameter constraints from all classes in the MRO."""
    super().__init_subclass__(**kwargs)
    merged: dict = {}
    for klass in reversed(cls.__mro__):
        own = klass.__dict__.get("_parameter_constraints")
        if own and isinstance(own, dict):
            merged.update(own)
    cls._parameter_constraints = merged

__sklearn_tags__()

Get tags from best_forecaster_ if available.

Source Code

View on GitHub

Show/Hide source

def __sklearn_tags__(self):
    """Get tags from best_forecaster_ if available."""
    if hasattr(self, "best_forecaster_"):
        return self.best_forecaster_.__sklearn_tags__()
    # Return tags from the base forecaster during initialization
    return self.forecaster.__sklearn_tags__()

get_metadata_routing()

Get metadata routing for this object.

Returns

Name	Type	Description
routing	MetadataRouter	A MetadataRouter encapsulating routing information.

Source Code

View on GitHub

Show/Hide source

def get_metadata_routing(self):
    """Get metadata routing for this object.

    Returns
    -------
    routing : MetadataRouter
        A MetadataRouter encapsulating routing information.
    """
    router = MetadataRouter(owner=self)

    # Add forecaster routing
    router.add(
        forecaster=self.forecaster,
        method_mapping=MethodMapping()
        .add(caller="fit", callee="fit")
        .add(caller="predict", callee="predict")
        .add(caller="predict_interval", callee="predict_interval")
        .add(caller="predict_class_proba", callee="predict_class_proba")
        .add(caller="observe_predict", callee="observe_predict")
        .add(caller="observe_predict_interval", callee="observe_predict_interval")
        .add(caller="observe_predict_class_proba", callee="observe_predict_class_proba"),
    )

    # Add scorer routing
    scorers, _ = self._get_scorers()
    # Always add as "scorer" regardless of single or multi-metric
    # _MultimetricScorer handles routing internally
    router.add(
        scorer=scorers,
        method_mapping=MethodMapping().add(caller="fit", callee="score"),
    )

    # Add CV splitter routing (if applicable)
    router.add(
        splitter=self.cv,
        method_mapping=MethodMapping().add(caller="fit", callee="split"),
    )

    return router

fit(y, X_actual=None, forecasting_horizon=1, X_future=None, X_forecast=None, **params)

Run fit with all sets of parameters.

Performs cross-validated hyperparameter search using the specified parameter grid/distributions. For each parameter combination, fits the forecaster on each training fold and evaluates on the test fold. Optionally refits the best forecaster on the entire dataset if refit=True.

Parameters

Name	Type	Description	Default
y	DataFrame	Target time series with a "time" column (datetime) and one or more numeric value columns.	required
X_actual	DataFrame or None	Actual feature observations with a "time" column aligned with y. Processed by the feature transformer to produce lags, rolling statistics, and other derived features. If None, only target-derived features are used.	None
forecasting_horizon	int	Number of time steps to forecast into the future.	1
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns.	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
self	The fitted search instance.

Source Code

View on GitHub

Show/Hide source

@_fit_context(prefer_skip_nested_validation=True)
def fit(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    forecasting_horizon: int = 1,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> BaseSearchCV:
    """Run fit with all sets of parameters.

    Performs cross-validated hyperparameter search using the specified
    parameter grid/distributions. For each parameter combination, fits
    the forecaster on each training fold and evaluates on the test fold.
    Optionally refits the best forecaster on the entire dataset if
    ``refit=True``.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series with a ``"time"`` column (datetime) and one
        or more numeric value columns.
    X_actual : pl.DataFrame or None, default=None
        Actual feature observations with a ``"time"`` column aligned
        with ``y``. Processed by the feature transformer to produce
        lags, rolling statistics, and other derived features. If
        ``None``, only target-derived features are used.
    forecasting_horizon : int, default=1
        Number of time steps to forecast into the future.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    self
        The fitted search instance.

    """
    _raise_for_params(params, self, "fit")

    # Validate input data for search

    validate_search_data(y, X_actual)

    scorers, refit_metric = self._get_scorers()

    # Validate forecaster/scorer type compatibility (step 1)
    _validate_forecaster_scorer_compatibility(self.forecaster, scorers)

    # Determine prediction context from scorer response methods
    response_method = _resolve_response_method(scorers)
    collected_coverage_rates = _collect_coverage_rates(scorers) if response_method == "predict_interval" else None

    y, X_actual = indexable(y, X_actual)
    params = _check_method_params(y, params=params)

    routed_params = self._get_routed_params_for_fit(params)

    # Get CV splitter
    cv_orig = check_cv(self.cv, forecasting_horizon)
    n_splits = cv_orig.get_n_splits(y, X_actual, **routed_params.splitter.split)

    base_forecaster = clone(self.forecaster)

    parallel = Parallel(n_jobs=self.n_jobs, pre_dispatch=self.pre_dispatch)

    fit_and_score_kwargs = {
        "scorer": scorers,
        "fit_params": routed_params.forecaster.fit,
        "predict_func_params": getattr(routed_params.forecaster, response_method, {}),
        "score_params": routed_params.scorer.score,
        "return_train_score": self.return_train_score,
        "return_n_test_samples": True,
        "return_times": True,
        "return_parameters": False,
        "error_score": self.error_score,
        "verbose": self.verbose,
        "coverage_rates": collected_coverage_rates,
    }
    results = {}
    with parallel:
        all_candidate_params = []
        all_out = []
        all_more_results = defaultdict(list)

        def evaluate_candidates(candidate_params, cv=None, more_results=None):
            """Evaluate candidate parameter settings using cross-validation."""
            cv = cv or cv_orig
            candidate_params = list(candidate_params)
            n_candidates = len(candidate_params)

            if self.verbose > 0:
                pass

            out = parallel(
                delayed(_fit_and_score)(
                    clone(base_forecaster),
                    y,
                    X_actual,
                    forecasting_horizon,
                    X_future=X_future,
                    X_forecast=X_forecast,
                    train=train,
                    test=test,
                    parameters=parameters,
                    split_progress=(split_idx, n_splits),
                    candidate_progress=(cand_idx, n_candidates),
                    **fit_and_score_kwargs,
                )
                for cand_idx, parameters in enumerate(candidate_params)
                for split_idx, (train, test) in enumerate(cv.split(y, X_actual, **routed_params.splitter.split))
            )

            if len(out) < 1:
                raise ValueError("No fits were performed. Was the CV iterator empty? Were there no candidates?")
            elif len(out) != n_candidates * n_splits:
                raise ValueError(
                    f"cv.split and cv.get_n_splits returned "
                    f"inconsistent results. Expected {n_candidates * n_splits} "
                    f"splits, got {len(out)}"
                )

            _warn_or_raise_about_fit_failures(out, self.error_score)

            # For callable self.scoring, the return type is only known after
            # calling. If the return type is a dictionary, the error scores
            # can now be inserted with the correct key.
            if callable(self.scoring):
                _insert_error_scores(out, self.error_score)

            all_candidate_params.extend(candidate_params)
            all_out.extend(out)

            if more_results is not None:
                for key, value in more_results.items():
                    all_more_results[key].extend(value)

            nonlocal results
            results = self._format_results(all_candidate_params, n_splits, all_out, all_more_results)

            return results

        self._run_search(evaluate_candidates)

        # multimetric is determined here because in the case of a callable
        # self.scoring the return type is only known after calling
        first_test_score = all_out[0]["test_scores"]
        self.multimetric_ = isinstance(first_test_score, dict)

        # check refit_metric now for a callable scorer that is multimetric
        if callable(self.scoring) and self.multimetric_:
            self._check_refit_for_multimetric(first_test_score)
            refit_metric = self.refit

    # For multi-metric evaluation, store the best_index_, best_params_ and
    # best_score_ iff refit is one of the scorer names
    # In single metric evaluation, refit_metric is "score"
    if self.refit or not self.multimetric_:
        self.best_index_ = self._select_best_index(self.refit, refit_metric, results)
        if not callable(self.refit):
            # With a non-custom callable, we can select the best score
            # based on the best index
            self.best_score_ = results[f"mean_test_{refit_metric}"][self.best_index_]
        self.best_params_ = results["params"][self.best_index_]

    if self.refit:
        # Clone the forecaster and parameters for refitting
        self.best_forecaster_ = clone(base_forecaster).set_params(**clone(self.best_params_, safe=False))

        refit_start_time = time.time()

        refit_params = dict(routed_params.forecaster.fit.items())
        if collected_coverage_rates is not None:
            refit_params["coverage_rates"] = collected_coverage_rates
        self.best_forecaster_.fit(
            y,
            X_actual,
            forecasting_horizon,
            X_future=X_future,
            X_forecast=X_forecast,
            **refit_params,
        )
        refit_end_time = time.time()
        self.refit_time_ = refit_end_time - refit_start_time

        # Fit scorers on the training data if they have a fit method
        if isinstance(scorers, _MultimetricScorer) or hasattr(scorers, "fit"):
            scorers.fit(y, forecaster=self.best_forecaster_)

    # Store the scorer (not as dict for single metric evaluation)
    if isinstance(scorers, _MultimetricScorer):
        self.scorer_ = scorers._scorers
    else:
        self.scorer_ = scorers

    self.cv_results_ = results
    self.n_splits_ = n_splits

    return self

predict(forecasting_horizon=None, groups=None, X_future=None, X_forecast=None, **params)

Generate point forecasts using the best forecaster.

Parameters

Name	Type	Description	Default
forecasting_horizon	int or None	Number of time steps to forecast into the future. If None, uses the horizon specified at fit time.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used.	None
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns.	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Point predictions with "vintage_time", "time", and one column per target variable.

Source Code

View on GitHub

Show/Hide source

@available_if(_search_forecaster_has("predict"))
def predict(
    self,
    forecasting_horizon: int | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Generate point forecasts using the best forecaster.

    Parameters
    ----------
    forecasting_horizon : int or None, default=None
        Number of time steps to forecast into the future.  If ``None``,
        uses the horizon specified at fit time.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Point predictions with ``"vintage_time"``, ``"time"``, and one
        column per target variable.

    """
    check_is_fitted(self)
    _raise_for_params(params, self, "predict")
    return self.best_forecaster_.predict(
        forecasting_horizon=forecasting_horizon,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
        **params,
    )

predict_interval(forecasting_horizon=None, coverage_rates=None, groups=None, X_future=None, X_forecast=None, **params)

Generate interval forecasts using the best forecaster.

Parameters

Name	Type	Description	Default
forecasting_horizon	int or None	Number of time steps to forecast into the future. If None, uses the horizon specified at fit time.	None
coverage_rates	list of float or None	Coverage levels for prediction intervals (e.g., [0.9, 0.95] for 90 % and 95 % intervals). If None, defaults to the rates used at fit time.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used.	None
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns.	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Interval predictions with "vintage_time", "time", and lower/upper bound columns for each target at each coverage rate.

Source Code

View on GitHub

Show/Hide source

@available_if(_search_forecaster_has("interval"))
def predict_interval(
    self,
    forecasting_horizon: int | None = None,
    coverage_rates: list[float] | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Generate interval forecasts using the best forecaster.

    Parameters
    ----------
    forecasting_horizon : int or None, default=None
        Number of time steps to forecast into the future.  If ``None``,
        uses the horizon specified at fit time.
    coverage_rates : list of float or None, default=None
        Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
        for 90 % and 95 % intervals).  If ``None``, defaults to the rates
        used at fit time.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Interval predictions with ``"vintage_time"``, ``"time"``, and
        lower/upper bound columns for each target at each coverage rate.

    """
    check_is_fitted(self)
    _raise_for_params(params, self, "predict_interval")
    return self.best_forecaster_.predict_interval(
        forecasting_horizon=forecasting_horizon,
        coverage_rates=coverage_rates,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
        **params,
    )

observe(y, X_actual=None, groups=None, X_future=None, X_forecast=None)

Observe new data with the best forecaster without refitting.

Parameters

Name	Type	Description	Default
y	DataFrame	Target time series with a "time" column (datetime) and one or more numeric value columns.	required
X_actual	DataFrame or None	New actual feature observations with a "time" column aligned with y. Forwarded to the best forecaster.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used.	None
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns.	None

Returns

Type	Description
self	The search object with updated observation buffers.

Source Code

View on GitHub

Show/Hide source

@available_if(_search_forecaster_has("observe"))
def observe(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
) -> BaseSearchCV:
    """Observe new data with the best forecaster without refitting.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series with a ``"time"`` column (datetime) and one
        or more numeric value columns.
    X_actual : pl.DataFrame or None, default=None
        New actual feature observations with a ``"time"`` column
        aligned with ``y``. Forwarded to the best forecaster.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns.

    Returns
    -------
    self
        The search object with updated observation buffers.

    """
    check_is_fitted(self)
    self.best_forecaster_.observe(y, X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
    return self

rewind(y, X_actual=None, groups=None, X_future=None, X_forecast=None)

Rewind the best forecaster observation buffers.

Parameters

Name	Type	Description	Default
y	DataFrame	Target time series with a "time" column (datetime) and one or more numeric value columns.	required
X_actual	DataFrame or None	Actual feature observations to restore the observation state to. Must align with y.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used.	None
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns.	None

Returns

Type	Description
self	The search object with rewound observation buffers.

Source Code

View on GitHub

Show/Hide source

@available_if(_search_forecaster_has("rewind"))
def rewind(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
) -> BaseSearchCV:
    """Rewind the best forecaster observation buffers.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series with a ``"time"`` column (datetime) and one
        or more numeric value columns.
    X_actual : pl.DataFrame or None, default=None
        Actual feature observations to restore the observation state
        to. Must align with ``y``.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns.

    Returns
    -------
    self
        The search object with rewound observation buffers.

    """
    check_is_fitted(self)
    self.best_forecaster_.rewind(y, X_actual, groups=groups, X_future=X_future, X_forecast=X_forecast)
    return self

observe_predict(y, X_actual=None, forecasting_horizon=None, groups=None, stride=None, predict_transformed=False, X_future=None, X_forecast=None, **params)

Observe new data and generate point forecasts.

Equivalent to calling observe(y, X_actual) then predict().

Parameters

Name	Type	Description	Default
y	DataFrame	Target time series with a "time" column (datetime) and one or more numeric value columns.	required
X_actual	DataFrame or None	Actual feature observations with a "time" column aligned with y. Sliced and observed incrementally at each step of the rolling loop.	None
forecasting_horizon	int or None	Number of time steps to forecast into the future. If None, uses the horizon specified at fit time.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used.	None
stride	int or None	Step size for rolling update-predict. If None, defaults to forecasting_horizon.	None
predict_transformed	bool	If True, return predictions in the transformed space without applying inverse target transformation.	False
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns.	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Point predictions with "vintage_time", "time", and one column per target variable.

Source Code

View on GitHub

Show/Hide source

@available_if(_search_forecaster_has("observe_predict"))
def observe_predict(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    forecasting_horizon: int | None = None,
    groups: list[str] | None = None,
    stride: int | None = None,
    predict_transformed: bool = False,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Observe new data and generate point forecasts.

    Equivalent to calling ``observe(y, X_actual)`` then ``predict()``.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series with a ``"time"`` column (datetime) and one
        or more numeric value columns.
    X_actual : pl.DataFrame or None, default=None
        Actual feature observations with a ``"time"`` column aligned
        with ``y``. Sliced and observed incrementally at each step
        of the rolling loop.
    forecasting_horizon : int or None, default=None
        Number of time steps to forecast into the future.  If ``None``,
        uses the horizon specified at fit time.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.
    stride : int or None, default=None
        Step size for rolling update-predict.  If ``None``, defaults to
        ``forecasting_horizon``.
    predict_transformed : bool, default=False
        If ``True``, return predictions in the transformed space without
        applying inverse target transformation.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Point predictions with ``"vintage_time"``, ``"time"``, and one
        column per target variable.

    """
    check_is_fitted(self)
    _raise_for_params(params, self, "observe_predict")
    return self.best_forecaster_.observe_predict(
        y,
        X_actual,
        forecasting_horizon,
        groups,
        stride,
        predict_transformed,
        X_future=X_future,
        X_forecast=X_forecast,
        **params,
    )

observe_predict_interval(y, X_actual=None, coverage_rates=None, groups=None, X_future=None, X_forecast=None, **params)

Observe new data and generate interval forecasts.

Equivalent to calling observe(y, X_actual) then predict_interval().

Parameters

Name	Type	Description	Default
y	DataFrame	Target time series with a "time" column (datetime) and one or more numeric value columns.	required
X_actual	DataFrame or None	Actual feature observations with a "time" column aligned with y. Sliced and observed incrementally at each step of the rolling loop.	None
coverage_rates	list of float or None	Coverage levels for prediction intervals (e.g., [0.9, 0.95] for 90 % and 95 % intervals). If None, defaults to the rates used at fit time.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used.	None
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns.	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Interval predictions with "vintage_time", "time", and lower/upper bound columns for each target at each coverage rate.

Source Code

View on GitHub

Show/Hide source

@available_if(_search_forecaster_has("observe_predict_interval"))
def observe_predict_interval(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    coverage_rates: list[float] | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Observe new data and generate interval forecasts.

    Equivalent to calling ``observe(y, X_actual)`` then
    ``predict_interval()``.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series with a ``"time"`` column (datetime) and one
        or more numeric value columns.
    X_actual : pl.DataFrame or None, default=None
        Actual feature observations with a ``"time"`` column aligned
        with ``y``. Sliced and observed incrementally at each step
        of the rolling loop.
    coverage_rates : list of float or None, default=None
        Coverage levels for prediction intervals (e.g., ``[0.9, 0.95]``
        for 90 % and 95 % intervals).  If ``None``, defaults to the rates
        used at fit time.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Interval predictions with ``"vintage_time"``, ``"time"``, and
        lower/upper bound columns for each target at each coverage rate.

    """
    check_is_fitted(self)
    _raise_for_params(params, self, "observe_predict_interval")
    return self.best_forecaster_.observe_predict_interval(
        y,
        X_actual,
        coverage_rates=coverage_rates,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
        **params,
    )

predict_class_proba(forecasting_horizon=None, groups=None, X_future=None, X_forecast=None, **params)

Generate class-probability forecasts using the best forecaster.

Parameters

Name	Type	Description	Default
forecasting_horizon	int or None	Number of time steps to forecast into the future. If None, uses the horizon specified at fit time.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used.	None
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns.	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Class-probability predictions with "vintage_time", "time", and probability columns for each target.

Source Code

View on GitHub

Show/Hide source

@available_if(_search_forecaster_has("class_proba"))
def predict_class_proba(
    self,
    forecasting_horizon: int | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Generate class-probability forecasts using the best forecaster.

    Parameters
    ----------
    forecasting_horizon : int or None, default=None
        Number of time steps to forecast into the future.  If ``None``,
        uses the horizon specified at fit time.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Class-probability predictions with ``"vintage_time"``,
        ``"time"``, and probability columns for each target.

    """
    check_is_fitted(self)
    _raise_for_params(params, self, "predict_class_proba")
    return self.best_forecaster_.predict_class_proba(
        forecasting_horizon=forecasting_horizon,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
        **params,
    )

observe_predict_class_proba(y, X_actual=None, groups=None, X_future=None, X_forecast=None, **params)

Observe new data and generate class-probability forecasts.

Equivalent to calling observe(y, X_actual) then predict_class_proba().

Parameters

Name	Type	Description	Default
y	DataFrame	Target time series with a "time" column (datetime) and one or more categorical value columns.	required
X_actual	DataFrame or None	Actual feature observations with a "time" column aligned with y. Sliced and observed incrementally at each step of the rolling loop.	None
groups	list of str or None	Panel group prefixes to operate on. If None, all groups are used.	None
X_future	DataFrame or None	Known future features with a "time" column.	None
X_forecast	DataFrame or None	External forecasts with "vintage_time" and "time" columns.	None
**params	dict	Metadata to route to nested estimators.	{}

Returns

Type	Description
DataFrame	Class-probability predictions with "vintage_time", "time", and probability columns for each target.

Source Code

View on GitHub

Show/Hide source

@available_if(_search_forecaster_has("observe_predict_class_proba"))
def observe_predict_class_proba(
    self,
    y: pl.DataFrame,
    X_actual: pl.DataFrame | None = None,
    groups: list[str] | None = None,
    X_future: pl.DataFrame | None = None,
    X_forecast: pl.DataFrame | None = None,
    **params,
) -> pl.DataFrame:
    """Observe new data and generate class-probability forecasts.

    Equivalent to calling ``observe(y, X_actual)`` then
    ``predict_class_proba()``.

    Parameters
    ----------
    y : pl.DataFrame
        Target time series with a ``"time"`` column (datetime) and one
        or more categorical value columns.
    X_actual : pl.DataFrame or None, default=None
        Actual feature observations with a ``"time"`` column aligned
        with ``y``. Sliced and observed incrementally at each step
        of the rolling loop.
    groups : list of str or None, default=None
        Panel group prefixes to operate on.  If ``None``, all groups
        are used.
    X_future : pl.DataFrame or None, default=None
        Known future features with a ``"time"`` column.
    X_forecast : pl.DataFrame or None, default=None
        External forecasts with ``"vintage_time"`` and ``"time"``
        columns.
    **params : dict
        Metadata to route to nested estimators.

    Returns
    -------
    pl.DataFrame
        Class-probability predictions with ``"vintage_time"``,
        ``"time"``, and probability columns for each target.

    """
    check_is_fitted(self)
    _raise_for_params(params, self, "observe_predict_class_proba")
    return self.best_forecaster_.observe_predict_class_proba(
        y,
        X_actual,
        groups=groups,
        X_future=X_future,
        X_forecast=X_forecast,
        **params,
    )