API: modeling.tuning.engine

skyulf.modeling._tuning.engine

Hyperparameter Tuner implementation.

Orchestrator module (F-18): holds TuningCalculator.fit/tune and TuningApplier. The individual responsibilities live in sibling leaf modules: params (search-space cleaning + estimator instantiation), splitters (CV splitter construction), metrics (metric validation + scorer resolution), grid_random (the grid/random fold loop), refit (best-model refit + decision-threshold tuning), and strategies/ (halving, optuna, and the shared searcher runner). The underscored methods kept on TuningCalculator are one-line delegates preserved for the existing test surface.

TuningApplier

Bases: BaseModelApplier

Applier for TuningCalculator.

Wraps the base model applier to provide predictions using the refitted best model.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

class TuningApplier(BaseModelApplier):
    """Applier for TuningCalculator.

    Wraps the base model applier to provide predictions using the refitted best model.
    """

    def __init__(self, base_applier: BaseModelApplier):
        """Store the base applier that performs the actual predictions.

        The tuning artifact is a ``(model, tuning_result)`` tuple; unless
        tuned decision thresholds apply, predictions unwrap the model and
        delegate, keeping the tuner transparent at serving time.
        """
        self.base_applier = base_applier

    def predict(
        self,
        df: pd.DataFrame | SkyulfDataFrame,
        model_artifact: Any,
    ) -> pd.Series | Any:
        """Predict with the refit best model, honoring tuned decision thresholds.

        Unwraps the ``(model, tuning_result)`` artifact; when threshold
        tuning (F-13) stored decision thresholds and the model exposes
        probabilities, those replace the model's default decision rule.
        Otherwise prediction delegates to the base applier. An artifact
        that isn't the expected tuple (e.g. a plain fitted model) yields an
        engine-aware all-null placeholder instead of raising, so downstream
        nodes still receive correctly shaped output.
        """
        # model_artifact is (fitted_model, tuning_result)
        if isinstance(model_artifact, tuple) and len(model_artifact) == 2:
            model, tuning_result = model_artifact
            df = prediction_features(df, tuning_result)
            thresholds = getattr(tuning_result, "decision_thresholds", None)
            if (
                thresholds is not None
                and hasattr(model, "predict_proba")
                and hasattr(model, "classes_")
            ):
                # F-13: apply the decision thresholds tuned on the validation
                # split instead of the model's default decision rule.
                X_np, _ = SklearnBridge.to_sklearn(df, validate_features=True)
                y_proba = model.predict_proba(X_np)
                preds = apply_thresholds(y_proba, thresholds, classes=model.classes_)
                index = cast(pd.DataFrame, df).index if hasattr(df, "index") else None
                return pd.Series(preds, index=index)
            return self.base_applier.predict(df, model)
        # Fallback: artifact isn't the expected (model, tuning_result) tuple
        # (e.g. a plain fitted model, before it's wrapped by the tuner). Return
        # an all-null placeholder of the right length/engine instead of
        # crashing - `df.index` doesn't exist on a Polars DataFrame, so build
        # the placeholder in an engine-aware way.
        if hasattr(df, "index"):
            return pd.Series(np.nan, index=cast(pd.DataFrame, df).index)
        return pd.Series(np.full(len(df), np.nan))

    def predict_proba(
        self,
        df: pd.DataFrame | SkyulfDataFrame,
        model_artifact: Any,
    ) -> pd.DataFrame | SkyulfDataFrame | None:
        """Class probabilities from the tuned model, via the base applier.

        Decision thresholds are deliberately not applied — probabilities
        are the pre-decision payload consumers (and threshold tuning
        itself) need untouched. Returns ``None`` when the artifact isn't
        the ``(model, tuning_result)`` tuple the tuner produces.
        """
        if isinstance(model_artifact, tuple) and len(model_artifact) == 2:
            model, result = model_artifact
            df = prediction_features(df, result)
            return self.base_applier.predict_proba(df, model)
        return None

__init__(base_applier)

Store the base applier that performs the actual predictions.

The tuning artifact is a (model, tuning_result) tuple; unless tuned decision thresholds apply, predictions unwrap the model and delegate, keeping the tuner transparent at serving time.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

def __init__(self, base_applier: BaseModelApplier):
    """Store the base applier that performs the actual predictions.

    The tuning artifact is a ``(model, tuning_result)`` tuple; unless
    tuned decision thresholds apply, predictions unwrap the model and
    delegate, keeping the tuner transparent at serving time.
    """
    self.base_applier = base_applier

predict(df, model_artifact)

Predict with the refit best model, honoring tuned decision thresholds.

Unwraps the (model, tuning_result) artifact; when threshold tuning (F-13) stored decision thresholds and the model exposes probabilities, those replace the model's default decision rule. Otherwise prediction delegates to the base applier. An artifact that isn't the expected tuple (e.g. a plain fitted model) yields an engine-aware all-null placeholder instead of raising, so downstream nodes still receive correctly shaped output.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

def predict(
    self,
    df: pd.DataFrame | SkyulfDataFrame,
    model_artifact: Any,
) -> pd.Series | Any:
    """Predict with the refit best model, honoring tuned decision thresholds.

    Unwraps the ``(model, tuning_result)`` artifact; when threshold
    tuning (F-13) stored decision thresholds and the model exposes
    probabilities, those replace the model's default decision rule.
    Otherwise prediction delegates to the base applier. An artifact
    that isn't the expected tuple (e.g. a plain fitted model) yields an
    engine-aware all-null placeholder instead of raising, so downstream
    nodes still receive correctly shaped output.
    """
    # model_artifact is (fitted_model, tuning_result)
    if isinstance(model_artifact, tuple) and len(model_artifact) == 2:
        model, tuning_result = model_artifact
        df = prediction_features(df, tuning_result)
        thresholds = getattr(tuning_result, "decision_thresholds", None)
        if (
            thresholds is not None
            and hasattr(model, "predict_proba")
            and hasattr(model, "classes_")
        ):
            # F-13: apply the decision thresholds tuned on the validation
            # split instead of the model's default decision rule.
            X_np, _ = SklearnBridge.to_sklearn(df, validate_features=True)
            y_proba = model.predict_proba(X_np)
            preds = apply_thresholds(y_proba, thresholds, classes=model.classes_)
            index = cast(pd.DataFrame, df).index if hasattr(df, "index") else None
            return pd.Series(preds, index=index)
        return self.base_applier.predict(df, model)
    # Fallback: artifact isn't the expected (model, tuning_result) tuple
    # (e.g. a plain fitted model, before it's wrapped by the tuner). Return
    # an all-null placeholder of the right length/engine instead of
    # crashing - `df.index` doesn't exist on a Polars DataFrame, so build
    # the placeholder in an engine-aware way.
    if hasattr(df, "index"):
        return pd.Series(np.nan, index=cast(pd.DataFrame, df).index)
    return pd.Series(np.full(len(df), np.nan))

predict_proba(df, model_artifact)

Class probabilities from the tuned model, via the base applier.

Decision thresholds are deliberately not applied — probabilities are the pre-decision payload consumers (and threshold tuning itself) need untouched. Returns None when the artifact isn't the (model, tuning_result) tuple the tuner produces.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

def predict_proba(
    self,
    df: pd.DataFrame | SkyulfDataFrame,
    model_artifact: Any,
) -> pd.DataFrame | SkyulfDataFrame | None:
    """Class probabilities from the tuned model, via the base applier.

    Decision thresholds are deliberately not applied — probabilities
    are the pre-decision payload consumers (and threshold tuning
    itself) need untouched. Returns ``None`` when the artifact isn't
    the ``(model, tuning_result)`` tuple the tuner produces.
    """
    if isinstance(model_artifact, tuple) and len(model_artifact) == 2:
        model, result = model_artifact
        df = prediction_features(df, result)
        return self.base_applier.predict_proba(df, model)
    return None

TuningCalculator

Bases: BaseModelCalculator

Tune a plain BaseModelCalculator and refit it with the best parameters.

fit accepts a TuningConfig or dict with strategy, metric, n_trials, search_space (a mapping of parameter names to candidate lists), and optional cross-validation and strategy_params settings. grid and random use Skyulf's candidate loop; halving_grid and halving_random use sklearn searchers; optuna uses OptunaSearchCV and requires the Optuna integration package.

Examples:

>>> tuner = TuningCalculator(LogisticRegressionCalculator())
>>> model, result = tuner.fit(X, y, {"strategy": "random", "search_space": {"C": [0.1, 1.0]}})

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

class TuningCalculator(BaseModelCalculator):
    """Tune a plain ``BaseModelCalculator`` and refit it with the best parameters.

    ``fit`` accepts a ``TuningConfig`` or dict with ``strategy``, ``metric``,
    ``n_trials``, ``search_space`` (a mapping of parameter names to candidate
    lists), and optional cross-validation and ``strategy_params`` settings.
    ``grid`` and ``random`` use Skyulf's candidate loop; ``halving_grid`` and
    ``halving_random`` use sklearn searchers; ``optuna`` uses OptunaSearchCV
    and requires the Optuna integration package.

    Examples:
        >>> tuner = TuningCalculator(LogisticRegressionCalculator())
        >>> model, result = tuner.fit(X, y, {"strategy": "random", "search_space": {"C": [0.1, 1.0]}})
    """

    def __init__(self, model_calculator: BaseModelCalculator):
        """Store the base calculator whose estimator will be tuned.

        Everything the search needs — model class, default parameters,
        problem type — is read off the wrapped calculator, so one tuner
        serves every sklearn-backed model node without per-model code.
        """
        self.model_calculator = model_calculator

    @property
    def problem_type(self) -> str:
        """The wrapped calculator's problem type (classification/regression/clustering).

        Delegated so splitter construction and metric/scorer resolution see
        the tuner as the base model — tuning must honor the same contract.
        """
        return self.model_calculator.problem_type

    # ------------------------------------------------------------------
    # Thin delegates to the split leaf modules (F-18). Kept so the pinned
    # internal surface — direct calls and monkeypatches in the test suite —
    # keeps resolving on ``TuningCalculator``.
    # ------------------------------------------------------------------

    def _clean_search_space(self, search_space: dict[str, Any]) -> dict[str, Any]:
        return clean_search_space(search_space)

    def pruning_support_reason(self, config: TuningConfig) -> str | None:
        """Inspect compatibility with the legacy direct incremental Optuna loop.

        Uses calculator defaults and the same estimator preparation as tuning.
        Use ``pruning_support`` to include native and CV-fold pruning modes.
        The current pruner selection does not affect capability; the explicit
        ``pruning=False`` opt-out does.
        """
        if config.strategy_params.get("pruning") is False:
            return "pruning=False explicitly disables incremental trial pruning"
        model_class = getattr(self.model_calculator, "model_class", None)
        if model_class is None:
            return "The model does not expose a tunable estimator class"
        optuna_config = replace(config, strategy="optuna")
        estimator, search_config, _, _ = self._prepare_search_estimator(
            model_class, optuna_config, None, None, None, None, None
        )
        return unsupported_pruning_reason(estimator, clean_search_space(search_config.search_space))

    def pruning_support(
        self,
        config: TuningConfig,
        *,
        n_splits: int | None = None,
        preprocessing: bool = False,
    ) -> dict[str, Any]:
        """Describe current pruning support without fitting or reading any data.

        Graph callers supply the actual split count when a validation partition
        overrides CV, and indicate whether fold preprocessing will be attached.
        The selected pruner does not disable capability; the legacy opt-out does.
        """
        if config.strategy_params.get("pruning") is False:
            return {
                "supported": False,
                "mode": "none",
                "reason": "pruning=False explicitly disables trial pruning",
            }
        model_class = getattr(self.model_calculator, "model_class", None)
        if model_class is None:
            return {
                "supported": False,
                "mode": "none",
                "reason": "The model does not expose a tunable estimator class",
            }
        optuna_config = replace(config, strategy="optuna")
        estimator, prepared, _, _ = self._prepare_search_estimator(
            model_class, optuna_config, None, None, None, None, None
        )
        if n_splits is None:
            n_splits = splitters.select_cv_by_type(config, self.problem_type).get_n_splits()
        plan = resolve_pruning_plan(
            estimator,
            clean_search_space(prepared.search_space),
            n_splits=n_splits,
            preprocessing=preprocessing,
        )
        return {"supported": plan.mode != "none", "mode": plan.mode, "reason": plan.reason}

    @staticmethod
    def _instantiate_model(model_class: Any, params: dict[str, Any]) -> Any:
        return instantiate_model(model_class, params)

    def _resolve_scorer(self, metric: str, y: Any) -> Any:
        return resolve_scorer(metric, y, getattr(self.model_calculator, "problem_type", None))

    @staticmethod
    def _is_multiclass_target(y: Any) -> bool:
        return is_multiclass_target(y)

    def _refit_best_model(
        self,
        tuning_result: TuningResult,
        tuning_config: TuningConfig,
        X_np: Any,
        y_np: Any,
        log_callback: Callable[[str], None] | None,
        iteration_callback: Callable[..., None] | None = None,
        sample_weight: Any = None,
    ) -> Any:
        return refit_best_model(
            self.model_calculator,
            tuning_result,
            tuning_config,
            X_np,
            y_np,
            log_callback,
            iteration_callback,
            **weight_kwargs(sample_weight),
        )

    def _resolve_threshold_metric(
        self,
        metric_name: str,
        log_callback: Callable[[str], None] | None,
        pos_label: Any = None,
    ) -> tuple[Callable[[Any, Any], float], str]:
        return resolve_threshold_metric(metric_name, log_callback, pos_label)

    def _tune_decision_thresholds(
        self,
        model: Any,
        tuning_result: TuningResult,
        tuning_config: TuningConfig,
        validation_data: tuple[Any, Any] | None,
        log_callback: Callable[[str], None] | None,
    ) -> None:
        tune_decision_thresholds(
            self.model_calculator,
            model,
            tuning_result,
            tuning_config,
            validation_data,
            log_callback,
        )

    def _fit_and_score_candidate_fold(
        self,
        candidate_idx: int,
        fold_idx: int,
        params: dict[str, Any],
        model_class: Any,
        cv: Any,
        X_any: Any,
        y_any: Any,
        X_arr: Any,
        y_arr: Any,
        train_idx: Any,
        val_idx: Any,
        metric: str,
        log_callback: Callable[[str], None] | None,
        preprocessing: FoldPreprocessor | None = None,
        fold_errors: list[str] | None = None,
        seed_params_overlay: dict[str, Any] | None = None,
        sample_weight: Any = None,
    ) -> float:
        return fit_and_score_candidate_fold(
            candidate_idx,
            fold_idx,
            params,
            model_class,
            cv,
            X_any,
            y_any,
            X_arr,
            y_arr,
            train_idx,
            val_idx,
            metric,
            log_callback,
            preprocessing,
            fold_errors,
            seed_params_overlay,
            model_calculator=self.model_calculator,
            **weight_kwargs(sample_weight),
        )

    def _run_grid_or_random_search(
        self,
        X_for_search: Any,
        y_for_search: Any,
        config: TuningConfig,
        model_class: Any,
        cv: Any,
        metric: str,
        progress_callback: Callable[[int, int, float | None, dict | None], None] | None,
        log_callback: Callable[[str], None] | None,
        preprocessing: FoldPreprocessor | None = None,
        sample_weight: Any = None,
    ) -> TuningResult:
        return run_grid_or_random_search(
            X_for_search,
            y_for_search,
            config,
            model_class,
            cv,
            metric,
            progress_callback,
            log_callback,
            preprocessing,
            model_calculator=self.model_calculator,
            **weight_kwargs(sample_weight),
        )

    @staticmethod
    def _collect_trials(searcher: Any, config: TuningConfig) -> list[dict[str, Any]]:
        return _runner.collect_trials(searcher, config)

    @staticmethod
    def _strip_model_prefix(params: Any) -> Any:
        return _runner.strip_model_prefix(params)

    # ------------------------------------------------------------------
    # Config + input validation (kept on the orchestrator).
    # ------------------------------------------------------------------

    @staticmethod
    def _build_tuning_config(config: dict[str, Any] | TuningConfig) -> TuningConfig:
        """Convert a raw config dict (or an already-built TuningConfig) into a TuningConfig."""
        if isinstance(config, TuningConfig):
            return config
        # Extract valid keys for TuningConfig
        valid_keys = TuningConfig.__annotations__.keys()
        filtered_config = {k: v for k, v in config.items() if k in valid_keys}
        return TuningConfig(**filtered_config)

    @staticmethod
    def _validate_no_nan_inf(
        arr: Any,
        nan_msg: str,
        inf_msg: str,
        object_nan_msg: str,
        *,
        allow_nan: bool = False,
    ) -> None:
        """Raise ValueError if a numpy array contains NaN/Inf (numeric) or NaN (object dtype).

        Many tuning errors ("No trials completed") are actually due to dirty data causing
        instant failures. We catch this early to give a clear message. Object-dtype arrays
        (e.g. mixed dtypes or leftover categorical/string columns that were never encoded)
        are also scanned via pd.isna, since np.isnan/np.isinf raise on non-numeric dtypes.

        ``allow_nan`` skips the NaN checks for models whose configured estimator
        reports native missing-value support; Inf is still rejected.
        """
        if not isinstance(arr, np.ndarray):
            return
        if np.issubdtype(arr.dtype, np.number):
            if not allow_nan and np.isnan(arr).any():
                raise ValueError(nan_msg)
            if np.isinf(arr).any():
                raise ValueError(inf_msg)
        elif arr.dtype == object and not allow_nan and pd.isna(arr).any():
            raise ValueError(object_nan_msg)

    def _allows_missing_features(self, config: TuningConfig) -> bool:
        """Check configured capabilities without fixing parameters that will be searched."""
        model_class = getattr(self.model_calculator, "model_class", None)
        if model_class is None:
            return False
        constructor_params, _ = split_class_weight_params(
            model_class, {**self.model_calculator.default_params, **seed_params(config)}
        )
        estimator = instantiate_model(model_class, constructor_params)
        if _estimator_allows_nan(estimator):
            return True
        # A search can replace a NaN-incompatible default (e.g. a regression
        # tree's absolute_error criterion). Check the library baseline for
        # searched axes rather than rejecting on a value that need not be used.
        # Candidate fits remain authoritative; no search product is enumerated.
        required_params = {
            name
            for name, parameter in inspect.signature(model_class).parameters.items()
            if parameter.default is inspect.Parameter.empty
        }
        fixed_params = {
            key: value
            for key, value in constructor_params.items()
            if key not in config.search_space or key in required_params
        }
        if fixed_params.keys() == constructor_params.keys():
            return False
        return _estimator_allows_nan(instantiate_model(model_class, fixed_params))

    def fit(
        self,
        X: pd.DataFrame | SkyulfDataFrame,
        y: pd.Series | Any,
        config: dict[str, Any] | TuningConfig,
        progress_callback: Callable[[int, int, float | None, dict | None], None] | None = None,
        log_callback: Callable[[str], None] | None = None,
        validation_data: tuple[pd.DataFrame | SkyulfDataFrame, pd.Series | Any] | None = None,
        iteration_callback: Callable[..., None] | None = None,
        preprocessing: FoldPreprocessor | None = None,
        validation_frames: tuple[Any, Any] | None = None,
        *,
        sample_weight: Any = None,
    ) -> Any:
        """Fits the tuner (runs tuning).

        Adapts the generic fit interface to the specific tune method.

        ``preprocessing`` (F-15): re-fits the given preprocessor on
        each candidate fold's training rows during search, so tuning scores
        stop leaking held-out rows into preprocessing statistics. The custom
        ``grid``/``random`` loop applies it directly; ``halving_*``/``optuna``
        get it via a fold-aware meta-estimator whose ``fit`` runs the chain
        inside every searcher-internal fold — safe even for chains that
        resample rows or re-encode the target. Holdout tuning
        (``validation_data`` set) refits the chain on the train rows only and
        scores candidates against the untouched validation split when
        ``validation_frames`` (the pre-transform validation payload) is
        supplied; frameless calls fall back to raw-payload scoring with an
        explicit log instead. When set, the final best-model refit runs the
        full split through the preprocessor once — the artifact serving uses.
        Polars payloads are accepted for both frame pairs — the engine and
        the fold-aware wrap convert them via ``to_pandas()`` with dtypes
        intact.
        """
        sample_weight = _prepare_tuning_weights(
            sample_weight, len(X), preprocessing, self.problem_type
        )
        tuning_config = self._build_tuning_config(config)
        original_columns = tuple(getattr(X, "columns", ()))

        # Core-only console progress: the backend always supplies its own
        # callback (and never sets `progress`), so this only ever fires for
        # direct SDK use.
        reporter: ConsoleTrialReporter | None = None
        if progress_callback is None and tuning_config.progress:
            reporter = ConsoleTrialReporter()
            progress_callback = reporter

        # For Time Series Split, sort data chronologically (and drop the time
        # column from features) before converting to numpy below - numpy has
        # no column names, so this must happen while X still carries them.
        # Mirrors the same fix already applied to perform_cross_validation();
        # without it, tuning with cv_type="time_series_split" silently leaks
        # the time column and evaluates folds out of chronological order.
        X, y, validation_data, validation_frames, split_metadata, sample_weight = (
            _prepare_policy_inputs(
                X,
                y,
                tuning_config,
                self.problem_type,
                validation_data,
                validation_frames,
                preprocessing,
                sample_weight,
            )
        )
        if split_metadata is None:
            X, y, validation_data, validation_frames, sample_weight = _prepare_time_series_data(
                X,
                y,
                tuning_config,
                validation_data,
                validation_frames,
                preprocessing,
                log_callback,
                sample_weight,
            )

        # Convert data to Numpy for tuning
        X_np, y_np = SklearnBridge.to_sklearn((X, y), validate_features=preprocessing is None)

        # --- VALIDATION: Check for NaNs/Inf in Data ---
        # Query the installed estimator rather than a model-name allowlist:
        # support can depend on its configuration (e.g. a tree's criterion).
        # Each candidate's fit still enforces its own constraints, including
        # restrictions not expressed in tags such as tree monotonicity.
        x_allows_nan = self._allows_missing_features(tuning_config)
        # With per-fold refit, X is the pre-transform payload: NaN there is
        # legitimate when the pipeline carries an imputer, which re-runs on
        # each fold's training rows before the model ever sees the data.
        self._validate_no_nan_inf(
            X_np,
            "Input features (X) contain NaN values. Please use an 'Imputer' node before this model.",
            "Input features (X) contain Infinite values. Please scale or clean your data.",
            "Input features (X) contain missing/NaN values. Please use an 'Imputer' node before this model.",
            allow_nan=x_allows_nan or preprocessing is not None,
        )
        self._validate_no_nan_inf(
            y_np,
            "Target variable (y) contains NaN values. Please drop rows with missing targets or impute them.",
            "Target variable (y) contains Infinite values.",
            "Target variable (y) contains missing/NaN values. Please drop rows with missing targets or impute them.",
        )
        # ----------------------------------------------

        validation_data_np = None
        if validation_data:
            X_val, y_val = validation_data
            X_val_np, y_val_np = SklearnBridge.to_sklearn(
                (X_val, y_val), validate_features=preprocessing is None
            )
            validation_data_np = (X_val_np, y_val_np)

        # Wrap the whole candidate/fold search: sklearn's ConvergenceWarning
        # (raised via `warnings.warn`, not the `logging` module) would
        # otherwise only reach the server's stderr, once per fold/candidate,
        # and never surface to the user. Aggregate into a single summary log
        # line instead of one per fold (a full grid/random search can run
        # hundreds of fold fits) and re-emit any other warning category
        # unchanged so existing behavior for those is preserved.
        with warnings.catch_warnings(record=True) as caught:
            warnings.simplefilter("always")
            tuning_result = self.tune(
                X_np,
                y_np,
                tuning_config,
                progress_callback=progress_callback,
                log_callback=log_callback,
                validation_data=validation_data_np,
                preprocessing=preprocessing,
                preprocessing_frames=(X, y) if preprocessing is not None else None,
                validation_frames=validation_frames,
                split_metadata=split_metadata,
                **weight_kwargs(sample_weight),
            )
        _report_search_warnings(caught, log_callback)
        _record_feature_exclusions(tuning_result, original_columns, X)

        # Refit the best model on the full dataset. With per-fold refit
        # enabled, the full dataset is the full-split frame run once through
        # the preprocessor — the same artifact the folds were scored against
        # and what serving will use for predictions.
        if preprocessing is not None:
            X_refit_frame, y_refit_frame, sample_weight = fit_preprocessor(
                preprocessing, X, y, sample_weight
            )
            X_refit, y_refit = SklearnBridge.to_sklearn(
                (X_refit_frame, y_refit_frame), validate_features=True
            )
        else:
            X_refit, y_refit = X_np, y_np
        model = refit_best_model(
            self.model_calculator,
            tuning_result,
            tuning_config,
            X_refit,
            y_refit,
            log_callback,
            iteration_callback=iteration_callback,
            **weight_kwargs(sample_weight),
        )

        # F-13: optionally search a decision threshold for a binary
        # classifier on the validation split. Best-effort — a failure logs and
        # leaves predict() on the default decision rule.
        _finish_threshold_selection(
            self.model_calculator,
            model,
            tuning_result,
            tuning_config,
            validation_data,
            log_callback,
            y,
            y_refit,
            preprocessing,
        )

        if reporter is not None:
            reporter.finish(tuning_result)

        return (model, tuning_result)

    @staticmethod
    def _to_numpy(data: Any) -> Any:
        """Converts a pandas object to a numpy array, leaving numpy arrays unchanged."""
        return data.to_numpy() if hasattr(data, "to_numpy") else data

    def _prepare_search_estimator(
        self,
        model_class: Any,
        config: TuningConfig,
        preprocessing: FoldPreprocessor | None,
        preprocessing_frames: tuple[Any, Any] | None,
        validation_data: tuple[Any, Any] | None,
        validation_frames: tuple[Any, Any] | None,
        log_callback: Callable[[str], None] | None,
        sample_weight: Any = None,
    ) -> tuple[Any, TuningConfig, bool, bool]:
        """Prepare the estimator and parameter routing for the requested search strategy."""
        # ``default_params`` may carry structural args (e.g. an ensemble's
        # resolved ``estimators``); the instantiator filters/routes them safely.
        constructor_params, class_weight = split_class_weight_params(
            model_class, {**self.model_calculator.default_params, **seed_params(config)}
        )
        base_estimator = instantiate_model(model_class, constructor_params)

        # halving/optuna searchers run their CV internally, where the engine's
        # per-fold hook cannot reach; wrap preprocessing + model in one
        # fit-time meta-estimator so the searcher's own folds drive a true
        # per-fold refit (F-15). Preprocessing runs inside ``fit`` on the
        # fold's training rows, so chains that resample rows or re-encode
        # the target are safe here too.
        searcher_strategy = config.strategy in ("halving_grid", "halving_random", "optuna")
        # Nonnative class weights also need a fit-time step, even without
        # preprocessing: each clone must derive weights from its train fold.
        weighted = (
            searcher_strategy
            and not constructor_accepts_class_weight(model_class)
            and (
                "class_weight" in self.model_calculator.default_params
                or "class_weight" in config.search_space
            )
        )
        wrapped = _can_wrap_preprocessing(
            preprocessing,
            searcher_strategy,
            preprocessing_frames,
            validation_data,
            validation_frames,
            log_callback,
        )
        estimator: Any = base_estimator
        search_config = config
        if wrapped or weighted or (searcher_strategy and sample_weight is not None):
            estimator, search_config = _wrap_search_estimator(
                base_estimator,
                config,
                preprocessing,
                preprocessing_frames,
                class_weight,
                wrapped,
                weighted,
                log_callback,
            )
        return estimator, search_config, wrapped, weighted or sample_weight is not None

    def _prepare_search_splitter(
        self,
        X: Any,
        y: Any,
        config: TuningConfig,
        preprocessing: FoldPreprocessor | None,
        preprocessing_frames: tuple[Any, Any] | None,
        validation_data: tuple[Any, Any] | None,
        validation_frames: tuple[Any, Any] | None,
    ) -> tuple[Any, Any, Any]:
        """Build holdout frame splits when available, otherwise the configured CV splitter."""
        # Holdout tuning with per-fold preprocessing refit: the train and
        # validation frames are concatenated (train rows masked -1 in a
        # PredefinedSplit) so every strategy refits the chain on train rows
        # only and scores candidates against the untouched validation split.
        holdout_refit = (
            preprocessing is not None
            and validation_data is not None
            and preprocessing_frames is not None
            and validation_frames is not None
        )
        if holdout_refit and preprocessing_frames is not None and validation_frames is not None:
            cv, X_for_search, y_for_search = splitters.build_predefined_split_cv_frames(
                preprocessing_frames, validation_frames
            )
        else:
            cv, X_for_search, y_for_search = splitters.build_cv_splitter(
                X, y, config, validation_data, self.model_calculator.problem_type
            )
        return cv, X_for_search, y_for_search

    def _run_searcher_and_collect_results(
        self,
        searcher: Any,
        config: TuningConfig,
        metric: str,
        search_data: tuple[Any, Any],
        wrapped: bool,
        weighted: bool,
        validation_data: tuple[Any, Any] | None,
        preprocessing_frames: tuple[Any, Any] | None,
        log_callback: Callable[[str], None] | None,
        sample_weight: Any = None,
    ) -> TuningResult:
        """Execute the searcher and restore public parameter names in its results."""
        X_for_search, y_for_search = search_data
        # 4. Run Search
        # The wrapped pipeline needs named frames (the adapter rebuilds a real
        # FeatureEngineer); numpy conversion would strip column names. In
        # holdout mode the splitter stage already produced the concatenated
        # train+validation frames the PredefinedSplit mask is aligned to.
        if wrapped:
            if validation_data is None and preprocessing_frames is not None:
                X_for_search, y_for_search = preprocessing_frames
            X_arr, y_arr = X_for_search, y_for_search
        else:
            X_arr = self._to_numpy(X_for_search)
            y_arr = self._to_numpy(y_for_search)
        trial_errors = _runner.execute_search(
            searcher, X_arr, y_arr, config, log_callback, **weight_kwargs(sample_weight)
        )

        # 5. Extract Results
        first_trial_error = trial_errors[0] if trial_errors else None
        best_params, best_score = _runner.extract_best_result(searcher, first_trial_error)

        # Collect trials
        trials = _runner.collect_trials(searcher, config)
        if wrapped or weighted:
            best_params = _runner.strip_model_prefix(best_params)
            trials = [
                {**trial, "params": _runner.strip_model_prefix(trial["params"])} for trial in trials
            ]

        # Final completion log for strategies that don't emit per-trial callbacks
        # (halving_grid / halving_random / optuna). The grid/random branch above
        # already logs completion inside its custom loop.
        _runner.log_final_completion(log_callback, config, trials, best_score, best_params)

        return TuningResult(
            best_params=best_params,
            best_score=best_score,
            n_trials=len(trials),
            trials=trials,
            scoring_metric=metric,
        )

    def tune(
        self,
        X: Any,
        y: Any,
        config: TuningConfig,
        progress_callback: Callable[[int, int, float | None, dict | None], None] | None = None,
        log_callback: Callable[[str], None] | None = None,
        validation_data: tuple[Any, Any] | None = None,
        preprocessing: FoldPreprocessor | None = None,
        preprocessing_frames: tuple[Any, Any] | None = None,
        validation_frames: tuple[Any, Any] | None = None,
        split_metadata: dict[str, np.ndarray] | None = None,
        cv_override: Any = None,
        *,
        sample_weight: Any = None,
    ) -> TuningResult:
        """Runs hyperparameter tuning."""
        sample_weight = _prepare_tuning_weights(
            sample_weight, len(X), preprocessing, self.problem_type
        )
        X, y, validation_data, validation_frames, preprocessing_frames, sample_weight = (
            _direct_time_data(
                X,
                y,
                config,
                validation_data,
                validation_frames,
                preprocessing_frames,
                preprocessing,
                log_callback,
                split_metadata,
                sample_weight,
            )
        )
        X, y, preprocessing_frames, split_metadata, sample_weight = _policy_search_data(
            X,
            y,
            config,
            self.problem_type,
            preprocessing,
            preprocessing_frames,
            split_metadata,
            sample_weight,
        )
        if config.cv_enabled and config.cv_type == "nested_cv":
            raw_x, raw_y = preprocessing_frames if preprocessing_frames is not None else (X, y)
            return run_nested_search(
                self,
                raw_x,
                raw_y,
                config,
                preprocessing=preprocessing,
                progress_callback=progress_callback,
                log_callback=log_callback,
                split_metadata=split_metadata,
                **weight_kwargs(sample_weight),
            )
        # 1. Prepare Estimator
        # We need a base estimator. Since our Calculator wraps the class,
        # we need to instantiate the underlying sklearn model with default params.
        # Assuming model_calculator is SklearnCalculator
        if not hasattr(self.model_calculator, "model_class"):
            raise ValueError("Tuner currently only supports SklearnCalculator")

        # `model_class` only on SklearnCalculator; `Any` keeps call sites type-clean.
        model_class: Any = self.model_calculator.model_class

        config = normalize_logistic_search_config(
            model_class, self.model_calculator.default_params, config
        )

        estimator, search_config, wrapped, weighted = self._prepare_search_estimator(
            model_class,
            config,
            preprocessing,
            preprocessing_frames,
            validation_data,
            validation_frames,
            log_callback,
            sample_weight,
        )

        # 2. Prepare Splitter
        # If validation data is provided, use PredefinedSplit to train on X and validate on validation_data
        # Otherwise use CV
        cv, X_for_search, y_for_search = self._prepare_search_splitter(
            X, y, config, preprocessing, preprocessing_frames, validation_data, validation_frames
        )
        cv = _effective_search_cv(cv, cv_override, config, self.problem_type, y, split_metadata)

        sample_weight = search_weights(sample_weight, len(X_for_search))
        preflight_weights(sample_weight, cv, X_for_search, y_for_search)

        # 3. Select Search Strategy
        # Handle multiclass metrics and map user-friendly names
        metric = resolve_metric(config, y, self.model_calculator.problem_type)

        if config.strategy in ["grid", "random"]:
            # Per-fold refit needs named frames (the adapter rebuilds a real
            # FeatureEngineer), not the numpy arrays used by the searchers.
            # Holdout mode is excluded: the splitter stage already produced
            # the concatenated train+validation frames the mask is aligned to.
            X_for_search, y_for_search = self._grid_search_frames(
                (X_for_search, y_for_search), preprocessing, preprocessing_frames, validation_data
            )
            # Use custom loop to support progress and log callbacks
            return run_grid_or_random_search(
                X_for_search,
                y_for_search,
                config,
                model_class,
                cv,
                metric,
                progress_callback,
                log_callback,
                preprocessing,
                model_calculator=self.model_calculator,
                **weight_kwargs(sample_weight),
            )
        elif config.strategy in ["halving_grid", "halving_random", "optuna"]:
            # Searchers score the payload handed to them below: the raw
            # pre-transform frames when wrapped (the fold-aware step encodes
            # inside fit but maps predictions back), otherwise the payload
            # labels. Resolve the scorer against that label space so binary
            # scorers get a valid pos_label even for string targets.
            scoring_y = (
                preprocessing_frames[1]
                if wrapped and preprocessing_frames is not None
                else y_for_search
            )
            scoring = resolve_scorer(
                metric, scoring_y, getattr(self.model_calculator, "problem_type", None)
            )
            if config.strategy in ["halving_grid", "halving_random"]:
                search_config = _halving.bound_sample_resources(search_config, len(X_for_search))
                searcher = _halving.build_halving_searcher(
                    search_config, estimator, cv, scoring, log_callback
                )
            else:
                searcher = _optuna_strategy.build_optuna_searcher(
                    search_config,
                    estimator,
                    cv,
                    scoring,
                    progress_callback,
                    log_callback,
                    weighted=sample_weight is not None,
                )
        else:
            raise_invalid_choice(
                config.strategy,
                ("grid", "random", "halving_grid", "halving_random", "optuna"),
                "tuning strategy",
            )

        return self._run_searcher_and_collect_results(
            searcher,
            config,
            metric,
            (X_for_search, y_for_search),
            wrapped,
            weighted,
            validation_data,
            preprocessing_frames,
            log_callback,
            **weight_kwargs(sample_weight),
        )

    @staticmethod
    def _grid_search_frames(
        search_frames: tuple[Any, Any],
        preprocessing: FoldPreprocessor | None,
        preprocessing_frames: tuple[Any, Any] | None,
        validation_data: tuple[Any, Any] | None,
    ) -> tuple[Any, Any]:
        """Retain named raw frames for per-fold refits outside explicit holdout mode."""
        if (
            preprocessing is not None
            and preprocessing_frames is not None
            and validation_data is None
        ):
            return preprocessing_frames
        return search_frames

problem_type property

The wrapped calculator's problem type (classification/regression/clustering).

Delegated so splitter construction and metric/scorer resolution see the tuner as the base model — tuning must honor the same contract.

__init__(model_calculator)

Store the base calculator whose estimator will be tuned.

Everything the search needs — model class, default parameters, problem type — is read off the wrapped calculator, so one tuner serves every sklearn-backed model node without per-model code.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

def __init__(self, model_calculator: BaseModelCalculator):
    """Store the base calculator whose estimator will be tuned.

    Everything the search needs — model class, default parameters,
    problem type — is read off the wrapped calculator, so one tuner
    serves every sklearn-backed model node without per-model code.
    """
    self.model_calculator = model_calculator

fit(X, y, config, progress_callback=None, log_callback=None, validation_data=None, iteration_callback=None, preprocessing=None, validation_frames=None, *, sample_weight=None)

Fits the tuner (runs tuning).

Adapts the generic fit interface to the specific tune method.

preprocessing (F-15): re-fits the given preprocessor on each candidate fold's training rows during search, so tuning scores stop leaking held-out rows into preprocessing statistics. The custom grid/random loop applies it directly; halving_*/optuna get it via a fold-aware meta-estimator whose fit runs the chain inside every searcher-internal fold — safe even for chains that resample rows or re-encode the target. Holdout tuning (validation_data set) refits the chain on the train rows only and scores candidates against the untouched validation split when validation_frames (the pre-transform validation payload) is supplied; frameless calls fall back to raw-payload scoring with an explicit log instead. When set, the final best-model refit runs the full split through the preprocessor once — the artifact serving uses. Polars payloads are accepted for both frame pairs — the engine and the fold-aware wrap convert them via to_pandas() with dtypes intact.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

def fit(
    self,
    X: pd.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    config: dict[str, Any] | TuningConfig,
    progress_callback: Callable[[int, int, float | None, dict | None], None] | None = None,
    log_callback: Callable[[str], None] | None = None,
    validation_data: tuple[pd.DataFrame | SkyulfDataFrame, pd.Series | Any] | None = None,
    iteration_callback: Callable[..., None] | None = None,
    preprocessing: FoldPreprocessor | None = None,
    validation_frames: tuple[Any, Any] | None = None,
    *,
    sample_weight: Any = None,
) -> Any:
    """Fits the tuner (runs tuning).

    Adapts the generic fit interface to the specific tune method.

    ``preprocessing`` (F-15): re-fits the given preprocessor on
    each candidate fold's training rows during search, so tuning scores
    stop leaking held-out rows into preprocessing statistics. The custom
    ``grid``/``random`` loop applies it directly; ``halving_*``/``optuna``
    get it via a fold-aware meta-estimator whose ``fit`` runs the chain
    inside every searcher-internal fold — safe even for chains that
    resample rows or re-encode the target. Holdout tuning
    (``validation_data`` set) refits the chain on the train rows only and
    scores candidates against the untouched validation split when
    ``validation_frames`` (the pre-transform validation payload) is
    supplied; frameless calls fall back to raw-payload scoring with an
    explicit log instead. When set, the final best-model refit runs the
    full split through the preprocessor once — the artifact serving uses.
    Polars payloads are accepted for both frame pairs — the engine and
    the fold-aware wrap convert them via ``to_pandas()`` with dtypes
    intact.
    """
    sample_weight = _prepare_tuning_weights(
        sample_weight, len(X), preprocessing, self.problem_type
    )
    tuning_config = self._build_tuning_config(config)
    original_columns = tuple(getattr(X, "columns", ()))

    # Core-only console progress: the backend always supplies its own
    # callback (and never sets `progress`), so this only ever fires for
    # direct SDK use.
    reporter: ConsoleTrialReporter | None = None
    if progress_callback is None and tuning_config.progress:
        reporter = ConsoleTrialReporter()
        progress_callback = reporter

    # For Time Series Split, sort data chronologically (and drop the time
    # column from features) before converting to numpy below - numpy has
    # no column names, so this must happen while X still carries them.
    # Mirrors the same fix already applied to perform_cross_validation();
    # without it, tuning with cv_type="time_series_split" silently leaks
    # the time column and evaluates folds out of chronological order.
    X, y, validation_data, validation_frames, split_metadata, sample_weight = (
        _prepare_policy_inputs(
            X,
            y,
            tuning_config,
            self.problem_type,
            validation_data,
            validation_frames,
            preprocessing,
            sample_weight,
        )
    )
    if split_metadata is None:
        X, y, validation_data, validation_frames, sample_weight = _prepare_time_series_data(
            X,
            y,
            tuning_config,
            validation_data,
            validation_frames,
            preprocessing,
            log_callback,
            sample_weight,
        )

    # Convert data to Numpy for tuning
    X_np, y_np = SklearnBridge.to_sklearn((X, y), validate_features=preprocessing is None)

    # --- VALIDATION: Check for NaNs/Inf in Data ---
    # Query the installed estimator rather than a model-name allowlist:
    # support can depend on its configuration (e.g. a tree's criterion).
    # Each candidate's fit still enforces its own constraints, including
    # restrictions not expressed in tags such as tree monotonicity.
    x_allows_nan = self._allows_missing_features(tuning_config)
    # With per-fold refit, X is the pre-transform payload: NaN there is
    # legitimate when the pipeline carries an imputer, which re-runs on
    # each fold's training rows before the model ever sees the data.
    self._validate_no_nan_inf(
        X_np,
        "Input features (X) contain NaN values. Please use an 'Imputer' node before this model.",
        "Input features (X) contain Infinite values. Please scale or clean your data.",
        "Input features (X) contain missing/NaN values. Please use an 'Imputer' node before this model.",
        allow_nan=x_allows_nan or preprocessing is not None,
    )
    self._validate_no_nan_inf(
        y_np,
        "Target variable (y) contains NaN values. Please drop rows with missing targets or impute them.",
        "Target variable (y) contains Infinite values.",
        "Target variable (y) contains missing/NaN values. Please drop rows with missing targets or impute them.",
    )
    # ----------------------------------------------

    validation_data_np = None
    if validation_data:
        X_val, y_val = validation_data
        X_val_np, y_val_np = SklearnBridge.to_sklearn(
            (X_val, y_val), validate_features=preprocessing is None
        )
        validation_data_np = (X_val_np, y_val_np)

    # Wrap the whole candidate/fold search: sklearn's ConvergenceWarning
    # (raised via `warnings.warn`, not the `logging` module) would
    # otherwise only reach the server's stderr, once per fold/candidate,
    # and never surface to the user. Aggregate into a single summary log
    # line instead of one per fold (a full grid/random search can run
    # hundreds of fold fits) and re-emit any other warning category
    # unchanged so existing behavior for those is preserved.
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        tuning_result = self.tune(
            X_np,
            y_np,
            tuning_config,
            progress_callback=progress_callback,
            log_callback=log_callback,
            validation_data=validation_data_np,
            preprocessing=preprocessing,
            preprocessing_frames=(X, y) if preprocessing is not None else None,
            validation_frames=validation_frames,
            split_metadata=split_metadata,
            **weight_kwargs(sample_weight),
        )
    _report_search_warnings(caught, log_callback)
    _record_feature_exclusions(tuning_result, original_columns, X)

    # Refit the best model on the full dataset. With per-fold refit
    # enabled, the full dataset is the full-split frame run once through
    # the preprocessor — the same artifact the folds were scored against
    # and what serving will use for predictions.
    if preprocessing is not None:
        X_refit_frame, y_refit_frame, sample_weight = fit_preprocessor(
            preprocessing, X, y, sample_weight
        )
        X_refit, y_refit = SklearnBridge.to_sklearn(
            (X_refit_frame, y_refit_frame), validate_features=True
        )
    else:
        X_refit, y_refit = X_np, y_np
    model = refit_best_model(
        self.model_calculator,
        tuning_result,
        tuning_config,
        X_refit,
        y_refit,
        log_callback,
        iteration_callback=iteration_callback,
        **weight_kwargs(sample_weight),
    )

    # F-13: optionally search a decision threshold for a binary
    # classifier on the validation split. Best-effort — a failure logs and
    # leaves predict() on the default decision rule.
    _finish_threshold_selection(
        self.model_calculator,
        model,
        tuning_result,
        tuning_config,
        validation_data,
        log_callback,
        y,
        y_refit,
        preprocessing,
    )

    if reporter is not None:
        reporter.finish(tuning_result)

    return (model, tuning_result)

pruning_support(config, *, n_splits=None, preprocessing=False)

Describe current pruning support without fitting or reading any data.

Graph callers supply the actual split count when a validation partition overrides CV, and indicate whether fold preprocessing will be attached. The selected pruner does not disable capability; the legacy opt-out does.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

def pruning_support(
    self,
    config: TuningConfig,
    *,
    n_splits: int | None = None,
    preprocessing: bool = False,
) -> dict[str, Any]:
    """Describe current pruning support without fitting or reading any data.

    Graph callers supply the actual split count when a validation partition
    overrides CV, and indicate whether fold preprocessing will be attached.
    The selected pruner does not disable capability; the legacy opt-out does.
    """
    if config.strategy_params.get("pruning") is False:
        return {
            "supported": False,
            "mode": "none",
            "reason": "pruning=False explicitly disables trial pruning",
        }
    model_class = getattr(self.model_calculator, "model_class", None)
    if model_class is None:
        return {
            "supported": False,
            "mode": "none",
            "reason": "The model does not expose a tunable estimator class",
        }
    optuna_config = replace(config, strategy="optuna")
    estimator, prepared, _, _ = self._prepare_search_estimator(
        model_class, optuna_config, None, None, None, None, None
    )
    if n_splits is None:
        n_splits = splitters.select_cv_by_type(config, self.problem_type).get_n_splits()
    plan = resolve_pruning_plan(
        estimator,
        clean_search_space(prepared.search_space),
        n_splits=n_splits,
        preprocessing=preprocessing,
    )
    return {"supported": plan.mode != "none", "mode": plan.mode, "reason": plan.reason}

pruning_support_reason(config)

Inspect compatibility with the legacy direct incremental Optuna loop.

Uses calculator defaults and the same estimator preparation as tuning. Use pruning_support to include native and CV-fold pruning modes. The current pruner selection does not affect capability; the explicit pruning=False opt-out does.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

def pruning_support_reason(self, config: TuningConfig) -> str | None:
    """Inspect compatibility with the legacy direct incremental Optuna loop.

    Uses calculator defaults and the same estimator preparation as tuning.
    Use ``pruning_support`` to include native and CV-fold pruning modes.
    The current pruner selection does not affect capability; the explicit
    ``pruning=False`` opt-out does.
    """
    if config.strategy_params.get("pruning") is False:
        return "pruning=False explicitly disables incremental trial pruning"
    model_class = getattr(self.model_calculator, "model_class", None)
    if model_class is None:
        return "The model does not expose a tunable estimator class"
    optuna_config = replace(config, strategy="optuna")
    estimator, search_config, _, _ = self._prepare_search_estimator(
        model_class, optuna_config, None, None, None, None, None
    )
    return unsupported_pruning_reason(estimator, clean_search_space(search_config.search_space))

tune(X, y, config, progress_callback=None, log_callback=None, validation_data=None, preprocessing=None, preprocessing_frames=None, validation_frames=None, split_metadata=None, cv_override=None, *, sample_weight=None)

Runs hyperparameter tuning.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

def tune(
    self,
    X: Any,
    y: Any,
    config: TuningConfig,
    progress_callback: Callable[[int, int, float | None, dict | None], None] | None = None,
    log_callback: Callable[[str], None] | None = None,
    validation_data: tuple[Any, Any] | None = None,
    preprocessing: FoldPreprocessor | None = None,
    preprocessing_frames: tuple[Any, Any] | None = None,
    validation_frames: tuple[Any, Any] | None = None,
    split_metadata: dict[str, np.ndarray] | None = None,
    cv_override: Any = None,
    *,
    sample_weight: Any = None,
) -> TuningResult:
    """Runs hyperparameter tuning."""
    sample_weight = _prepare_tuning_weights(
        sample_weight, len(X), preprocessing, self.problem_type
    )
    X, y, validation_data, validation_frames, preprocessing_frames, sample_weight = (
        _direct_time_data(
            X,
            y,
            config,
            validation_data,
            validation_frames,
            preprocessing_frames,
            preprocessing,
            log_callback,
            split_metadata,
            sample_weight,
        )
    )
    X, y, preprocessing_frames, split_metadata, sample_weight = _policy_search_data(
        X,
        y,
        config,
        self.problem_type,
        preprocessing,
        preprocessing_frames,
        split_metadata,
        sample_weight,
    )
    if config.cv_enabled and config.cv_type == "nested_cv":
        raw_x, raw_y = preprocessing_frames if preprocessing_frames is not None else (X, y)
        return run_nested_search(
            self,
            raw_x,
            raw_y,
            config,
            preprocessing=preprocessing,
            progress_callback=progress_callback,
            log_callback=log_callback,
            split_metadata=split_metadata,
            **weight_kwargs(sample_weight),
        )
    # 1. Prepare Estimator
    # We need a base estimator. Since our Calculator wraps the class,
    # we need to instantiate the underlying sklearn model with default params.
    # Assuming model_calculator is SklearnCalculator
    if not hasattr(self.model_calculator, "model_class"):
        raise ValueError("Tuner currently only supports SklearnCalculator")

    # `model_class` only on SklearnCalculator; `Any` keeps call sites type-clean.
    model_class: Any = self.model_calculator.model_class

    config = normalize_logistic_search_config(
        model_class, self.model_calculator.default_params, config
    )

    estimator, search_config, wrapped, weighted = self._prepare_search_estimator(
        model_class,
        config,
        preprocessing,
        preprocessing_frames,
        validation_data,
        validation_frames,
        log_callback,
        sample_weight,
    )

    # 2. Prepare Splitter
    # If validation data is provided, use PredefinedSplit to train on X and validate on validation_data
    # Otherwise use CV
    cv, X_for_search, y_for_search = self._prepare_search_splitter(
        X, y, config, preprocessing, preprocessing_frames, validation_data, validation_frames
    )
    cv = _effective_search_cv(cv, cv_override, config, self.problem_type, y, split_metadata)

    sample_weight = search_weights(sample_weight, len(X_for_search))
    preflight_weights(sample_weight, cv, X_for_search, y_for_search)

    # 3. Select Search Strategy
    # Handle multiclass metrics and map user-friendly names
    metric = resolve_metric(config, y, self.model_calculator.problem_type)

    if config.strategy in ["grid", "random"]:
        # Per-fold refit needs named frames (the adapter rebuilds a real
        # FeatureEngineer), not the numpy arrays used by the searchers.
        # Holdout mode is excluded: the splitter stage already produced
        # the concatenated train+validation frames the mask is aligned to.
        X_for_search, y_for_search = self._grid_search_frames(
            (X_for_search, y_for_search), preprocessing, preprocessing_frames, validation_data
        )
        # Use custom loop to support progress and log callbacks
        return run_grid_or_random_search(
            X_for_search,
            y_for_search,
            config,
            model_class,
            cv,
            metric,
            progress_callback,
            log_callback,
            preprocessing,
            model_calculator=self.model_calculator,
            **weight_kwargs(sample_weight),
        )
    elif config.strategy in ["halving_grid", "halving_random", "optuna"]:
        # Searchers score the payload handed to them below: the raw
        # pre-transform frames when wrapped (the fold-aware step encodes
        # inside fit but maps predictions back), otherwise the payload
        # labels. Resolve the scorer against that label space so binary
        # scorers get a valid pos_label even for string targets.
        scoring_y = (
            preprocessing_frames[1]
            if wrapped and preprocessing_frames is not None
            else y_for_search
        )
        scoring = resolve_scorer(
            metric, scoring_y, getattr(self.model_calculator, "problem_type", None)
        )
        if config.strategy in ["halving_grid", "halving_random"]:
            search_config = _halving.bound_sample_resources(search_config, len(X_for_search))
            searcher = _halving.build_halving_searcher(
                search_config, estimator, cv, scoring, log_callback
            )
        else:
            searcher = _optuna_strategy.build_optuna_searcher(
                search_config,
                estimator,
                cv,
                scoring,
                progress_callback,
                log_callback,
                weighted=sample_weight is not None,
            )
    else:
        raise_invalid_choice(
            config.strategy,
            ("grid", "random", "halving_grid", "halving_random", "optuna"),
            "tuning strategy",
        )

    return self._run_searcher_and_collect_results(
        searcher,
        config,
        metric,
        (X_for_search, y_for_search),
        wrapped,
        weighted,
        validation_data,
        preprocessing_frames,
        log_callback,
        **weight_kwargs(sample_weight),
    )

__getattr__(name)

Read-only views of the optuna load state under its legacy names (F-14).

Reads the live module global so tests that replace this module's _optuna_state binding see their replacement through the views.

Source code in skyulf-core/skyulf/modeling/_tuning/engine.py

def __getattr__(name: str) -> Any:
    """Read-only views of the optuna load state under its legacy names (F-14).

    Reads the live module global so tests that replace this module's
    ``_optuna_state`` binding see their replacement through the views.
    """
    if name == "HAS_OPTUNA":
        return _optuna_state.has_optuna
    if name == "OptunaSearchCV":
        return _optuna_state.search_cv
    if name == "optuna":
        return _optuna_state.optuna_module
    if name == "_optuna_load_attempted":
        return _optuna_state.attempted
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")