API: modeling

skyulf.modeling

Modeling module for Skyulf.

BaseModelApplier

Bases: ABC

Predicting half of a model node: turns a fitted artifact into predictions.

:meth:predict_proba is optional. The base implementation returns None, which is how callers detect an estimator that cannot produce class probabilities.

Source code in skyulf-core/skyulf/modeling/base.py

class BaseModelApplier(ABC):
    """Predicting half of a model node: turns a fitted artifact into predictions.

    :meth:`predict_proba` is optional. The base implementation returns ``None``,
    which is how callers detect an estimator that cannot produce class
    probabilities.
    """

    @abstractmethod
    def predict(self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any) -> pd.Series | Any:
        """Generates predictions."""

    def predict_proba(
        self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any
    ) -> pd.DataFrame | SkyulfDataFrame | None:
        """Generates prediction probabilities if supported.

        Returns DataFrame where columns are classes.
        """
        return None

predict(df, model_artifact) abstractmethod

Generates predictions.

Source code in skyulf-core/skyulf/modeling/base.py

@abstractmethod
def predict(self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any) -> pd.Series | Any:
    """Generates predictions."""

predict_proba(df, model_artifact)

Generates prediction probabilities if supported.

Returns DataFrame where columns are classes.

Source code in skyulf-core/skyulf/modeling/base.py

def predict_proba(
    self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any
) -> pd.DataFrame | SkyulfDataFrame | None:
    """Generates prediction probabilities if supported.

    Returns DataFrame where columns are classes.
    """
    return None

BaseModelCalculator

Bases: ABC

Fitting half of a model node: declares the problem type and trains the estimator.

Subclasses must supply :attr:problem_type and :meth:fit. The :attr:default_params and tuning hooks carry plain-model defaults that structural models such as ensembles override.

Source code in skyulf-core/skyulf/modeling/base.py

class BaseModelCalculator(ABC):
    """Fitting half of a model node: declares the problem type and trains the estimator.

    Subclasses must supply :attr:`problem_type` and :meth:`fit`. The
    :attr:`default_params` and tuning hooks carry plain-model defaults that
    structural models such as ensembles override.
    """

    @property
    @abstractmethod
    def problem_type(self) -> str:
        """Returns 'classification', 'regression', or 'clustering'."""

    #: Config keys that a model's ``prepare_tuning_params`` absorbs into its
    #: own structural state (e.g. an ensemble's resolved ``estimators``)
    #: rather than treating as a literal single-item search-space candidate.
    #: Empty for plain models. See ``_BaseEnsembleCalculator`` in
    #: ``ensemble.py`` for the non-trivial override.
    STRUCTURAL_TUNING_KEYS: tuple[str, ...] = ()

    @property
    def default_params(self) -> dict[str, Any]:
        """Default hyperparameters for the model."""
        return {}

    def prepare_tuning_params(self, config: dict[str, Any]) -> None:
        """Hook for structural models (e.g. ensembles) to absorb their sub-estimator selection.

        Runs before the tuner builds the base model. No-op for plain models.
        Ensembles override this to inject the resolved
        ``estimators`` (and ``final_estimator``) into :attr:`default_params` so
        the tuner can construct a valid meta-estimator.
        """
        return None

    def build_tuning_search_space(self, config: dict[str, Any], strategy: str) -> dict[str, Any]:
        """Hook: let a model auto-build its tuning search space.

        Returns an empty dict for plain models (the caller keeps the
        user-provided space). Ensembles override this to expand their base
        learners' parameter grids into nested ``<name>__<param>`` keys.
        """
        return {}

    def _boosting_fit_kwargs(
        self,
        model: Any,
        X_np: Any,
        y_np: Any,
        iteration_callback: Callable[..., None] | None,
    ) -> dict[str, Any]:
        """Hook: extra kwargs for the underlying ``model.fit(...)`` call.

        Boosting calculators (XGBoost/LightGBM) override this to attach an
        eval set + iteration callback when one is supplied; every other model
        keeps the plain fit. May also mutate ``model`` (XGBoost 3.x carries
        callbacks on the estimator itself); returning ``"_detach_callbacks":
        True`` tells the caller to clear ``model.callbacks`` after fit so the
        saved artifact doesn't pickle live callback closures.
        """
        return {}

    @abstractmethod
    def fit(
        self,
        X: pd.DataFrame | SkyulfDataFrame,
        y: pd.Series | Any,
        config: dict[str, Any],
        progress_callback: Callable[..., 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,
        *,
        sample_weight: Any = None,
    ) -> Any:
        """Trains the model and returns the fitted model artifact.

        ``class_weight`` is a model setting in ``config["params"]`` for
        supported classifiers. ``sample_weight`` is a separate vector aligned
        with the rows of ``X`` and ``y``; it can also weight regression fits.
        A tuner keeps class weights in ``base_model.params`` (or searches them
        in ``search_space``) and slices row weights for each training fold.

        The return type is intentionally `Any` rather than a narrower
        TypeVar/Protocol: most calculators (see `sklearn_wrapper.py`) return a
        single fitted estimator, but `TuningCalculator`
        (`_tuning/engine.py::fit`) returns a `(model, tuning_result)` tuple
        instead — the artifact shape is model-family-dependent, not just
        heterogeneous across libraries (sklearn estimator, xgboost booster,
        custom wrapper) but also heterogeneous *within* a single calculator
        depending on whether tuning was applied. Consumers already
        `isinstance(self.model, tuple)`-narrow where needed (see
        `StatefulEstimator.evaluate`); a forced union type here wouldn't
        remove that narrowing, so `Any` is the honest, pragmatic choice.
        """

default_params property

Default hyperparameters for the model.

problem_type abstractmethod property

Returns 'classification', 'regression', or 'clustering'.

build_tuning_search_space(config, strategy)

Hook: let a model auto-build its tuning search space.

Returns an empty dict for plain models (the caller keeps the user-provided space). Ensembles override this to expand their base learners' parameter grids into nested <name>__<param> keys.

Source code in skyulf-core/skyulf/modeling/base.py

def build_tuning_search_space(self, config: dict[str, Any], strategy: str) -> dict[str, Any]:
    """Hook: let a model auto-build its tuning search space.

    Returns an empty dict for plain models (the caller keeps the
    user-provided space). Ensembles override this to expand their base
    learners' parameter grids into nested ``<name>__<param>`` keys.
    """
    return {}

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

Trains the model and returns the fitted model artifact.

class_weight is a model setting in config["params"] for supported classifiers. sample_weight is a separate vector aligned with the rows of X and y; it can also weight regression fits. A tuner keeps class weights in base_model.params (or searches them in search_space) and slices row weights for each training fold.

The return type is intentionally Any rather than a narrower TypeVar/Protocol: most calculators (see sklearn_wrapper.py) return a single fitted estimator, but TuningCalculator (_tuning/engine.py::fit) returns a (model, tuning_result) tuple instead — the artifact shape is model-family-dependent, not just heterogeneous across libraries (sklearn estimator, xgboost booster, custom wrapper) but also heterogeneous within a single calculator depending on whether tuning was applied. Consumers already isinstance(self.model, tuple)-narrow where needed (see StatefulEstimator.evaluate); a forced union type here wouldn't remove that narrowing, so Any is the honest, pragmatic choice.

Source code in skyulf-core/skyulf/modeling/base.py

@abstractmethod
def fit(
    self,
    X: pd.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    config: dict[str, Any],
    progress_callback: Callable[..., 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,
    *,
    sample_weight: Any = None,
) -> Any:
    """Trains the model and returns the fitted model artifact.

    ``class_weight`` is a model setting in ``config["params"]`` for
    supported classifiers. ``sample_weight`` is a separate vector aligned
    with the rows of ``X`` and ``y``; it can also weight regression fits.
    A tuner keeps class weights in ``base_model.params`` (or searches them
    in ``search_space``) and slices row weights for each training fold.

    The return type is intentionally `Any` rather than a narrower
    TypeVar/Protocol: most calculators (see `sklearn_wrapper.py`) return a
    single fitted estimator, but `TuningCalculator`
    (`_tuning/engine.py::fit`) returns a `(model, tuning_result)` tuple
    instead — the artifact shape is model-family-dependent, not just
    heterogeneous across libraries (sklearn estimator, xgboost booster,
    custom wrapper) but also heterogeneous *within* a single calculator
    depending on whether tuning was applied. Consumers already
    `isinstance(self.model, tuple)`-narrow where needed (see
    `StatefulEstimator.evaluate`); a forced union type here wouldn't
    remove that narrowing, so `Any` is the honest, pragmatic choice.
    """

prepare_tuning_params(config)

Hook for structural models (e.g. ensembles) to absorb their sub-estimator selection.

Runs before the tuner builds the base model. No-op for plain models. Ensembles override this to inject the resolved estimators (and final_estimator) into :attr:default_params so the tuner can construct a valid meta-estimator.

Source code in skyulf-core/skyulf/modeling/base.py

def prepare_tuning_params(self, config: dict[str, Any]) -> None:
    """Hook for structural models (e.g. ensembles) to absorb their sub-estimator selection.

    Runs before the tuner builds the base model. No-op for plain models.
    Ensembles override this to inject the resolved
    ``estimators`` (and ``final_estimator``) into :attr:`default_params` so
    the tuner can construct a valid meta-estimator.
    """
    return None

BernoulliNBApplier

Bases: SklearnApplier

Bernoulli Naive Bayes Applier.

Source code in skyulf-core/skyulf/modeling/naive_bayes.py

class BernoulliNBApplier(SklearnApplier):
    """Bernoulli Naive Bayes Applier."""

BernoulliNBCalculator

Bases: SklearnCalculator

Bernoulli Naive Bayes Calculator.

Source code in skyulf-core/skyulf/modeling/naive_bayes.py

@NodeRegistry.register("bernoulli_nb", BernoulliNBApplier)
@node_meta(
    id="bernoulli_nb",
    name="Bernoulli Naive Bayes (binary / text)",
    category="Modeling",
    description=(
        "Naive Bayes classifier designed for binary/boolean features. "
        "Each feature is treated as a binary indicator of a token's presence. "
        "Also works with continuous features via a binarization threshold."
    ),
    params={"alpha": 1.0, "binarize": 0.0, "fit_prior": True},
    tags=["text", "nlp", "naive_bayes"],
    learns_from_data=True,
)
class BernoulliNBCalculator(SklearnCalculator):
    """Bernoulli Naive Bayes Calculator."""

    def __init__(self):
        """Pin the estimator to ``BernoulliNB``, adding ``binarize=0.0`` to the NB defaults."""
        super().__init__(
            model_class=BernoulliNB,
            default_params={"alpha": 1.0, "binarize": 0.0, "fit_prior": True},
            problem_type="classification",
        )

    @property
    def problem_type(self) -> str:
        """Return a constant ``"classification"``, so the base's stored value goes unused."""
        return "classification"

problem_type property

Return a constant "classification", so the base's stored value goes unused.

__init__()

Pin the estimator to BernoulliNB, adding binarize=0.0 to the NB defaults.

Source code in skyulf-core/skyulf/modeling/naive_bayes.py

def __init__(self):
    """Pin the estimator to ``BernoulliNB``, adding ``binarize=0.0`` to the NB defaults."""
    super().__init__(
        model_class=BernoulliNB,
        default_params={"alpha": 1.0, "binarize": 0.0, "fit_prior": True},
        problem_type="classification",
    )

CalibratedClassifierApplier

Bases: SklearnApplier

Calibrated Classifier Applier (well-calibrated predict_proba).

Source code in skyulf-core/skyulf/modeling/classification.py

class CalibratedClassifierApplier(SklearnApplier):
    """Calibrated Classifier Applier (well-calibrated predict_proba)."""

CalibratedClassifierCalculator

Bases: SklearnCalculator

Calibrated Classifier Calculator with a selectable base estimator.

The frontend sends base_estimator as a string key (e.g. "random_forest"); it is resolved here into a fresh estimator instance before CalibratedClassifierCV is constructed. Defaults to logistic regression for backward compatibility. random_state seeds supported base estimators, defaults to 42, and accepts None for unseeded fits. It does not shuffle calibration CV folds or affect deterministic bases. A tuning search space can select one or more base_estimator keys; each candidate and the final refit use the selected classifier family.

Source code in skyulf-core/skyulf/modeling/classification.py

@NodeRegistry.register("calibrated_classifier", CalibratedClassifierApplier)
@node_meta(
    id="calibrated_classifier",
    name="Calibrated Classifier",
    category="Modeling",
    description=(
        "Wraps a base classifier with CalibratedClassifierCV so predicted "
        "probabilities are well-calibrated (Platt/sigmoid or isotonic)."
    ),
    params={"base_estimator": "logistic_regression", "method": "sigmoid", "cv": 5},
    tags=["requires_scaling", "classification"],
    learns_from_data=True,
)
class CalibratedClassifierCalculator(SklearnCalculator):
    """Calibrated Classifier Calculator with a selectable base estimator.

    The frontend sends ``base_estimator`` as a string key (e.g.
    ``"random_forest"``); it is resolved here into a fresh estimator instance
    before ``CalibratedClassifierCV`` is constructed. Defaults to logistic
    regression for backward compatibility. ``random_state`` seeds supported
    base estimators, defaults to 42, and accepts ``None`` for unseeded fits.
    It does not shuffle calibration CV folds or affect deterministic bases.
    A tuning search space can select one or more ``base_estimator`` keys;
    each candidate and the final refit use the selected classifier family.
    """

    # Map of selectable base estimators → factory. Each must support
    # ``predict_proba`` (or ``decision_function``) so calibration is meaningful.
    BASE_ESTIMATORS: ClassVar[dict[str, Callable[[], BaseEstimator]]] = _CALIBRATED_BASE_ESTIMATORS

    STRUCTURAL_TUNING_KEYS: tuple[str, ...] = ("base_estimator",)

    def __init__(self):
        """Bind ``CalibratedClassifierCV`` over a logistic-regression base estimator.

        Fixes ``method="sigmoid"`` and ``cv=5`` alongside that base estimator as
        overridable defaults, and starts ``_tuning_base_config`` empty so
        :attr:`default_params` yields the bound estimator until
        :meth:`prepare_tuning_params` records a different selection.
        """
        super().__init__(
            model_class=_SeededCalibratedClassifierCV,
            default_params={
                "estimator": LogisticRegression(max_iter=1000),
                "method": "sigmoid",
                "cv": 5,
            },
            problem_type="classification",
        )
        self._tuning_base_config: dict[str, Any] = {}

    @property
    def default_params(self) -> dict[str, Any]:
        """Return the defaults with any tuning-selected base estimator resolved in.

        An unrecognized ``base_estimator`` key warns and falls back to logistic
        regression rather than raising. The factory runs on every access, so
        separate tuning trials never share one estimator instance.
        """
        params = dict(self._default_params)
        if self._tuning_base_config:
            key = self._tuning_base_config.get("base_estimator")
            if isinstance(key, str):
                params["estimator"] = _make_calibrated_base_estimator(key)
        return params

    def prepare_tuning_params(self, config: dict[str, Any]) -> None:
        """Absorb ``base_estimator`` so it selects an estimator instead of being tuned.

        Only keys named in :attr:`STRUCTURAL_TUNING_KEYS` are taken, read from
        either the nested ``params`` dict or the flat config.
        """
        src = config.get("params") if isinstance(config.get("params"), dict) else config
        src = src or {}
        self._tuning_base_config = {k: src[k] for k in self.STRUCTURAL_TUNING_KEYS if k in src}

    def fit(
        self,
        X: Any,
        y: Any,
        config: dict[str, Any],
        progress_callback: Callable[..., Any] | None = None,
        log_callback: Callable[..., Any] | None = None,
        validation_data: Any = None,
        iteration_callback: Callable[..., Any] | None = None,
        *,
        sample_weight: Any = None,
    ) -> Any:
        """Resolve the base estimator in ``config``, then delegate to the base fit."""
        config = self._resolve_base_estimator(config)
        return super().fit(
            X,
            y,
            config,
            progress_callback,
            log_callback,
            validation_data,
            iteration_callback=iteration_callback,
            sample_weight=sample_weight,
        )

    @classmethod
    def _resolve_base_estimator(cls, config: dict[str, Any] | None) -> dict[str, Any]:
        """Translate a ``base_estimator`` string key into an estimator instance.

        Supports both the flat config shape and the nested ``{"params": {...}}``
        shape used by the model-training payload. Unknown keys fall back to
        logistic regression with a warning.
        """
        if not config:
            return config or {}
        resolved = dict(config)
        nested = isinstance(resolved.get("params"), dict)
        bucket = dict(resolved["params"]) if nested else resolved
        key = bucket.pop("base_estimator", None)
        if isinstance(key, str):
            bucket["estimator"] = _make_calibrated_base_estimator(key)
        if nested:
            resolved["params"] = bucket
            return resolved
        return bucket

default_params property

Return the defaults with any tuning-selected base estimator resolved in.

An unrecognized base_estimator key warns and falls back to logistic regression rather than raising. The factory runs on every access, so separate tuning trials never share one estimator instance.

__init__()

Bind CalibratedClassifierCV over a logistic-regression base estimator.

Fixes method="sigmoid" and cv=5 alongside that base estimator as overridable defaults, and starts _tuning_base_config empty so :attr:default_params yields the bound estimator until :meth:prepare_tuning_params records a different selection.

Source code in skyulf-core/skyulf/modeling/classification.py

def __init__(self):
    """Bind ``CalibratedClassifierCV`` over a logistic-regression base estimator.

    Fixes ``method="sigmoid"`` and ``cv=5`` alongside that base estimator as
    overridable defaults, and starts ``_tuning_base_config`` empty so
    :attr:`default_params` yields the bound estimator until
    :meth:`prepare_tuning_params` records a different selection.
    """
    super().__init__(
        model_class=_SeededCalibratedClassifierCV,
        default_params={
            "estimator": LogisticRegression(max_iter=1000),
            "method": "sigmoid",
            "cv": 5,
        },
        problem_type="classification",
    )
    self._tuning_base_config: dict[str, Any] = {}

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

Resolve the base estimator in config, then delegate to the base fit.

Source code in skyulf-core/skyulf/modeling/classification.py

def fit(
    self,
    X: Any,
    y: Any,
    config: dict[str, Any],
    progress_callback: Callable[..., Any] | None = None,
    log_callback: Callable[..., Any] | None = None,
    validation_data: Any = None,
    iteration_callback: Callable[..., Any] | None = None,
    *,
    sample_weight: Any = None,
) -> Any:
    """Resolve the base estimator in ``config``, then delegate to the base fit."""
    config = self._resolve_base_estimator(config)
    return super().fit(
        X,
        y,
        config,
        progress_callback,
        log_callback,
        validation_data,
        iteration_callback=iteration_callback,
        sample_weight=sample_weight,
    )

prepare_tuning_params(config)

Absorb base_estimator so it selects an estimator instead of being tuned.

Only keys named in :attr:STRUCTURAL_TUNING_KEYS are taken, read from either the nested params dict or the flat config.

Source code in skyulf-core/skyulf/modeling/classification.py

def prepare_tuning_params(self, config: dict[str, Any]) -> None:
    """Absorb ``base_estimator`` so it selects an estimator instead of being tuned.

    Only keys named in :attr:`STRUCTURAL_TUNING_KEYS` are taken, read from
    either the nested ``params`` dict or the flat config.
    """
    src = config.get("params") if isinstance(config.get("params"), dict) else config
    src = src or {}
    self._tuning_base_config = {k: src[k] for k in self.STRUCTURAL_TUNING_KEYS if k in src}

FoldPreprocessor

Bases: Protocol

Structural contract for per-fold preprocessing.

Implementations must be re-fittable: fit_transform may be called repeatedly (once per fold), and each call must discard any state from the previous fold. transform applies the artifacts learned by the most recent fit_transform to held-out rows without learning from them. y is passed alongside X because target-aware steps (target encoders, label encoders fitted on the target, resampling) need it at fit time and may transform it at apply time.

Source code in skyulf-core/skyulf/modeling/fold_preprocessing.py

@runtime_checkable
class FoldPreprocessor(Protocol):
    """Structural contract for per-fold preprocessing.

    Implementations must be re-fittable: ``fit_transform`` may be called
    repeatedly (once per fold), and each call must discard any state from
    the previous fold. ``transform`` applies the artifacts learned by the
    most recent ``fit_transform`` to held-out rows without learning from
    them. ``y`` is passed alongside ``X`` because target-aware steps
    (target encoders, label encoders fitted on the target, resampling)
    need it at fit time and may transform it at apply time.
    """

    def fit_transform(self, X: Any, y: Any) -> tuple[Any, Any]:
        """Fit on this fold's training rows only; return transformed (X, y)."""
        ...

    def transform(self, X: Any, y: Any) -> tuple[Any, Any]:
        """Apply the fitted artifacts to held-out rows without refitting."""
        ...

fit_transform(X, y)

Fit on this fold's training rows only; return transformed (X, y).

Source code in skyulf-core/skyulf/modeling/fold_preprocessing.py

def fit_transform(self, X: Any, y: Any) -> tuple[Any, Any]:
    """Fit on this fold's training rows only; return transformed (X, y)."""
    ...

transform(X, y)

Apply the fitted artifacts to held-out rows without refitting.

Source code in skyulf-core/skyulf/modeling/fold_preprocessing.py

def transform(self, X: Any, y: Any) -> tuple[Any, Any]:
    """Apply the fitted artifacts to held-out rows without refitting."""
    ...

HyperparameterField dataclass

Describe a single tunable hyperparameter.

Source code in skyulf-core/skyulf/modeling/hyperparameters/_field.py

@dataclass
class HyperparameterField:
    """Describe a single tunable hyperparameter."""

    name: str
    label: str
    type: str  # "number", "select", "boolean"
    default: Any
    description: str = ""
    min: float | None = None
    max: float | None = None
    step: float | None = None
    options: list[dict[str, Any]] | None = (
        None  # For 'select' type: [{"label": "L1", "value": "l1"}]
    )
    depends_on: dict[str, Any] | None = (
        None  # Only relevant/shown when another param equals a given value,
        # e.g. {"param": "penalty", "value": "elasticnet"} for `l1_ratio`.
    )
    exclusive_options: list[Any] | None = (
        None  # For multi-select search-space tuning: values here can't be
        # combined with any other option in the same search space (e.g.
        # `penalty="elasticnet"` mixed with "l1"/"l2" produces invalid
        # per-trial combos elsewhere, so selecting one deselects the rest).
    )
    tunable: bool = (
        True  # False = fixed-parameter control only: shown in basic-mode
        # hyperparameters, hidden from the advanced search space. A seed is
        # never a sensible tuning target, so `random_state` fields set False.
    )

    def to_dict(self) -> dict[str, Any]:
        """Plain-dict view of the field for JSON consumers (backend API, frontend).

        ``dataclasses.asdict`` recurses, so nested ``options``/``depends_on``
        structures arrive as plain data too — the registry serves these dicts
        onward and clients never touch the dataclass itself.
        """
        return asdict(self)

to_dict()

Plain-dict view of the field for JSON consumers (backend API, frontend).

dataclasses.asdict recurses, so nested options/depends_on structures arrive as plain data too — the registry serves these dicts onward and clients never touch the dataclass itself.

Source code in skyulf-core/skyulf/modeling/hyperparameters/_field.py

def to_dict(self) -> dict[str, Any]:
    """Plain-dict view of the field for JSON consumers (backend API, frontend).

    ``dataclasses.asdict`` recurses, so nested ``options``/``depends_on``
    structures arrive as plain data too — the registry serves these dicts
    onward and clients never touch the dataclass itself.
    """
    return asdict(self)

KMeansApplier

Bases: _NumericOnlyClusteringApplier

K-Means Applier.

KMeans.predict() genuinely supports out-of-sample cluster assignment (unlike DBSCAN/Agglomerative, which only implement fit_predict() on the training data) — this is exactly why K-Means is deployable for inference.

Source code in skyulf-core/skyulf/modeling/clustering.py

class KMeansApplier(_NumericOnlyClusteringApplier):
    """K-Means Applier.

    `KMeans.predict()` genuinely supports out-of-sample cluster assignment
    (unlike DBSCAN/Agglomerative, which only implement `fit_predict()` on the
    training data) — this is exactly why K-Means is deployable for inference.
    """

KMeansCalculator

Bases: _NumericOnlyClusteringCalculatorMixin, SklearnCalculator

K-Means Calculator.

Source code in skyulf-core/skyulf/modeling/clustering.py

@NodeRegistry.register("kmeans", KMeansApplier)
@node_meta(
    id="kmeans",
    name="K-Means",
    category="Modeling",
    description="Partition rows into a fixed number of clusters (segments) by similarity.",
    params={"n_clusters": 3, "n_init": 10},
    tags=["clustering", "requires_scaling"],
    learns_from_data=True,
)
class KMeansCalculator(_NumericOnlyClusteringCalculatorMixin, SklearnCalculator):
    """K-Means Calculator."""

    def __init__(self):
        """Pin the estimator to ``KMeans`` and default ``n_clusters=3``, ``n_init=10``."""
        super().__init__(
            model_class=KMeans,
            default_params={
                "n_clusters": 3,
                "n_init": 10,
            },
            problem_type="clustering",
        )

__init__()

Pin the estimator to KMeans and default n_clusters=3, n_init=10.

Source code in skyulf-core/skyulf/modeling/clustering.py

def __init__(self):
    """Pin the estimator to ``KMeans`` and default ``n_clusters=3``, ``n_init=10``."""
    super().__init__(
        model_class=KMeans,
        default_params={
            "n_clusters": 3,
            "n_init": 10,
        },
        problem_type="clustering",
    )

LogisticRegressionApplier

Bases: SklearnApplier

Logistic Regression Applier.

Source code in skyulf-core/skyulf/modeling/classification.py

class LogisticRegressionApplier(SklearnApplier):
    """Logistic Regression Applier."""

LogisticRegressionCalculator

Bases: SklearnCalculator

Logistic Regression Calculator.

Source code in skyulf-core/skyulf/modeling/classification.py

@NodeRegistry.register("logistic_regression", LogisticRegressionApplier)
@node_meta(
    id="logistic_regression",
    name="Logistic Regression",
    category="Modeling",
    description="Linear model for classification.",
    params={"max_iter": 1000, "solver": "lbfgs"},
    # "text": logistic regression on TF-IDF/vectorized features is a common,
    # well-performing baseline for text classification (alongside Naive Bayes
    # and the SGD-based linear SVM approximation below).
    tags=["requires_scaling", "classification", "text", "nlp"],
    learns_from_data=True,
)
class LogisticRegressionCalculator(SklearnCalculator):
    """Logistic Regression Calculator."""

    # sklearn solver -> penalties it actually supports. Manual/UI configuration
    # allows selecting solver and penalty independently (unlike the tuner's own
    # search space, which restricts solver to "saga" whenever penalty is
    # varied), so an incompatible combination reaches `fit()` unchecked and
    # would otherwise surface as an opaque sklearn ValueError at model-fit time.
    _SOLVER_PENALTIES: ClassVar[dict[str, set[Any]]] = {
        "lbfgs": {"l2", None},
        "liblinear": {"l1", "l2"},
        "newton-cg": {"l2", None},
        "newton-cholesky": {"l2", None},
        "sag": {"l2", None},
        "saga": {"l1", "l2", "elasticnet", None},
    }

    def __init__(self):
        """Bind ``LogisticRegression`` with ``max_iter=1000`` and ``solver="lbfgs"``.

        Both are defaults, not fixtures: ``fit``'s config may override either,
        and the resulting pair is then checked against ``_SOLVER_PENALTIES``.
        """
        super().__init__(
            model_class=LogisticRegression,
            default_params={
                "max_iter": 1000,
                "solver": "lbfgs",
            },
            problem_type="classification",
        )

    def fit(
        self,
        X: Any,
        y: Any,
        config: dict[str, Any],
        progress_callback: Callable[..., Any] | None = None,
        log_callback: Callable[..., Any] | None = None,
        validation_data: Any = None,
        iteration_callback: Callable[..., Any] | None = None,
        *,
        sample_weight: Any = None,
    ) -> Any:
        """Reject an unsupported solver/penalty pair, then delegate to the base fit."""
        self._validate_solver_penalty(config)
        return super().fit(
            X,
            y,
            config,
            progress_callback,
            log_callback,
            validation_data,
            iteration_callback=iteration_callback,
            sample_weight=sample_weight,
        )

    def _resolve_fit_params(self, config: dict[str, Any]) -> dict[str, Any]:
        """Merges fit params, then normalizes ``penalty`` for sklearn >=1.8.

        sklearn >=1.8 deprecates the ``penalty`` constructor arg entirely (in
        favor of ``l1_ratio``/``C``) and will remove it in sklearn 1.10. We
        keep ``penalty`` ("l1"/"l2"/"elasticnet"/None) as our own public
        config/UI field unchanged. Translate it for the newer API while
        retaining the native penalty argument on older sklearn versions.
        """
        params = super()._resolve_fit_params(config)
        return normalize_logistic_regression_params(params)

    @classmethod
    def _extract_solver_penalty_params(cls, config: dict[str, Any] | None) -> dict[str, Any] | None:
        """Returns the params dict from config, or None if unavailable/not a dict."""
        if not config:
            return None
        params = config.get("params", config)
        if not isinstance(params, dict):
            return None
        return params

    @classmethod
    def _raise_incompatible_solver_penalty(cls, solver: Any, penalty: Any) -> None:
        """Raises a ValueError listing solvers compatible with the requested penalty."""
        compatible_solvers = sorted(
            s for s, penalties in cls._SOLVER_PENALTIES.items() if penalty in penalties
        )
        raise ValueError(
            f"Logistic Regression: solver={solver!r} does not support "
            f"penalty={penalty!r}. Solvers compatible with this penalty: "
            f"{compatible_solvers or 'none'}."
        )

    def _validate_solver_penalty(self, config: dict[str, Any] | None) -> None:
        """Raise a clear, actionable error for an invalid solver/penalty pair.

        sklearn's own error for this (e.g. "Solver lbfgs supports only 'l2' or
        None penalties") is only raised deep inside `LogisticRegression.fit`,
        after data has already been split/validated upstream. Failing fast
        here with the full list of compatible solvers is more actionable.

        Validates against the *merged* effective params (``default_params``
        overlaid with the config's overrides), not just the raw config —
        otherwise overriding only ``penalty`` (very common; ``solver``
        defaults to ``"lbfgs"``) would skip validation entirely since
        ``solver`` never appears in the raw override dict, letting an
        incompatible combo reach sklearn's own opaque error at fit time.
        """
        overrides = self._extract_solver_penalty_params(config) or {}
        params = {**self.default_params, **overrides}
        solver = params.get("solver")
        if solver is None or "penalty" not in params:
            return
        penalty = params.get("penalty")
        compatible = self._SOLVER_PENALTIES.get(solver)
        if compatible is not None and penalty not in compatible:
            self._raise_incompatible_solver_penalty(solver, penalty)

__init__()

Bind LogisticRegression with max_iter=1000 and solver="lbfgs".

Both are defaults, not fixtures: fit's config may override either, and the resulting pair is then checked against _SOLVER_PENALTIES.

Source code in skyulf-core/skyulf/modeling/classification.py

def __init__(self):
    """Bind ``LogisticRegression`` with ``max_iter=1000`` and ``solver="lbfgs"``.

    Both are defaults, not fixtures: ``fit``'s config may override either,
    and the resulting pair is then checked against ``_SOLVER_PENALTIES``.
    """
    super().__init__(
        model_class=LogisticRegression,
        default_params={
            "max_iter": 1000,
            "solver": "lbfgs",
        },
        problem_type="classification",
    )

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

Reject an unsupported solver/penalty pair, then delegate to the base fit.

Source code in skyulf-core/skyulf/modeling/classification.py

def fit(
    self,
    X: Any,
    y: Any,
    config: dict[str, Any],
    progress_callback: Callable[..., Any] | None = None,
    log_callback: Callable[..., Any] | None = None,
    validation_data: Any = None,
    iteration_callback: Callable[..., Any] | None = None,
    *,
    sample_weight: Any = None,
) -> Any:
    """Reject an unsupported solver/penalty pair, then delegate to the base fit."""
    self._validate_solver_penalty(config)
    return super().fit(
        X,
        y,
        config,
        progress_callback,
        log_callback,
        validation_data,
        iteration_callback=iteration_callback,
        sample_weight=sample_weight,
    )

MultinomialNBApplier

Bases: SklearnApplier

Multinomial Naive Bayes Applier.

Source code in skyulf-core/skyulf/modeling/naive_bayes.py

class MultinomialNBApplier(SklearnApplier):
    """Multinomial Naive Bayes Applier."""

MultinomialNBCalculator

Bases: SklearnCalculator

Multinomial Naive Bayes Calculator.

Source code in skyulf-core/skyulf/modeling/naive_bayes.py

@NodeRegistry.register("multinomial_nb", MultinomialNBApplier)
@node_meta(
    id="multinomial_nb",
    name="Multinomial Naive Bayes (counts / text)",
    category="Modeling",
    description=(
        "Naive Bayes classifier for multinomially-distributed features "
        "(e.g. token counts or TF-IDF). "
        "Requires non-negative input features."
    ),
    params={"alpha": 1.0, "fit_prior": True},
    tags=["text", "nlp", "naive_bayes"],
    learns_from_data=True,
)
class MultinomialNBCalculator(SklearnCalculator):
    """Multinomial Naive Bayes Calculator."""

    def __init__(self):
        """Pin the estimator to ``MultinomialNB`` with ``alpha=1.0`` and ``fit_prior=True``."""
        super().__init__(
            model_class=MultinomialNB,
            default_params={"alpha": 1.0, "fit_prior": True},
            problem_type="classification",
        )

    @property
    def problem_type(self) -> str:
        """Return a constant ``"classification"``, so the base's stored value goes unused."""
        return "classification"

problem_type property

Return a constant "classification", so the base's stored value goes unused.

__init__()

Pin the estimator to MultinomialNB with alpha=1.0 and fit_prior=True.

Source code in skyulf-core/skyulf/modeling/naive_bayes.py

def __init__(self):
    """Pin the estimator to ``MultinomialNB`` with ``alpha=1.0`` and ``fit_prior=True``."""
    super().__init__(
        model_class=MultinomialNB,
        default_params={"alpha": 1.0, "fit_prior": True},
        problem_type="classification",
    )

RandomForestClassifierApplier

Bases: SklearnApplier

Random Forest Classifier Applier.

Source code in skyulf-core/skyulf/modeling/classification.py

class RandomForestClassifierApplier(SklearnApplier):
    """Random Forest Classifier Applier."""

RandomForestClassifierCalculator

Bases: SklearnCalculator

Random Forest Classifier Calculator.

Source code in skyulf-core/skyulf/modeling/classification.py

@NodeRegistry.register("random_forest_classifier", RandomForestClassifierApplier)
@node_meta(
    id="random_forest_classifier",
    name="Random Forest Classifier",
    category="Modeling",
    description="Ensemble of decision trees.",
    params={"n_estimators": 50, "max_depth": 10, "min_samples_split": 5},
    tags=["classification"],
    learns_from_data=True,
)
class RandomForestClassifierCalculator(SklearnCalculator):
    """Random Forest Classifier Calculator."""

    def __init__(self):
        """Bind ``RandomForestClassifier`` with 50 shallow, constrained trees.

        ``max_depth=10``, ``min_samples_split=5`` and ``min_samples_leaf=2``
        restrict tree growth that sklearn leaves open by default, and the forest
        is half sklearn's default size. ``n_jobs=-1`` fits across every core.
        All are overridable from ``fit``'s config.
        """
        super().__init__(
            model_class=RandomForestClassifier,
            default_params={
                "n_estimators": 50,
                "max_depth": 10,
                "min_samples_split": 5,
                "min_samples_leaf": 2,
                "n_jobs": -1,
            },
            problem_type="classification",
        )

__init__()

Bind RandomForestClassifier with 50 shallow, constrained trees.

max_depth=10, min_samples_split=5 and min_samples_leaf=2 restrict tree growth that sklearn leaves open by default, and the forest is half sklearn's default size. n_jobs=-1 fits across every core. All are overridable from fit's config.

Source code in skyulf-core/skyulf/modeling/classification.py

def __init__(self):
    """Bind ``RandomForestClassifier`` with 50 shallow, constrained trees.

    ``max_depth=10``, ``min_samples_split=5`` and ``min_samples_leaf=2``
    restrict tree growth that sklearn leaves open by default, and the forest
    is half sklearn's default size. ``n_jobs=-1`` fits across every core.
    All are overridable from ``fit``'s config.
    """
    super().__init__(
        model_class=RandomForestClassifier,
        default_params={
            "n_estimators": 50,
            "max_depth": 10,
            "min_samples_split": 5,
            "min_samples_leaf": 2,
            "n_jobs": -1,
        },
        problem_type="classification",
    )

RandomForestRegressorApplier

Bases: SklearnApplier

Random Forest Regressor Applier.

Source code in skyulf-core/skyulf/modeling/regression.py

class RandomForestRegressorApplier(SklearnApplier):
    """Random Forest Regressor Applier."""

RandomForestRegressorCalculator

Bases: SklearnCalculator

Random Forest Regressor Calculator.

Source code in skyulf-core/skyulf/modeling/regression.py

@NodeRegistry.register("random_forest_regressor", RandomForestRegressorApplier)
@node_meta(
    id="random_forest_regressor",
    name="Random Forest Regressor",
    category="Modeling",
    description="Ensemble of decision trees for regression.",
    params={"n_estimators": 50, "max_depth": 10, "min_samples_split": 5},
    tags=["regression"],
    learns_from_data=True,
)
class RandomForestRegressorCalculator(SklearnCalculator):
    """Random Forest Regressor Calculator."""

    def __init__(self):
        """Bind ``RandomForestRegressor`` with 50 shallow, constrained trees.

        ``max_depth=10``, ``min_samples_split=5`` and ``min_samples_leaf=2``
        restrict tree growth that sklearn leaves open by default, and the
        forest is half sklearn's default size. ``n_jobs=-1`` fits across every
        core. All are overridable from ``fit``'s config.
        """
        super().__init__(
            model_class=RandomForestRegressor,
            default_params={
                "n_estimators": 50,
                "max_depth": 10,
                "min_samples_split": 5,
                "min_samples_leaf": 2,
                "n_jobs": -1,
            },
            problem_type="regression",
        )

__init__()

Bind RandomForestRegressor with 50 shallow, constrained trees.

max_depth=10, min_samples_split=5 and min_samples_leaf=2 restrict tree growth that sklearn leaves open by default, and the forest is half sklearn's default size. n_jobs=-1 fits across every core. All are overridable from fit's config.

Source code in skyulf-core/skyulf/modeling/regression.py

def __init__(self):
    """Bind ``RandomForestRegressor`` with 50 shallow, constrained trees.

    ``max_depth=10``, ``min_samples_split=5`` and ``min_samples_leaf=2``
    restrict tree growth that sklearn leaves open by default, and the
    forest is half sklearn's default size. ``n_jobs=-1`` fits across every
    core. All are overridable from ``fit``'s config.
    """
    super().__init__(
        model_class=RandomForestRegressor,
        default_params={
            "n_estimators": 50,
            "max_depth": 10,
            "min_samples_split": 5,
            "min_samples_leaf": 2,
            "n_jobs": -1,
        },
        problem_type="regression",
    )

RidgeRegressionApplier

Bases: SklearnApplier

Ridge Regression Applier.

Source code in skyulf-core/skyulf/modeling/regression.py

class RidgeRegressionApplier(SklearnApplier):
    """Ridge Regression Applier."""

RidgeRegressionCalculator

Bases: SklearnCalculator

Ridge Regression Calculator.

Source code in skyulf-core/skyulf/modeling/regression.py

@NodeRegistry.register("ridge_regression", RidgeRegressionApplier)
@node_meta(
    id="ridge_regression",
    name="Ridge Regression",
    category="Modeling",
    description="Linear least squares with l2 regularization.",
    params={"alpha": 1.0, "solver": "auto"},
    tags=["requires_scaling", "regression"],
    learns_from_data=True,
)
class RidgeRegressionCalculator(SklearnCalculator):
    """Ridge Regression Calculator."""

    def __init__(self):
        """Bind ``Ridge`` with sklearn's own defaults.

        ``alpha=1.0`` is the L2 penalty strength and ``solver="auto"`` lets
        sklearn pick the solver from the data's shape and dtype. Both are
        restated so the node's tunable surface is explicit; ``fit``'s config
        overrides either.
        """
        super().__init__(
            model_class=Ridge,
            default_params={
                "alpha": 1.0,
                "solver": "auto",
            },
            problem_type="regression",
        )

__init__()

Bind Ridge with sklearn's own defaults.

alpha=1.0 is the L2 penalty strength and solver="auto" lets sklearn pick the solver from the data's shape and dtype. Both are restated so the node's tunable surface is explicit; fit's config overrides either.

Source code in skyulf-core/skyulf/modeling/regression.py

def __init__(self):
    """Bind ``Ridge`` with sklearn's own defaults.

    ``alpha=1.0`` is the L2 penalty strength and ``solver="auto"`` lets
    sklearn pick the solver from the data's shape and dtype. Both are
    restated so the node's tunable surface is explicit; ``fit``'s config
    overrides either.
    """
    super().__init__(
        model_class=Ridge,
        default_params={
            "alpha": 1.0,
            "solver": "auto",
        },
        problem_type="regression",
    )

SGDClassifierApplier

Bases: SklearnApplier

Stochastic Gradient Descent Classifier Applier.

Source code in skyulf-core/skyulf/modeling/classification.py

class SGDClassifierApplier(SklearnApplier):
    """Stochastic Gradient Descent Classifier Applier."""

SGDClassifierCalculator

Bases: SklearnCalculator

SGD Classifier Calculator.

Source code in skyulf-core/skyulf/modeling/classification.py

@NodeRegistry.register("sgd_classifier", SGDClassifierApplier)
@node_meta(
    id="sgd_classifier",
    name="SGD Classifier (text / linear)",
    category="Modeling",
    description=(
        "Linear classifiers (SVM, logistic regression, etc.) with SGD training. "
        "Highly efficient for high-dimensional sparse/dense text representations "
        "and large datasets."
    ),
    params={
        "loss": "log_loss",
        "penalty": "l2",
        "alpha": 0.0001,
        "l1_ratio": 0.15,
        "max_iter": 1000,
    },
    # Text-classification-scoped only (no "classification" tag): SGD with
    # hinge/log loss is a fast linear-SVM/logistic-regression approximation
    # that excels on sparse high-dimensional text features (TF-IDF/counts),
    # so it's offered via the Text Classification node rather than the
    # general Classification node, which already has logistic_regression and
    # other dense-feature-friendly linear models covering that role.
    tags=["text", "nlp", "linear", "requires_scaling"],
    learns_from_data=True,
)
class SGDClassifierCalculator(SklearnCalculator):
    """SGD Classifier Calculator."""

    def __init__(self):
        """Bind ``SGDClassifier`` as a logistic-regression approximation.

        ``loss="log_loss"`` replaces sklearn's default ``"hinge"``, and that is
        what makes ``predict_proba`` available to the applier — a hinge loss
        yields a linear SVM with no probability output. ``l1_ratio=0.15`` stays
        inert unless ``penalty`` is switched to ``"elasticnet"``.
        """
        super().__init__(
            model_class=SGDClassifier,
            default_params={
                "loss": "log_loss",
                "penalty": "l2",
                "alpha": 0.0001,
                "l1_ratio": 0.15,
                "max_iter": 1000,
            },
            problem_type="classification",
        )

__init__()

Bind SGDClassifier as a logistic-regression approximation.

loss="log_loss" replaces sklearn's default "hinge", and that is what makes predict_proba available to the applier — a hinge loss yields a linear SVM with no probability output. l1_ratio=0.15 stays inert unless penalty is switched to "elasticnet".

Source code in skyulf-core/skyulf/modeling/classification.py

def __init__(self):
    """Bind ``SGDClassifier`` as a logistic-regression approximation.

    ``loss="log_loss"`` replaces sklearn's default ``"hinge"``, and that is
    what makes ``predict_proba`` available to the applier — a hinge loss
    yields a linear SVM with no probability output. ``l1_ratio=0.15`` stays
    inert unless ``penalty`` is switched to ``"elasticnet"``.
    """
    super().__init__(
        model_class=SGDClassifier,
        default_params={
            "loss": "log_loss",
            "penalty": "l2",
            "alpha": 0.0001,
            "l1_ratio": 0.15,
            "max_iter": 1000,
        },
        problem_type="classification",
    )

SklearnApplier

Bases: BaseModelApplier

Base applier for Scikit-Learn models.

Source code in skyulf-core/skyulf/modeling/sklearn_wrapper.py

class SklearnApplier(BaseModelApplier):
    """Base applier for Scikit-Learn models."""

    def predict(self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any) -> Any:
        """Predict over a frame and return the labels as a pandas ``Series``.

        The input index is preserved for pandas frames; polars and wrapper
        inputs fall back to a default index.
        """
        # Convert to Numpy
        _validate_feature_schema(df, model_artifact)
        X_np, _ = SklearnBridge.to_sklearn(df, validate_features=True)

        preds = model_artifact.predict(X_np)

        # Return as Pandas Series for consistency
        # If input was Pandas, try to preserve index
        index = None
        if hasattr(df, "index"):
            index = cast(pd.DataFrame, df).index
        elif hasattr(df, "to_pandas"):
            # If it's a wrapper or Polars, we might lose index unless we convert
            # For now, default index is acceptable for predictions
            pass

        return pd.Series(preds, index=index)

    def predict_proba(self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any) -> Any | None:
        """Return per-class probabilities as a ``DataFrame``, or ``None`` if unsupported.

        An estimator with no ``predict_proba`` yields ``None`` instead of
        raising; column labels come from ``classes_`` when it is exposed.
        """
        if not hasattr(model_artifact, "predict_proba"):
            return None

        _validate_feature_schema(df, model_artifact)
        X_np, _ = SklearnBridge.to_sklearn(df, validate_features=True)
        probs = model_artifact.predict_proba(X_np)

        # Return as DataFrame
        index = None
        if hasattr(df, "index"):
            index = cast(pd.DataFrame, df).index

        # Column names usually 0, 1, etc. or classes_. Coerce to native
        # Python types (str) so downstream JSON serialization of the
        # resulting DataFrame's columns doesn't choke on numpy scalar
        # types (e.g. np.int64), mirroring the class_names normalization
        # already done in modeling/_evaluation/classification.py.
        columns = None
        if hasattr(model_artifact, "classes_"):
            columns = pd.Index([str(c) for c in model_artifact.classes_])

        return pd.DataFrame(probs, index=index, columns=columns)

predict(df, model_artifact)

Predict over a frame and return the labels as a pandas Series.

The input index is preserved for pandas frames; polars and wrapper inputs fall back to a default index.

Source code in skyulf-core/skyulf/modeling/sklearn_wrapper.py

def predict(self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any) -> Any:
    """Predict over a frame and return the labels as a pandas ``Series``.

    The input index is preserved for pandas frames; polars and wrapper
    inputs fall back to a default index.
    """
    # Convert to Numpy
    _validate_feature_schema(df, model_artifact)
    X_np, _ = SklearnBridge.to_sklearn(df, validate_features=True)

    preds = model_artifact.predict(X_np)

    # Return as Pandas Series for consistency
    # If input was Pandas, try to preserve index
    index = None
    if hasattr(df, "index"):
        index = cast(pd.DataFrame, df).index
    elif hasattr(df, "to_pandas"):
        # If it's a wrapper or Polars, we might lose index unless we convert
        # For now, default index is acceptable for predictions
        pass

    return pd.Series(preds, index=index)

predict_proba(df, model_artifact)

Return per-class probabilities as a DataFrame, or None if unsupported.

An estimator with no predict_proba yields None instead of raising; column labels come from classes_ when it is exposed.

Source code in skyulf-core/skyulf/modeling/sklearn_wrapper.py

def predict_proba(self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any) -> Any | None:
    """Return per-class probabilities as a ``DataFrame``, or ``None`` if unsupported.

    An estimator with no ``predict_proba`` yields ``None`` instead of
    raising; column labels come from ``classes_`` when it is exposed.
    """
    if not hasattr(model_artifact, "predict_proba"):
        return None

    _validate_feature_schema(df, model_artifact)
    X_np, _ = SklearnBridge.to_sklearn(df, validate_features=True)
    probs = model_artifact.predict_proba(X_np)

    # Return as DataFrame
    index = None
    if hasattr(df, "index"):
        index = cast(pd.DataFrame, df).index

    # Column names usually 0, 1, etc. or classes_. Coerce to native
    # Python types (str) so downstream JSON serialization of the
    # resulting DataFrame's columns doesn't choke on numpy scalar
    # types (e.g. np.int64), mirroring the class_names normalization
    # already done in modeling/_evaluation/classification.py.
    columns = None
    if hasattr(model_artifact, "classes_"):
        columns = pd.Index([str(c) for c in model_artifact.classes_])

    return pd.DataFrame(probs, index=index, columns=columns)

SklearnCalculator

Bases: BaseModelCalculator

Base calculator for Scikit-Learn models.

Source code in skyulf-core/skyulf/modeling/sklearn_wrapper.py

class SklearnCalculator(BaseModelCalculator):
    """Base calculator for Scikit-Learn models."""

    def __init__(
        self,
        model_class: type[BaseEstimator],
        default_params: dict[str, Any],
        problem_type: str,
    ):
        """Bind the wrapped estimator class, its default params and its task type.

        Every per-estimator subclass supplies all three; the rest of ``fit``
        stays generic across estimators.
        """
        # `Any` because sklearn stubs make BaseEstimator subclasses appear non-callable.
        self.model_class: Any = model_class
        self._default_params = default_params
        self._problem_type = problem_type

    @property
    def default_params(self) -> dict[str, Any]:
        """Expose the per-estimator defaults that :meth:`fit` merges ``config`` over."""
        return self._default_params

    @property
    def problem_type(self) -> str:
        """Expose the task type fixed at construction; it drives evaluation dispatch."""
        return self._problem_type

    def fit(
        self,
        X: pd.DataFrame | SkyulfDataFrame,
        y: pd.Series | Any,
        config: dict[str, Any],
        progress_callback=None,
        log_callback=None,
        validation_data=None,
        iteration_callback=None,
        *,
        sample_weight: Any = None,
    ) -> Any:
        """Fit the Scikit-Learn model."""
        if sample_weight is not None and self.problem_type not in ("classification", "regression"):
            raise SampleWeightError(
                "sample_weight is supported only for classification and regression in V1."
            )
        # 1. Merge Config with Defaults
        params = self._resolve_fit_params(config)

        # Share normalization and nonnative conversion with tuning so every
        # fit computes class weights from its own training labels.
        params, class_weight_to_apply = split_class_weight_params(self.model_class, params)

        msg = f"Initializing {self.model_class.__name__} with params: {params}"
        logger.info(msg)
        if log_callback:
            log_callback(msg)

        # 2. Instantiate Model
        valid_params = self._filter_supported_params(params)
        model = self.model_class(**valid_params)

        # 3. Fit
        # Convert to Numpy using Bridge (handles Polars/Pandas/Wrappers)
        X_np, y_np = SklearnBridge.to_sklearn((X, y), validate_features=True)

        sample_weight = sample_weight_for_fit(model, class_weight_to_apply, y_np, sample_weight)

        # sklearn's ConvergenceWarning (raised via `warnings.warn`, not the
        # `logging` module) would otherwise only reach the server's stderr
        # and never surface to the user — unlike the skyulf-core node
        # advisories already routed through `WarningCaptureHandler` via
        # `logger.warning(...)`. Capture everything sklearn emits during
        # `fit`, re-route ConvergenceWarning through this model's own
        # (``skyulf.*``-tree) logger so every sklearn-backed model gets the
        # same UI-visible treatment regardless of solver/estimator, and
        # re-emit any other warning category unchanged so existing
        # console/log behavior for those is preserved.
        with warnings.catch_warnings(record=True) as caught:
            warnings.simplefilter("always")
            boosting_kwargs = self._boosting_fit_kwargs(model, X_np, y_np, iteration_callback)
            detach_callbacks = boosting_kwargs.pop("_detach_callbacks", False)
            if sample_weight is not None:
                model.fit(X_np, y_np, sample_weight=sample_weight, **boosting_kwargs)
            else:
                model.fit(X_np, y_np, **boosting_kwargs)
        # Never leave live callback closures on the fitted model — artifacts
        # get pickled for storage/serving.
        if detach_callbacks:
            model.callbacks = None
        if hasattr(X, "columns"):
            model._skyulf_input_schema = SkyulfSchema.from_dataframe(X)
        for w in caught:
            if issubclass(w.category, ConvergenceWarning):
                conv_msg = f"{self.model_class.__name__} did not fully converge: {w.message}"
                logger.warning(conv_msg)
                if log_callback:
                    log_callback(conv_msg)
            else:
                warnings.warn_explicit(w.message, w.category, w.filename, w.lineno)

        return model

    def _resolve_fit_params(self, config: dict[str, Any]) -> dict[str, Any]:
        """Merges ``default_params`` with overrides from ``config``.

        Supports two configuration structures: a nested ``{'params': {...}}`` dict
        (preferred), or a flat legacy dict where non-reserved, non-dict keys are
        treated as params.
        """
        params = self.default_params.copy()
        if not config:
            return self._inject_default_seed(params)

        # We support two configuration structures:
        # 1. Nested: {'params': {'C': 1.0, ...}} - Preferred
        # 2. Flat: {'C': 1.0, 'type': '...', ...} - Legacy/Simple support

        # Check for explicit 'params' dictionary first
        overrides = config.get("params", {})

        # If 'params' key exists but is None or empty, check if there are other keys at top level
        # that might be params. But be careful not to mix them.
        # If config has 'params', we assume it's the source of truth.

        if not overrides and "params" not in config:
            # Fallback to flat config if 'params' key is completely missing
            reserved_keys = {
                "type",
                "target_column",
                "node_id",
                "step_type",
                "inputs",
            }
            overrides = {
                k: v
                for k, v in config.items()
                if k not in reserved_keys and not isinstance(v, dict)
            }

        if overrides:
            params.update(overrides)

        return self._inject_default_seed(params)

    def _inject_default_seed(self, params: dict[str, Any]) -> dict[str, Any]:
        """Single owner for seeding (finding F-21).

        Model defaults no longer carry their own ``random_state`` literals.
        If the caller didn't configure one, inject ``DEFAULT_RANDOM_STATE``
        here — but only when the wrapped estimator's constructor accepts it
        (named parameter or ``**kwargs``), so unsupported estimators don't
        get a dropped-param warning. An explicit user value (including
        ``None`` for "unseeded") always wins.
        """
        if "random_state" in params:
            return params
        sig = inspect.signature(self.model_class)
        accepts = (
            any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values())
            or "random_state" in sig.parameters
        )
        if accepts:
            params["random_state"] = DEFAULT_RANDOM_STATE
        return params

    def _filter_supported_params(self, params: dict[str, Any]) -> dict[str, Any]:
        """Filters ``params`` down to those accepted by the model class constructor.

        Skips filtering when the constructor accepts ``**kwargs`` (e.g. XGBoost 2.x),
        since every named param would otherwise fail the membership check even though valid.
        """
        sig = inspect.signature(self.model_class)
        accepts_kwargs = any(
            p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()
        )

        if accepts_kwargs:
            return params

        valid_params = {k: v for k, v in params.items() if k in sig.parameters}
        dropped = set(params.keys()) - set(valid_params.keys())
        if dropped:
            logger.warning(
                f"Dropped parameters not supported by {self.model_class.__name__}: {dropped}"
            )
        return valid_params

    def _compute_sample_weight_for_fit(self, model: Any, class_weight: Any, y_np: Any) -> Any:
        """Translate a `class_weight` value into a per-sample weight array.

        For models with no native `class_weight` support, raising a clear error
        instead of silently no-op'ing if the model's `.fit()` doesn't accept
        `sample_weight` either.
        """
        return sample_weight_for_fit(model, class_weight, y_np)

default_params property

Expose the per-estimator defaults that :meth:fit merges config over.

problem_type property

Expose the task type fixed at construction; it drives evaluation dispatch.

__init__(model_class, default_params, problem_type)

Bind the wrapped estimator class, its default params and its task type.

Every per-estimator subclass supplies all three; the rest of fit stays generic across estimators.

Source code in skyulf-core/skyulf/modeling/sklearn_wrapper.py

def __init__(
    self,
    model_class: type[BaseEstimator],
    default_params: dict[str, Any],
    problem_type: str,
):
    """Bind the wrapped estimator class, its default params and its task type.

    Every per-estimator subclass supplies all three; the rest of ``fit``
    stays generic across estimators.
    """
    # `Any` because sklearn stubs make BaseEstimator subclasses appear non-callable.
    self.model_class: Any = model_class
    self._default_params = default_params
    self._problem_type = problem_type

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

Fit the Scikit-Learn model.

Source code in skyulf-core/skyulf/modeling/sklearn_wrapper.py

def fit(
    self,
    X: pd.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    config: dict[str, Any],
    progress_callback=None,
    log_callback=None,
    validation_data=None,
    iteration_callback=None,
    *,
    sample_weight: Any = None,
) -> Any:
    """Fit the Scikit-Learn model."""
    if sample_weight is not None and self.problem_type not in ("classification", "regression"):
        raise SampleWeightError(
            "sample_weight is supported only for classification and regression in V1."
        )
    # 1. Merge Config with Defaults
    params = self._resolve_fit_params(config)

    # Share normalization and nonnative conversion with tuning so every
    # fit computes class weights from its own training labels.
    params, class_weight_to_apply = split_class_weight_params(self.model_class, params)

    msg = f"Initializing {self.model_class.__name__} with params: {params}"
    logger.info(msg)
    if log_callback:
        log_callback(msg)

    # 2. Instantiate Model
    valid_params = self._filter_supported_params(params)
    model = self.model_class(**valid_params)

    # 3. Fit
    # Convert to Numpy using Bridge (handles Polars/Pandas/Wrappers)
    X_np, y_np = SklearnBridge.to_sklearn((X, y), validate_features=True)

    sample_weight = sample_weight_for_fit(model, class_weight_to_apply, y_np, sample_weight)

    # sklearn's ConvergenceWarning (raised via `warnings.warn`, not the
    # `logging` module) would otherwise only reach the server's stderr
    # and never surface to the user — unlike the skyulf-core node
    # advisories already routed through `WarningCaptureHandler` via
    # `logger.warning(...)`. Capture everything sklearn emits during
    # `fit`, re-route ConvergenceWarning through this model's own
    # (``skyulf.*``-tree) logger so every sklearn-backed model gets the
    # same UI-visible treatment regardless of solver/estimator, and
    # re-emit any other warning category unchanged so existing
    # console/log behavior for those is preserved.
    with warnings.catch_warnings(record=True) as caught:
        warnings.simplefilter("always")
        boosting_kwargs = self._boosting_fit_kwargs(model, X_np, y_np, iteration_callback)
        detach_callbacks = boosting_kwargs.pop("_detach_callbacks", False)
        if sample_weight is not None:
            model.fit(X_np, y_np, sample_weight=sample_weight, **boosting_kwargs)
        else:
            model.fit(X_np, y_np, **boosting_kwargs)
    # Never leave live callback closures on the fitted model — artifacts
    # get pickled for storage/serving.
    if detach_callbacks:
        model.callbacks = None
    if hasattr(X, "columns"):
        model._skyulf_input_schema = SkyulfSchema.from_dataframe(X)
    for w in caught:
        if issubclass(w.category, ConvergenceWarning):
            conv_msg = f"{self.model_class.__name__} did not fully converge: {w.message}"
            logger.warning(conv_msg)
            if log_callback:
                log_callback(conv_msg)
        else:
            warnings.warn_explicit(w.message, w.category, w.filename, w.lineno)

    return model

StackingClassifierApplier

Bases: SklearnApplier

Stacking Classifier Applier (meta-learner over base classifiers).

Source code in skyulf-core/skyulf/modeling/ensemble.py

class StackingClassifierApplier(SklearnApplier):
    """Stacking Classifier Applier (meta-learner over base classifiers)."""

StackingClassifierCalculator

Bases: _BaseEnsembleCalculator

Stacking Classifier Calculator with selectable base + final learners.

Source code in skyulf-core/skyulf/modeling/ensemble.py

@NodeRegistry.register("stacking_classifier", StackingClassifierApplier)
@node_meta(
    id="stacking_classifier",
    name="Stacking Classifier",
    category="Ensemble",
    description=(
        "Trains a final classifier on the out-of-fold predictions of several "
        "base classifiers. Uses internal CV folds to avoid leakage."
    ),
    params={
        "base_estimators": ["random_forest", "gradient_boosting", "svc"],
        "final_estimator": "logistic_regression",
        "cv": 5,
    },
    tags=["requires_scaling", "classification"],
    learns_from_data=True,
)
class StackingClassifierCalculator(_BaseEnsembleCalculator):
    """Stacking Classifier Calculator with selectable base + final learners."""

    BASE_ESTIMATORS = BASE_ESTIMATORS_CLF
    DEFAULT_KEYS = ("random_forest", "gradient_boosting", "svc")
    DEFAULT_FINAL_KEY = "logistic_regression"
    MODEL_KEY = "stacking_classifier"
    IS_STACKING = True

    def __init__(self):
        """Pin the meta-estimator to ``StackingClassifier`` with ``cv=5``."""
        super().__init__(
            model_class=StackingClassifier,
            default_params={"cv": 5},
            problem_type="classification",
        )

__init__()

Pin the meta-estimator to StackingClassifier with cv=5.

Source code in skyulf-core/skyulf/modeling/ensemble.py

def __init__(self):
    """Pin the meta-estimator to ``StackingClassifier`` with ``cv=5``."""
    super().__init__(
        model_class=StackingClassifier,
        default_params={"cv": 5},
        problem_type="classification",
    )

StackingRegressorApplier

Bases: SklearnApplier

Stacking Regressor Applier (meta-learner over base regressors).

Source code in skyulf-core/skyulf/modeling/ensemble.py

class StackingRegressorApplier(SklearnApplier):
    """Stacking Regressor Applier (meta-learner over base regressors)."""

StackingRegressorCalculator

Bases: _BaseEnsembleCalculator

Stacking Regressor Calculator with selectable base + final learners.

Source code in skyulf-core/skyulf/modeling/ensemble.py

@NodeRegistry.register("stacking_regressor", StackingRegressorApplier)
@node_meta(
    id="stacking_regressor",
    name="Stacking Regressor",
    category="Ensemble",
    description=(
        "Trains a final regressor on the out-of-fold predictions of several "
        "base regressors. Uses internal CV folds to avoid leakage."
    ),
    params={
        "base_estimators": ["random_forest", "gradient_boosting", "ridge"],
        "final_estimator": "ridge",
        "cv": 5,
    },
    tags=["requires_scaling", "regression"],
    learns_from_data=True,
)
class StackingRegressorCalculator(_BaseEnsembleCalculator):
    """Stacking Regressor Calculator with selectable base + final learners."""

    BASE_ESTIMATORS = BASE_ESTIMATORS_REG
    DEFAULT_KEYS = ("random_forest", "gradient_boosting", "ridge")
    DEFAULT_FINAL_KEY = "ridge"
    MODEL_KEY = "stacking_regressor"
    IS_STACKING = True

    def __init__(self):
        """Pin the meta-estimator to ``StackingRegressor`` with ``cv=5``."""
        super().__init__(
            model_class=StackingRegressor,
            default_params={"cv": 5},
            problem_type="regression",
        )

__init__()

Pin the meta-estimator to StackingRegressor with cv=5.

Source code in skyulf-core/skyulf/modeling/ensemble.py

def __init__(self):
    """Pin the meta-estimator to ``StackingRegressor`` with ``cv=5``."""
    super().__init__(
        model_class=StackingRegressor,
        default_params={"cv": 5},
        problem_type="regression",
    )

StatefulEstimator

Drive a model node end to end, holding the fitted model between calls.

Wraps a calculator/applier pair and owns the in-memory model artifact, so fitting, cross-validation, prediction and evaluation all share one fitted state instead of retraining.

Source code in skyulf-core/skyulf/modeling/base.py

class StatefulEstimator:
    """Drive a model node end to end, holding the fitted model between calls.

    Wraps a calculator/applier pair and owns the in-memory ``model`` artifact,
    so fitting, cross-validation, prediction and evaluation all share one
    fitted state instead of retraining.
    """

    def __init__(self, calculator: BaseModelCalculator, applier: BaseModelApplier, node_id: str):
        """Store the node pair and start with no fitted model.

        ``node_id`` is recorded into the evaluation payload alongside
        ``job_id``; ``model`` stays ``None`` until the first fit.
        """
        self.calculator = calculator
        self.applier = applier
        self.node_id = node_id
        self.model = None  # In-memory model storage

    @staticmethod
    def _is_non_empty_split(data: Any) -> bool:
        """Engine-agnostic non-empty check for a dataset split.

        Handles pandas (`.empty`), polars/Skyulf wrappers (`.is_empty()`),
        and (X, y) tuples - previously only pandas DataFrames and tuples
        were recognized, so a bare polars DataFrame split (test/validation)
        was silently treated as absent.
        """
        if data is None:
            return False
        if isinstance(data, tuple):
            return len(data) == 2 and data[0] is not None and len(data[0]) > 0
        if hasattr(data, "empty"):
            return not data.empty
        if hasattr(data, "is_empty"):
            return not data.is_empty()
        try:
            return len(data) > 0
        except TypeError:
            return False

    def _extract_xy(self, data: Any, target_column: str) -> tuple[Any, Any]:
        """Instance-method wrapper around the module-level ``extract_xy()``.

        Kept for backward compatibility with existing call sites/tests.
        """
        return extract_xy(data, target_column)

    def cross_validate(
        self,
        dataset: SplitDataset,
        target_column: str,
        config: dict[str, Any],
        n_folds: int = 5,
        cv_type: str = "k_fold",
        shuffle: bool = True,
        random_state: int = 42,
        time_column: str | None = None,
        progress_callback: Callable[[int, int], None] | None = None,
        log_callback: Callable[[str], None] | None = None,
        preprocessing: FoldPreprocessor | None = None,
        *,
        cv_nested_type: str = "auto",
        group_column: str | None = None,
        gap: int = 0,
        test_size: int | None = None,
        max_train_size: int | None = None,
        inner_folds: int | None = None,
    ) -> dict[str, Any]:
        """Performs cross-validation on the training split."""
        X_train, y_train = self._extract_xy(dataset.train, target_column)
        from ._policy_cv import (  # noqa: PLC0415 - avoid tuning import cycle
            policy_config,
        )
        from ._tuning.cv_policy import (  # noqa: PLC0415 - avoid tuning import cycle
            validate_holdout_metadata,
        )

        policy = policy_config(
            cv_type,
            n_folds,
            shuffle,
            random_state,
            time_column,
            {
                "cv_nested_type": cv_nested_type,
                "group_column": group_column,
                "gap": gap,
                "test_size": test_size,
                "max_train_size": max_train_size,
                "inner_folds": inner_folds,
            },
        )
        for heldout in (dataset.test, dataset.validation):
            if self._is_non_empty_split(heldout):
                validate_holdout_metadata(
                    X_train,
                    self._extract_xy(heldout, target_column)[0],
                    policy,
                    self.calculator.problem_type,
                )

        return perform_cross_validation(
            calculator=self.calculator,
            applier=self.applier,
            X=X_train,
            y=y_train,
            config=config,
            n_folds=n_folds,
            cv_type=cv_type,
            shuffle=shuffle,
            random_state=random_state,
            time_column=time_column,
            progress_callback=progress_callback,
            log_callback=log_callback,
            preprocessing=preprocessing,
            cv_nested_type=cv_nested_type,
            group_column=group_column,
            gap=gap,
            test_size=test_size,
            max_train_size=max_train_size,
            inner_folds=inner_folds,
            **(
                {"sample_weight": dataset.train_sample_weight}
                if dataset.train_sample_weight is not None
                else {}
            ),
        )

    @staticmethod
    def _drop_target_column(data: Any, target_column: str) -> Any:
        """Drop target_column from data, handling pandas (kwarg) and Polars (list-arg) APIs."""
        try:
            return data.drop(columns=[target_column])
        except TypeError:
            # Polars
            return data.drop([target_column])

    def _extract_split_features(self, split_data: Any, target_column: str) -> Any:
        """Extract the feature matrix from a test/validation split, dropping the target if present.

        Handles both the ``(X, y)`` tuple form and the plain DataFrame form
        (pandas or Polars), so the same logic can be reused for the test and
        validation splits of ``fit_predict``.
        """
        if isinstance(split_data, tuple):
            return self._extract_xy(split_data, target_column)[0]

        if target_column in split_data.columns:
            return self._drop_target_column(split_data, target_column)
        return split_data

    def _normalize_fit_predict_dataset(
        self,
        dataset: SplitDataset
        | pd.DataFrame
        | pl.DataFrame
        | SkyulfDataFrame
        | tuple[pd.DataFrame, pd.Series]
        | tuple[pd.DataFrame, pd.DataFrame],
        target_column: str,
        log_callback: Callable[[str], None] | None,
    ) -> SplitDataset:
        """Wrap raw DataFrame/tuple ``fit_predict`` input into a SplitDataset.

        Handles pandas, raw (unwrapped) Polars, and wrapped ``SkyulfDataFrame``
        input alike -- checking only ``isinstance(dataset, pd.DataFrame)``
        would silently misroute a raw ``pl.DataFrame`` (e.g. the no-splitter
        fallback in ``pipeline.py``'s ``fit()``, which hands the modeling
        layer a bare frame of whatever engine produced it) into the
        ``SplitDataset``-shaped branch below, crashing with
        ``AttributeError: 'DataFrame' object has no attribute 'train'``.
        """
        if isinstance(dataset, tuple):
            # Check if it's (train_df, test_df) or (X, y)
            elem0 = dataset[0]
            elem1 = dataset[1]
            if (
                isinstance(elem0, pd.DataFrame)
                and isinstance(elem1, pd.DataFrame)
                and target_column in elem0.columns
            ):
                # It's (train_df, test_df)
                return SplitDataset(train=elem0, test=elem1, validation=None)

            # Fallback: Treat input as training data (e.g. X, y tuple) and initialize empty test set.
            msg = (
                "WARNING: No test set provided. Using entire input as training data. "
                "Ensure data was split BEFORE preprocessing to avoid data leakage."
            )
            logger.warning(msg)
            if log_callback:
                log_callback(msg)

            return SplitDataset(train=cast(Any, dataset), test=pd.DataFrame(), validation=None)

        if hasattr(dataset, "shape") and hasattr(dataset, "columns"):
            # A single frame-like object (pandas, raw Polars, or wrapper) --
            # build the empty "test" placeholder with a same-engine empty
            # frame rather than always defaulting to pandas.
            empty_test = get_engine(dataset).create_dataframe({})
            return SplitDataset(train=cast(Any, dataset), test=empty_test, validation=None)

        return dataset

    def fit_predict(
        self,
        dataset: SplitDataset
        | pd.DataFrame
        | pl.DataFrame
        | SkyulfDataFrame
        | tuple[pd.DataFrame, pd.Series]
        | tuple[pd.DataFrame, pd.DataFrame],
        target_column: str,
        config: dict[str, Any],
        progress_callback: Callable[[int, int], None] | None = None,
        log_callback: Callable[[str], None] | None = None,
        job_id: str = "unknown",
        preprocessing: FoldPreprocessor | None = None,
        preprocessing_train: tuple[Any, Any] | None = None,
        preprocessing_validation: tuple[Any, Any] | None = None,
        iteration_callback: Callable[..., None] | None = None,
        *,
        preprocessing_sample_weight: Any = None,
    ) -> dict[str, pd.Series]:
        """Fits the model on training data and returns predictions for all splits.

        ``preprocessing`` (F-15): forwarded to calculators that
        support per-fold refit (``TuningCalculator``). When set,
        ``preprocessing_train`` must carry the pre-transform ``(X, y)``
        payload the calculator should fit/tune on, so fold slicing stays
        aligned with the preprocessor; predictions still run on this
        dataset's (post-transform) splits. ``preprocessing_validation`` is
        the matching pre-transform validation payload for holdout tuning —
        ``dataset.validation`` is post-transform, so the refit cannot score
        against it directly. ``preprocessing_sample_weight`` is the original
        positional weight vector aligned with ``preprocessing_train``. It is
        required when the processed dataset carries weights: processed rows may
        have been reordered or filtered, so their weights cannot be reused.
        """
        # Handle raw DataFrame or Tuple input by wrapping it in a dummy SplitDataset
        dataset = self._normalize_fit_predict_dataset(dataset, target_column, log_callback)

        fit_weight = _fit_predict_weights(
            dataset, preprocessing, preprocessing_train, preprocessing_sample_weight
        )

        # 1. Prepare Data
        X_train, y_train = self._extract_xy(dataset.train, target_column)

        validation_data = None
        if self._is_non_empty_split(dataset.validation):
            X_val, y_val = self._extract_xy(dataset.validation, target_column)
            validation_data = (X_val, y_val)

        # 2. Train Model
        if preprocessing is not None:
            if preprocessing_train is None:
                raise ValueError("preprocessing requires the preprocessing_train (X, y) payload")
            # Only TuningCalculator accepts the hook today; the backend only
            # passes it when wrapping one, so a narrow cast keeps the generic
            # calculator interface clean.
            self.model = cast(Any, self.calculator).fit(
                preprocessing_train[0],
                preprocessing_train[1],
                config,
                progress_callback=progress_callback,
                log_callback=log_callback,
                validation_data=validation_data,
                preprocessing=preprocessing,
                validation_frames=preprocessing_validation,
                iteration_callback=iteration_callback,
                **({"sample_weight": fit_weight} if fit_weight is not None else {}),
            )
        else:
            self.model = self.calculator.fit(
                X_train,
                y_train,
                config,
                progress_callback=progress_callback,
                log_callback=log_callback,
                validation_data=validation_data,
                iteration_callback=iteration_callback,
                **({"sample_weight": fit_weight} if fit_weight is not None else {}),
            )

        # 3. Predict on all splits
        predictions = {}

        # Train Predictions
        predictions["train"] = self.applier.predict(X_train, self.model)

        # Test Predictions
        test_df = dataset.test[0] if isinstance(dataset.test, tuple) else dataset.test
        is_test_empty = len(test_df) == 0

        if not is_test_empty:
            X_test = self._extract_split_features(dataset.test, target_column)
            predictions["test"] = self.applier.predict(X_test, self.model)

        # Validation Predictions
        if self._is_non_empty_split(dataset.validation):
            X_val = self._extract_split_features(dataset.validation, target_column)
            predictions["validation"] = self.applier.predict(X_val, self.model)

        return predictions

    def evaluate(
        self,
        dataset: SplitDataset,
        target_column: str,
        job_id: str = "unknown",
        reference_column: str = "",
    ) -> Any:
        """Evaluates the model on all splits and returns a detailed report.

        ``reference_column`` is clustering-only: an optional column (e.g. a
        known label like species name) excluded from training features but
        used here purely to build a post-hoc cluster/label breakdown.
        """
        if self.model is None:
            raise ValueError("Model has not been trained yet. Call fit_predict() first.")

        problem_type = self.calculator.problem_type

        splits_payload = {}

        # Container for raw predictions
        evaluation_data: dict[str, Any] = {
            "job_id": job_id,
            "node_id": self.node_id,
            "problem_type": problem_type,
            "splits": {},
        }

        # 2. Evaluate Train
        splits_payload["train"] = self._evaluate_split(
            "train", dataset.train, target_column, problem_type, evaluation_data, reference_column
        )

        # 3. Evaluate Test
        has_test = self._is_non_empty_split(dataset.test)

        if has_test:
            splits_payload["test"] = self._evaluate_split(
                "test", dataset.test, target_column, problem_type, evaluation_data, reference_column
            )

        # 4. Evaluate Validation
        if dataset.validation is not None:
            has_val = self._is_non_empty_split(dataset.validation)

            if has_val:
                splits_payload["validation"] = self._evaluate_split(
                    "validation",
                    dataset.validation,
                    target_column,
                    problem_type,
                    evaluation_data,
                    reference_column,
                )

        self._attach_evaluation_coverage(dataset, splits_payload, evaluation_data)

        # Return report object (simplified for now, assuming schema matches)
        return {
            "problem_type": problem_type,
            "splits": splits_payload,
            "raw_data": evaluation_data,
        }

    @staticmethod
    def _attach_evaluation_coverage(
        dataset: SplitDataset, reports: dict[str, Any], raw_data: dict[str, Any]
    ) -> None:
        """Expose eligible-row denominators, including splits completely excluded by filters."""
        for name in ("train", "test", "validation"):
            payload = getattr(dataset, name)
            if payload is None:
                continue
            report = reports.get(name)
            coverage = _evaluation_split_coverage(dataset, name, report)
            if coverage is None:
                continue
            if report is None and coverage["excluded_rows"] != 0 and coverage["scored_rows"] == 0:
                report = ModelEvaluationReport(
                    dataset_name=name,
                    metrics={},
                    omitted_metrics={"evaluation": "No eligible rows remain after preprocessing."},
                )
                reports[name] = report
                raw_data["splits"][name] = (
                    {"labels": []}
                    if raw_data["problem_type"] == "clustering"
                    else {"y_true": [], "y_pred": []}
                )
            if report is not None:
                report.coverage = coverage
                raw_data["splits"].setdefault(name, {})["coverage"] = deepcopy(coverage)

    def _evaluate_split(
        self,
        split_name: str,
        data: Any,
        target_column: str,
        problem_type: str,
        evaluation_data: dict[str, Any],
        reference_column: str = "",
    ) -> Any:
        """Evaluates a single dataset split, recording raw predictions into ``evaluation_data``.

        Returns the split's evaluation report, or ``None`` if it can't be
        evaluated.
        """
        # Delegate to the same engine-agnostic (pandas/polars/tuple) X/y
        # extraction used by fit_predict, instead of duplicating
        # ad-hoc pandas-only logic that silently dropped polars splits.
        try:
            X, y = self._extract_xy(data, target_column)
        except ValueError:
            return None  # Cannot evaluate without target
        if X is None:
            return None
        if problem_type != "clustering" and y is None:
            return None

        X = self._evaluation_features(X)
        y_pred = self.applier.predict(X, self.model)
        model_to_evaluate = self._unwrap_tuned_model()

        if problem_type == "clustering":
            # Unsupervised: there is no y_true, only the cluster label
            # assigned to each row. KMeans genuinely supports out-of-sample
            # `.predict()`, so (unlike DBSCAN/Agglomerative) evaluating each
            # split independently with its own predicted labels is valid.
            split_report = self._evaluate_split_with_model(
                model_to_evaluate, split_name, X, y_pred, problem_type, reference_column
            )
            evaluation_data["splits"][split_name] = self._build_clustering_split_raw_data(
                y_pred, split_report
            )
            return split_report

        y_proba = self._predict_proba_payload(X, problem_type)
        evaluation_data["splits"][split_name] = self._build_split_raw_data(y, y_pred, y_proba)

        return self._evaluate_split_with_model(
            model_to_evaluate, split_name, X, y, problem_type, predictions=y_pred
        )

    @staticmethod
    def _build_split_raw_data(
        y: Any, y_pred: Any, y_proba: dict[str, Any] | None
    ) -> dict[str, Any]:
        """Builds the raw ``y_true``/``y_pred``/(optional) ``y_proba`` payload for a split."""
        split_data = {
            "y_true": y.tolist() if hasattr(y, "tolist") else list(y),
            "y_pred": (y_pred.tolist() if hasattr(y_pred, "tolist") else list(y_pred)),
        }
        if y_proba:
            split_data["y_proba"] = y_proba
        return split_data

    def _evaluation_features(self, X: Any) -> Any:
        """Exclude persisted split metadata from both predictions and metric evaluation."""
        from ._tuning.cv_policy import (  # noqa: PLC0415 - avoid tuning import cycle
            prediction_features,
        )

        if isinstance(self.model, tuple) and len(self.model) == 2:
            return prediction_features(X, self.model[1])
        return X

    @staticmethod
    def _build_clustering_split_raw_data(labels: Any, split_report: Any = None) -> dict[str, Any]:
        """Builds the raw ``labels`` (+ clustering summary/metrics) payload for a clustering split.

        ``split_report`` is the ``ModelEvaluationReport`` for this split, if evaluation
        succeeded; its ``clustering`` field (cluster sizes/centroids) and quality
        ``metrics`` (silhouette/Calinski-Harabasz/Davies-Bouldin) are embedded so the
        API doesn't need a second round-trip to expose them.
        """
        raw: dict[str, Any] = {
            "labels": labels.tolist() if hasattr(labels, "tolist") else list(labels)
        }
        if split_report is not None:
            clustering = getattr(split_report, "clustering", None)
            if clustering is not None:
                raw["clustering"] = clustering.model_dump()
            metrics = getattr(split_report, "metrics", None)
            if metrics is not None:
                raw["metrics"] = dict(metrics)
        return raw

    def _unwrap_tuned_model(self) -> Any:
        """Unpacks ``self.model`` if it's a ``(model, ...)`` tuple, as produced by the Tuner."""
        # Check if first element looks like a model (has fit/predict)
        # or if it's just a convention from TuningCalculator
        if isinstance(self.model, tuple) and len(self.model) == 2:
            return self.model[0]
        return self.model

    def _predict_proba_payload(self, X: Any, problem_type: str) -> dict[str, Any] | None:
        """Returns the ``{"classes", "values"}`` probability payload for classification splits."""
        if problem_type != "classification":
            return None
        y_proba_df = self.applier.predict_proba(X, self.model)
        if y_proba_df is None:
            return None
        y_proba_df = cast(pd.DataFrame, y_proba_df)
        return {
            "classes": y_proba_df.columns.tolist(),
            "values": y_proba_df.to_numpy().tolist(),
        }

    @staticmethod
    def _evaluate_split_with_model(
        model_to_evaluate: Any,
        split_name: str,
        X: Any,
        y: Any,
        problem_type: str,
        reference_column: str = "",
        *,
        predictions: Any | None = None,
    ) -> Any:
        """Dispatches to the classification, regression, or clustering evaluator.

        For clustering, ``y`` is the *predicted* cluster labels for this split
        (there is no ground-truth target), computed by the caller via
        ``self.applier.predict(X, self.model)``.
        """
        if problem_type == "classification":
            return evaluate_classification_model(
                model=model_to_evaluate,
                dataset_name=split_name,
                X_test=X,
                y_test=y,
                predictions=predictions,
            )
        elif problem_type == "regression":
            return evaluate_regression_model(
                model=model_to_evaluate, dataset_name=split_name, X_test=X, y_test=y
            )
        elif problem_type == "clustering":
            return evaluate_clustering_model(
                model=model_to_evaluate,
                X=X,
                labels=y,
                dataset_name=split_name,
                reference_column=reference_column,
            )
        else:
            raise_invalid_choice(
                problem_type, ("classification", "regression", "clustering"), "problem type"
            )

__init__(calculator, applier, node_id)

Store the node pair and start with no fitted model.

node_id is recorded into the evaluation payload alongside job_id; model stays None until the first fit.

Source code in skyulf-core/skyulf/modeling/base.py

def __init__(self, calculator: BaseModelCalculator, applier: BaseModelApplier, node_id: str):
    """Store the node pair and start with no fitted model.

    ``node_id`` is recorded into the evaluation payload alongside
    ``job_id``; ``model`` stays ``None`` until the first fit.
    """
    self.calculator = calculator
    self.applier = applier
    self.node_id = node_id
    self.model = None  # In-memory model storage

cross_validate(dataset, target_column, config, n_folds=5, cv_type='k_fold', shuffle=True, random_state=42, time_column=None, progress_callback=None, log_callback=None, preprocessing=None, *, cv_nested_type='auto', group_column=None, gap=0, test_size=None, max_train_size=None, inner_folds=None)

Performs cross-validation on the training split.

Source code in skyulf-core/skyulf/modeling/base.py

def cross_validate(
    self,
    dataset: SplitDataset,
    target_column: str,
    config: dict[str, Any],
    n_folds: int = 5,
    cv_type: str = "k_fold",
    shuffle: bool = True,
    random_state: int = 42,
    time_column: str | None = None,
    progress_callback: Callable[[int, int], None] | None = None,
    log_callback: Callable[[str], None] | None = None,
    preprocessing: FoldPreprocessor | None = None,
    *,
    cv_nested_type: str = "auto",
    group_column: str | None = None,
    gap: int = 0,
    test_size: int | None = None,
    max_train_size: int | None = None,
    inner_folds: int | None = None,
) -> dict[str, Any]:
    """Performs cross-validation on the training split."""
    X_train, y_train = self._extract_xy(dataset.train, target_column)
    from ._policy_cv import (  # noqa: PLC0415 - avoid tuning import cycle
        policy_config,
    )
    from ._tuning.cv_policy import (  # noqa: PLC0415 - avoid tuning import cycle
        validate_holdout_metadata,
    )

    policy = policy_config(
        cv_type,
        n_folds,
        shuffle,
        random_state,
        time_column,
        {
            "cv_nested_type": cv_nested_type,
            "group_column": group_column,
            "gap": gap,
            "test_size": test_size,
            "max_train_size": max_train_size,
            "inner_folds": inner_folds,
        },
    )
    for heldout in (dataset.test, dataset.validation):
        if self._is_non_empty_split(heldout):
            validate_holdout_metadata(
                X_train,
                self._extract_xy(heldout, target_column)[0],
                policy,
                self.calculator.problem_type,
            )

    return perform_cross_validation(
        calculator=self.calculator,
        applier=self.applier,
        X=X_train,
        y=y_train,
        config=config,
        n_folds=n_folds,
        cv_type=cv_type,
        shuffle=shuffle,
        random_state=random_state,
        time_column=time_column,
        progress_callback=progress_callback,
        log_callback=log_callback,
        preprocessing=preprocessing,
        cv_nested_type=cv_nested_type,
        group_column=group_column,
        gap=gap,
        test_size=test_size,
        max_train_size=max_train_size,
        inner_folds=inner_folds,
        **(
            {"sample_weight": dataset.train_sample_weight}
            if dataset.train_sample_weight is not None
            else {}
        ),
    )

evaluate(dataset, target_column, job_id='unknown', reference_column='')

Evaluates the model on all splits and returns a detailed report.

reference_column is clustering-only: an optional column (e.g. a known label like species name) excluded from training features but used here purely to build a post-hoc cluster/label breakdown.

Source code in skyulf-core/skyulf/modeling/base.py

def evaluate(
    self,
    dataset: SplitDataset,
    target_column: str,
    job_id: str = "unknown",
    reference_column: str = "",
) -> Any:
    """Evaluates the model on all splits and returns a detailed report.

    ``reference_column`` is clustering-only: an optional column (e.g. a
    known label like species name) excluded from training features but
    used here purely to build a post-hoc cluster/label breakdown.
    """
    if self.model is None:
        raise ValueError("Model has not been trained yet. Call fit_predict() first.")

    problem_type = self.calculator.problem_type

    splits_payload = {}

    # Container for raw predictions
    evaluation_data: dict[str, Any] = {
        "job_id": job_id,
        "node_id": self.node_id,
        "problem_type": problem_type,
        "splits": {},
    }

    # 2. Evaluate Train
    splits_payload["train"] = self._evaluate_split(
        "train", dataset.train, target_column, problem_type, evaluation_data, reference_column
    )

    # 3. Evaluate Test
    has_test = self._is_non_empty_split(dataset.test)

    if has_test:
        splits_payload["test"] = self._evaluate_split(
            "test", dataset.test, target_column, problem_type, evaluation_data, reference_column
        )

    # 4. Evaluate Validation
    if dataset.validation is not None:
        has_val = self._is_non_empty_split(dataset.validation)

        if has_val:
            splits_payload["validation"] = self._evaluate_split(
                "validation",
                dataset.validation,
                target_column,
                problem_type,
                evaluation_data,
                reference_column,
            )

    self._attach_evaluation_coverage(dataset, splits_payload, evaluation_data)

    # Return report object (simplified for now, assuming schema matches)
    return {
        "problem_type": problem_type,
        "splits": splits_payload,
        "raw_data": evaluation_data,
    }

fit_predict(dataset, target_column, config, progress_callback=None, log_callback=None, job_id='unknown', preprocessing=None, preprocessing_train=None, preprocessing_validation=None, iteration_callback=None, *, preprocessing_sample_weight=None)

Fits the model on training data and returns predictions for all splits.

preprocessing (F-15): forwarded to calculators that support per-fold refit (TuningCalculator). When set, preprocessing_train must carry the pre-transform (X, y) payload the calculator should fit/tune on, so fold slicing stays aligned with the preprocessor; predictions still run on this dataset's (post-transform) splits. preprocessing_validation is the matching pre-transform validation payload for holdout tuning — dataset.validation is post-transform, so the refit cannot score against it directly. preprocessing_sample_weight is the original positional weight vector aligned with preprocessing_train. It is required when the processed dataset carries weights: processed rows may have been reordered or filtered, so their weights cannot be reused.

Source code in skyulf-core/skyulf/modeling/base.py

def fit_predict(
    self,
    dataset: SplitDataset
    | pd.DataFrame
    | pl.DataFrame
    | SkyulfDataFrame
    | tuple[pd.DataFrame, pd.Series]
    | tuple[pd.DataFrame, pd.DataFrame],
    target_column: str,
    config: dict[str, Any],
    progress_callback: Callable[[int, int], None] | None = None,
    log_callback: Callable[[str], None] | None = None,
    job_id: str = "unknown",
    preprocessing: FoldPreprocessor | None = None,
    preprocessing_train: tuple[Any, Any] | None = None,
    preprocessing_validation: tuple[Any, Any] | None = None,
    iteration_callback: Callable[..., None] | None = None,
    *,
    preprocessing_sample_weight: Any = None,
) -> dict[str, pd.Series]:
    """Fits the model on training data and returns predictions for all splits.

    ``preprocessing`` (F-15): forwarded to calculators that
    support per-fold refit (``TuningCalculator``). When set,
    ``preprocessing_train`` must carry the pre-transform ``(X, y)``
    payload the calculator should fit/tune on, so fold slicing stays
    aligned with the preprocessor; predictions still run on this
    dataset's (post-transform) splits. ``preprocessing_validation`` is
    the matching pre-transform validation payload for holdout tuning —
    ``dataset.validation`` is post-transform, so the refit cannot score
    against it directly. ``preprocessing_sample_weight`` is the original
    positional weight vector aligned with ``preprocessing_train``. It is
    required when the processed dataset carries weights: processed rows may
    have been reordered or filtered, so their weights cannot be reused.
    """
    # Handle raw DataFrame or Tuple input by wrapping it in a dummy SplitDataset
    dataset = self._normalize_fit_predict_dataset(dataset, target_column, log_callback)

    fit_weight = _fit_predict_weights(
        dataset, preprocessing, preprocessing_train, preprocessing_sample_weight
    )

    # 1. Prepare Data
    X_train, y_train = self._extract_xy(dataset.train, target_column)

    validation_data = None
    if self._is_non_empty_split(dataset.validation):
        X_val, y_val = self._extract_xy(dataset.validation, target_column)
        validation_data = (X_val, y_val)

    # 2. Train Model
    if preprocessing is not None:
        if preprocessing_train is None:
            raise ValueError("preprocessing requires the preprocessing_train (X, y) payload")
        # Only TuningCalculator accepts the hook today; the backend only
        # passes it when wrapping one, so a narrow cast keeps the generic
        # calculator interface clean.
        self.model = cast(Any, self.calculator).fit(
            preprocessing_train[0],
            preprocessing_train[1],
            config,
            progress_callback=progress_callback,
            log_callback=log_callback,
            validation_data=validation_data,
            preprocessing=preprocessing,
            validation_frames=preprocessing_validation,
            iteration_callback=iteration_callback,
            **({"sample_weight": fit_weight} if fit_weight is not None else {}),
        )
    else:
        self.model = self.calculator.fit(
            X_train,
            y_train,
            config,
            progress_callback=progress_callback,
            log_callback=log_callback,
            validation_data=validation_data,
            iteration_callback=iteration_callback,
            **({"sample_weight": fit_weight} if fit_weight is not None else {}),
        )

    # 3. Predict on all splits
    predictions = {}

    # Train Predictions
    predictions["train"] = self.applier.predict(X_train, self.model)

    # Test Predictions
    test_df = dataset.test[0] if isinstance(dataset.test, tuple) else dataset.test
    is_test_empty = len(test_df) == 0

    if not is_test_empty:
        X_test = self._extract_split_features(dataset.test, target_column)
        predictions["test"] = self.applier.predict(X_test, self.model)

    # Validation Predictions
    if self._is_non_empty_split(dataset.validation):
        X_val = self._extract_split_features(dataset.validation, target_column)
        predictions["validation"] = self.applier.predict(X_val, self.model)

    return predictions

VotingClassifierApplier

Bases: SklearnApplier

Voting Classifier Applier (hard/soft vote over base classifiers).

Source code in skyulf-core/skyulf/modeling/ensemble.py

class VotingClassifierApplier(SklearnApplier):
    """Voting Classifier Applier (hard/soft vote over base classifiers)."""

VotingClassifierCalculator

Bases: _BaseEnsembleCalculator

Voting Classifier Calculator with selectable base learners.

Source code in skyulf-core/skyulf/modeling/ensemble.py

@NodeRegistry.register("voting_classifier", VotingClassifierApplier)
@node_meta(
    id="voting_classifier",
    name="Voting Classifier",
    category="Ensemble",
    description=(
        "Combines several classifiers by majority vote (hard) or averaged "
        "probabilities (soft). Fits each base model once; no internal CV."
    ),
    params={
        "base_estimators": ["random_forest", "logistic_regression", "gradient_boosting"],
        "voting": "soft",
    },
    tags=["requires_scaling", "classification"],
    learns_from_data=True,
)
class VotingClassifierCalculator(_BaseEnsembleCalculator):
    """Voting Classifier Calculator with selectable base learners."""

    BASE_ESTIMATORS = BASE_ESTIMATORS_CLF
    DEFAULT_KEYS = ("random_forest", "logistic_regression", "gradient_boosting")
    MODEL_KEY = "voting_classifier"
    HAS_VOTING = True

    def __init__(self):
        """Pin the meta-estimator to ``VotingClassifier`` with ``voting="soft"``."""
        super().__init__(
            model_class=VotingClassifier,
            default_params={"voting": "soft"},
            problem_type="classification",
        )

__init__()

Pin the meta-estimator to VotingClassifier with voting="soft".

Source code in skyulf-core/skyulf/modeling/ensemble.py

def __init__(self):
    """Pin the meta-estimator to ``VotingClassifier`` with ``voting="soft"``."""
    super().__init__(
        model_class=VotingClassifier,
        default_params={"voting": "soft"},
        problem_type="classification",
    )

VotingRegressorApplier

Bases: SklearnApplier

Voting Regressor Applier (averaged predictions over base regressors).

Source code in skyulf-core/skyulf/modeling/ensemble.py

class VotingRegressorApplier(SklearnApplier):
    """Voting Regressor Applier (averaged predictions over base regressors)."""

VotingRegressorCalculator

Bases: _BaseEnsembleCalculator

Voting Regressor Calculator with selectable base learners.

Source code in skyulf-core/skyulf/modeling/ensemble.py

@NodeRegistry.register("voting_regressor", VotingRegressorApplier)
@node_meta(
    id="voting_regressor",
    name="Voting Regressor",
    category="Ensemble",
    description=(
        "Averages the predictions of several regressors (optionally weighted). "
        "Fits each base model once; no internal CV."
    ),
    params={
        "base_estimators": ["linear_regression", "random_forest", "gradient_boosting"],
    },
    tags=["requires_scaling", "regression"],
    learns_from_data=True,
)
class VotingRegressorCalculator(_BaseEnsembleCalculator):
    """Voting Regressor Calculator with selectable base learners."""

    BASE_ESTIMATORS = BASE_ESTIMATORS_REG
    DEFAULT_KEYS = ("linear_regression", "random_forest", "gradient_boosting")
    MODEL_KEY = "voting_regressor"

    def __init__(self):
        """Pin the meta-estimator to ``VotingRegressor``, with no estimator-level defaults."""
        super().__init__(
            model_class=VotingRegressor,
            default_params={},
            problem_type="regression",
        )

__init__()

Pin the meta-estimator to VotingRegressor, with no estimator-level defaults.

Source code in skyulf-core/skyulf/modeling/ensemble.py

def __init__(self):
    """Pin the meta-estimator to ``VotingRegressor``, with no estimator-level defaults."""
    super().__init__(
        model_class=VotingRegressor,
        default_params={},
        problem_type="regression",
    )

apply_thresholds(y_proba, thresholds, classes=None, *, positive_class=None)

Convert predicted probabilities into class predictions using per-class thresholds.

Binary (thresholds is a single float, or a dict over the two classes): predicts the positive (second) class when y_proba[:, 1] >= threshold, else the first class. A one-entry dict applies its cutoff to the named class's probability, with ties selecting that class. An explicit positive_class must agree with that key. A full-coverage pair (the shape returned by optimize_thresholds, where t0 == 1 - t1) is compared in scaled space (p1/t1 >= p0/t0), which reduces to the same rule. Either way an exact tie predicts the positive class, matching >=.

Multiclass (3+ classes, thresholds covering every class): scaled argmax — classes[argmax(y_proba / thresholds, axis=1)]. Equal thresholds across all classes reduce to plain argmax. Each threshold must be finite and strictly positive; values above one are valid weights.

Parameters:

Name	Type	Description	Default
y_proba	Any	Array-like of shape (n_samples, n_classes), predicted probabilities in the same column order as classes.	required
thresholds	dict[Any, float] | float	A single float (binary), or a dict mapping every class label present in classes to its threshold.	required
classes	Any	Explicit class label order matching y_proba's columns. Required when y_proba has more than 2 columns and thresholds is a dict (to know column-to-class mapping).	None
positive_class	Any	Optional binary positive label; it wins exact cutoff ties.	None

Returns:

Type	Description
ndarray	1D numpy array of predicted class labels, length n_samples.

Raises:

Type	Description
ValueError	If y_proba isn't 2D, or thresholds doesn't cover every class implied by y_proba's column count.

Source code in skyulf-core/skyulf/modeling/_evaluation/thresholds.py

def apply_thresholds(
    y_proba: Any,
    thresholds: dict[Any, float] | float,
    classes: Any = None,
    *,
    positive_class: Any = None,
) -> np.ndarray:
    """Convert predicted probabilities into class predictions using per-class thresholds.

    Binary (``thresholds`` is a single float, or a dict over the two classes):
    predicts the positive (second) class when ``y_proba[:, 1] >= threshold``,
    else the first class. A one-entry dict applies its cutoff to the named
    class's probability, with ties selecting that class. An explicit
    ``positive_class`` must agree with that key. A full-coverage pair
    (the shape returned by ``optimize_thresholds``, where ``t0 == 1 - t1``)
    is compared in scaled space (``p1/t1 >= p0/t0``), which
    reduces to the same rule. Either way an exact tie predicts the positive
    class, matching ``>=``.

    Multiclass (3+ classes, ``thresholds`` covering every class): scaled
    argmax — ``classes[argmax(y_proba / thresholds, axis=1)]``. Equal
    thresholds across all classes reduce to plain argmax. Each threshold must
    be finite and strictly positive; values above one are valid weights.

    Args:
        y_proba: Array-like of shape (n_samples, n_classes), predicted
            probabilities in the same column order as ``classes``.
        thresholds: A single float (binary), or a dict mapping every class
            label present in ``classes`` to its threshold.
        classes: Explicit class label order matching ``y_proba``'s columns.
            Required when ``y_proba`` has more than 2 columns and
            ``thresholds`` is a dict (to know column-to-class mapping).
        positive_class: Optional binary positive label; it wins exact cutoff ties.

    Returns:
        1D numpy array of predicted class labels, length n_samples.

    Raises:
        ValueError: If ``y_proba`` isn't 2D, or ``thresholds`` doesn't cover
            every class implied by ``y_proba``'s column count.
    """
    y_proba, classes = _probability_columns(y_proba, classes)
    n_classes = y_proba.shape[1]
    if positive_class is not None:
        return _apply_named_binary(y_proba, thresholds, classes, positive_class)

    if n_classes == 2 and not isinstance(thresholds, dict):
        threshold = float(thresholds)
        return np.where(y_proba[:, 1] >= threshold, classes[1], classes[0])

    if not isinstance(thresholds, dict):
        raise ValueError(
            "thresholds must be a dict mapping each class to its threshold "
            "for multiclass input (or when passing a single-entry dict for binary)."
        )

    if n_classes == 2 and len(thresholds) == 1:
        return _apply_named_binary(y_proba, thresholds, classes, next(iter(thresholds)))

    thresholds_array = _class_threshold_array(thresholds, classes)
    if n_classes == 2:
        return _apply_binary_weights(y_proba, thresholds_array, classes)
    scaled = y_proba / thresholds_array
    return classes[np.argmax(scaled, axis=1)]

calculate_classification_metrics(model, X, y, *, X_np=None, y_np=None, predictions=None, proba=None, omitted_metrics=None)

Compute classification metrics for predictions.

X_np/y_np/predictions/proba let a caller that already converted X/y to numpy and/or already called model.predict()/model.predict_proba() (e.g. evaluate_classification_model) pass those results straight through, avoiding a redundant conversion/inference pass on the same data. When omitted, each is (re)computed here exactly as before. omitted_metrics, when supplied, collects failure or non-finite-result reasons for the report without inserting strings into the scalar metrics.

Source code in skyulf-core/skyulf/modeling/_evaluation/metrics.py

def calculate_classification_metrics(
    model: Any,
    X: pd.DataFrame | pl.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    *,
    X_np: Any = None,
    y_np: Any = None,
    predictions: Any = None,
    proba: Any = None,
    omitted_metrics: dict[str, str] | None = None,
) -> dict[str, float]:
    """Compute classification metrics for predictions.

    ``X_np``/``y_np``/``predictions``/``proba`` let a caller that already
    converted ``X``/``y`` to numpy and/or already called
    ``model.predict()``/``model.predict_proba()`` (e.g.
    ``evaluate_classification_model``) pass those results straight through,
    avoiding a redundant conversion/inference pass on the same data. When
    omitted, each is (re)computed here exactly as before.
    ``omitted_metrics``, when supplied, collects failure or non-finite-result
    reasons for the report without inserting strings into the scalar metrics.
    """
    # Convert to Numpy for compatibility (skip if the caller already has it)
    if X_np is None or y_np is None:
        X_np, y_np = SklearnBridge.to_sklearn((X, y), validate_features=True)

    # Use DataFrame directly if possible to preserve feature names
    # Only convert to numpy if model doesn't support pandas or if X is not pandas

    if predictions is None:
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", message=".*valid feature names.*")
            predictions = model.predict(X_np)

    # For metrics calculation, we might need numpy arrays for y
    y_arr = y_np

    metrics: dict[str, float] = {
        "accuracy": float(accuracy_score(y_arr, predictions)),
        "balanced_accuracy": float(balanced_accuracy_score(y_arr, predictions)),
        "precision_weighted": float(
            precision_score(y_arr, predictions, average="weighted", zero_division=0)
        ),
        "recall_weighted": float(
            recall_score(y_arr, predictions, average="weighted", zero_division=0)
        ),
        "f1_weighted": float(f1_score(y_arr, predictions, average="weighted", zero_division=0)),
        "matthews_corrcoef": float(matthews_corrcoef(y_arr, predictions)),
    }

    _add_binary_unweighted_metrics(metrics, model, y_arr, predictions, omitted_metrics)

    if geometric_mean_score is not None:
        _try_add_metric(
            metrics,
            "g_score",
            geometric_mean_score,
            y_arr,
            predictions,
            average="weighted",
            omitted_metrics=omitted_metrics,
        )

    _add_probability_based_metrics(
        metrics, model, X_np, y_arr, proba=proba, omitted_metrics=omitted_metrics
    )
    if omitted_metrics is not None:
        for key, value in metrics.items():
            if not np.isfinite(value):
                omitted_metrics[key] = (
                    "Metric is undefined for this evaluation split (non-finite result)."
                )

    return metrics

calculate_clustering_metrics(X, labels, *, silhouette_sample_size=DEFAULT_SILHOUETTE_SAMPLE_SIZE, random_state=DEFAULT_SILHOUETTE_RANDOM_STATE)

Compute unsupervised clustering-quality metrics for a fitted model's labels.

All three metrics only need the feature matrix and the cluster labels (no ground-truth target), so they can be computed on any split a KMeans model has genuinely predicted on (train/test/validation alike).

Source code in skyulf-core/skyulf/modeling/_evaluation/metrics.py

def calculate_clustering_metrics(
    X: pd.DataFrame | pl.DataFrame | SkyulfDataFrame,
    labels: Any,
    *,
    silhouette_sample_size: SilhouetteSampleSize = DEFAULT_SILHOUETTE_SAMPLE_SIZE,
    random_state: int = DEFAULT_SILHOUETTE_RANDOM_STATE,
) -> dict[str, float]:
    """Compute unsupervised clustering-quality metrics for a fitted model's labels.

    All three metrics only need the feature matrix and the cluster labels
    (no ground-truth target), so they can be computed on any split a KMeans
    model has genuinely predicted on (train/test/validation alike).
    """
    X_np, _ = SklearnBridge.to_sklearn((X, None), validate_features=True)
    labels_np = np.asarray(labels).ravel()
    silhouette_sample_size = _validate_silhouette_sample_size(silhouette_sample_size)
    # Guard before the row-count check: polars collapses a 0-column frame to
    # shape (0, 0), so otherwise a 0-feature input raises the misleading
    # row-count error instead of the sklearn-style "0 feature" message.
    if X_np.ndim != 2 or X_np.shape[1] == 0:
        raise ValueError(
            f"Found array with 0 feature(s) (shape={X_np.shape}) while a minimum of 1 is required."
        )
    if X_np.shape[0] != len(labels_np):
        raise ValueError("X and labels must have the same number of rows")

    # DBSCAN reserves -1 for samples that do not belong to any cluster. Noise
    # must not inflate the cluster count or influence quality scores. Keep the
    # common no-noise path allocation-free for large label arrays.
    if np.any(labels_np == -1):
        non_noise = labels_np != -1
        X_np = X_np[non_noise]
        labels_np = labels_np[non_noise]

    n_samples = len(labels_np)
    representative_by_label = _collect_silhouette_representatives(
        labels_np,
        sample_size=silhouette_sample_size,
    )
    n_unique = len(representative_by_label)
    metrics: dict[str, float] = {"n_clusters": float(n_unique)}

    # These metrics are undefined for fewer than 2 clusters, or when the
    # cluster count reaches the sample count — guard rather than let sklearn raise.
    if 1 < n_unique < n_samples:
        sampled_indices = _select_silhouette_sample_indices(
            labels_np,
            sample_size=silhouette_sample_size,
            random_state=random_state,
            representative_by_label=representative_by_label,
        )
        sampled_X = X_np[sampled_indices]
        sampled_labels = labels_np[sampled_indices]
        metrics["silhouette_score"] = float(
            sklearn_metrics.silhouette_score(sampled_X, sampled_labels)
        )
        metrics["silhouette_sample_size"] = float(len(sampled_indices))
        metrics["calinski_harabasz_score"] = float(
            sklearn_metrics.calinski_harabasz_score(X_np, labels_np)
        )
        metrics["davies_bouldin_score"] = float(
            sklearn_metrics.davies_bouldin_score(X_np, labels_np)
        )

    return metrics

calculate_regression_metrics(model, X, y, *, X_np=None, y_np=None, predictions=None)

Compute regression metrics for predictions.

X_np/y_np/predictions let a caller that already converted X/y to numpy and/or already called model.predict() (e.g. evaluate_regression_model) pass those results straight through, avoiding a redundant conversion/inference pass on the same data.

Source code in skyulf-core/skyulf/modeling/_evaluation/metrics.py

def calculate_regression_metrics(
    model: Any,
    X: pd.DataFrame | pl.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    *,
    X_np: Any = None,
    y_np: Any = None,
    predictions: Any = None,
) -> dict[str, float]:
    """Compute regression metrics for predictions.

    ``X_np``/``y_np``/``predictions`` let a caller that already converted
    ``X``/``y`` to numpy and/or already called ``model.predict()`` (e.g.
    ``evaluate_regression_model``) pass those results straight through,
    avoiding a redundant conversion/inference pass on the same data.
    """
    # Convert to Numpy for compatibility (skip if the caller already has it)
    if X_np is None or y_np is None:
        X_np, y_np = SklearnBridge.to_sklearn((X, y), validate_features=True)

    # Use DataFrame directly if possible to preserve feature names
    if predictions is None:
        predictions = model.predict(X_np)

    y_arr = y_np

    mse_value = mean_squared_error(y_arr, predictions)
    metrics: dict[str, float] = {
        "mae": float(mean_absolute_error(y_arr, predictions)),
        "mse": float(mse_value),
        "rmse": float(math.sqrt(mse_value)),
        "r2": float(r2_score(y_arr, predictions)),
        "mape": float(mean_absolute_percentage_error(y_arr, predictions)),
        "explained_variance": float(explained_variance_score(y_arr, predictions)),
    }

    return metrics

compute_shap_explanation(model, X, max_samples=200, max_display_samples=_DEFAULT_MAX_DISPLAY_SAMPLES)

Compute a SHAP explanation for a trained model.

Pairs a global summary with a small set of per-sample explanations for richer single-run plots.

Best-effort: returns None (never raises) if shap isn't installed, the model type is unsupported, or computation fails for any reason.

Returns a dict shaped as

{ "feature_names": [...], "mean_abs_importance": {feature: value, ...}, "samples": [ { "base_value": float, "feature_values": {feature: value, ...}, "shap_values": {feature: value, ...}, }, ... ], "interactions": { "feature_names": [...], # top-K features, or None if unavailable "matrix": [[...], ...], # mean(|interaction value|), same order as feature_names } | None, }

Source code in skyulf-core/skyulf/modeling/_explainability/shap_explanation.py

def compute_shap_explanation(
    model: Any,
    X: pd.DataFrame,
    max_samples: int = 200,
    max_display_samples: int = _DEFAULT_MAX_DISPLAY_SAMPLES,
) -> dict[str, Any] | None:
    """Compute a SHAP explanation for a trained model.

    Pairs a global summary with a small set of per-sample explanations for
    richer single-run plots.

    Best-effort: returns `None` (never raises) if `shap` isn't installed,
    the model type is unsupported, or computation fails for any reason.

    Returns a dict shaped as:
        {
            "feature_names": [...],
            "mean_abs_importance": {feature: value, ...},
            "samples": [
                {
                    "base_value": float,
                    "feature_values": {feature: value, ...},
                    "shap_values": {feature: value, ...},
                },
                ...
            ],
            "interactions": {
                "feature_names": [...],  # top-K features, or None if unavailable
                "matrix": [[...], ...],  # mean(|interaction value|), same order as feature_names
            } | None,
        }
    """
    try:
        import shap  # ty: ignore[unresolved-import]  # noqa: PLC0415 - optional explainability extra
        import shap.maskers  # ty: ignore[unresolved-import]  # noqa: PLC0415 - optional explainability extra
    except ImportError:
        return None

    # `shap.utils._exceptions` is a private module that may not exist in all
    # SHAP versions. We try to import it for the precise `ExplainerError`
    # catch below, but if it's unavailable we fall back to a broader
    # `Exception` catch + message-based heuristic — this keeps explainability
    # working across SHAP versions instead of silently returning `None`.
    _shap_exceptions: Any = None
    try:
        import shap.utils._exceptions  # ty: ignore[unresolved-import]  # noqa: PLC0415 - optional explainability extra

        _shap_exceptions = shap.utils._exceptions
    except ImportError:
        logger.debug(
            "shap.utils._exceptions unavailable; using broad catch for additivity failures"
        )

    try:
        if X is None or X.empty:
            return None

        sample = (
            X.sample(n=max_samples, random_state=DEFAULT_RANDOM_STATE)
            if len(X) > max_samples
            else X
        )

        feature_names = list(sample.columns)
        if not feature_names:
            return None

        explainer, is_exact_tree = _build_explainer(shap, model, sample)
        explanation = _explain_sample(explainer, sample, _shap_exceptions, is_exact_tree, model)
        shap_values = getattr(explanation, "values", explanation)

        mean_abs_importance = _mean_abs_per_feature(shap_values, feature_names)
        if mean_abs_importance is None:
            return None

        resolved = _per_sample_shap_and_base(
            shap_values, getattr(explanation, "base_values", 0.0), model, sample
        )
        samples = _display_shap_samples(resolved, sample, feature_names, max_display_samples)

        interactions = _compute_interaction_summary(model, sample, feature_names)

        return {
            "feature_names": feature_names,
            "mean_abs_importance": mean_abs_importance,
            "samples": samples,
            "interactions": interactions,
        }
    except Exception as exc:  # noqa: BLE001 - explainability is best-effort; failure is logged, not raised (see below)
        # Previously logged at `debug` level, which meant a broken SHAP path
        # was silently indistinguishable from "model type unsupported" —
        # this rotted invisibly for every job until the UI's `shap_explanation`
        # data was found to be missing entirely. `warning` (with the actual
        # exception) ensures this is visible without raising and breaking
        # training, since explainability is always best-effort.
        logger.warning(
            "Failed to compute SHAP explanation for model_type=%s: %s",
            type(model).__name__,
            exc,
            exc_info=True,
        )
        return None

get_default_search_space(model_key, strategy='random')

Return an independent copy of the default search space for model_key.

For grid-based strategies (grid / halving_grid) the trimmed GRID_SEARCH_SPACES dict is used so the cartesian product stays manageable. All other strategies (random, halving_random, optuna) use the richer DEFAULT_SEARCH_SPACES.

Voting regression has no tunable meta-parameters. Its empty space can be expanded with the selected learners by build_ensemble_search_space.

Source code in skyulf-core/skyulf/modeling/hyperparameters/_registry.py

def get_default_search_space(model_key: str, strategy: str = "random") -> dict[str, Any]:
    """Return an independent copy of the default search space for *model_key*.

    For grid-based strategies (``grid`` / ``halving_grid``) the trimmed
    ``GRID_SEARCH_SPACES`` dict is used so the cartesian product stays
    manageable. All other strategies (``random``, ``halving_random``,
    ``optuna``) use the richer ``DEFAULT_SEARCH_SPACES``.

    Voting regression has no tunable meta-parameters. Its empty space can
    be expanded with the selected learners by ``build_ensemble_search_space``.
    """
    if strategy in _GRID_STRATEGIES:
        return deepcopy(GRID_SEARCH_SPACES.get(model_key, DEFAULT_SEARCH_SPACES.get(model_key, {})))
    return deepcopy(DEFAULT_SEARCH_SPACES.get(model_key, {}))

get_hyperparameters(model_key)

Return one model's hyperparameter fields in dict form.

The lookup the backend's meta router uses to render a model node's parameter controls and tuning search space. Unregistered keys yield an empty list rather than raising, so an unknown model simply shows no tunable parameters.

Source code in skyulf-core/skyulf/modeling/hyperparameters/_registry.py

def get_hyperparameters(model_key: str) -> list[dict[str, Any]]:
    """Return one model's hyperparameter fields in dict form.

    The lookup the backend's meta router uses to render a model node's
    parameter controls and tuning search space. Unregistered keys yield an
    empty list rather than raising, so an unknown model simply shows no
    tunable parameters.
    """
    params = MODEL_HYPERPARAMETERS.get(model_key, [])
    return [p.to_dict() for p in params]

optimize_thresholds(y_true, y_proba, metric, classes=None, strategy=None, grid_points=101)

Search for per-class decision thresholds that maximize metric.

Parameters:

Name	Type	Description	Default
y_true	Any	1D array-like of true labels.	required
y_proba	Any	Array-like of shape (n_samples, n_classes), predicted probabilities in the same column order as classes.	required
metric	Callable[[Any, Any], float]	Callable (y_true, y_pred) -> float to maximize. Fully caller-supplied — this function ships no default metric.	required
classes	Any	Explicit class label order matching y_proba's columns. Defaults to sorted(np.unique(y_true)).	None
strategy	str | None	"grid" or "nelder-mead". If None (default), auto-selects "grid" for exactly 2 classes and "nelder-mead" for 3+ classes. Explicit "grid" requires exactly two classes.	None
grid_points	int	Number of threshold candidates for the "grid" strategy, evenly spaced over (0, 1) exclusive.	101

Returns:

Type	Description
dict[Any, float]	Dict mapping each class label to its tuned threshold.

Raises:

Type	Description
ValueError	If strategy is unknown or "grid" is requested for anything other than binary classification.

Source code in skyulf-core/skyulf/modeling/_evaluation/thresholds.py

def optimize_thresholds(
    y_true: Any,
    y_proba: Any,
    metric: Callable[[Any, Any], float],
    classes: Any = None,
    strategy: str | None = None,
    grid_points: int = 101,
) -> dict[Any, float]:
    """Search for per-class decision thresholds that maximize ``metric``.

    Args:
        y_true: 1D array-like of true labels.
        y_proba: Array-like of shape (n_samples, n_classes), predicted
            probabilities in the same column order as ``classes``.
        metric: Callable ``(y_true, y_pred) -> float`` to maximize. Fully
            caller-supplied — this function ships no default metric.
        classes: Explicit class label order matching ``y_proba``'s columns.
            Defaults to ``sorted(np.unique(y_true))``.
        strategy: ``"grid"`` or ``"nelder-mead"``. If ``None`` (default),
            auto-selects ``"grid"`` for exactly 2 classes and
            ``"nelder-mead"`` for 3+ classes. Explicit ``"grid"`` requires
            exactly two classes.
        grid_points: Number of threshold candidates for the ``"grid"``
            strategy, evenly spaced over (0, 1) exclusive.

    Returns:
        Dict mapping each class label to its tuned threshold.

    Raises:
        ValueError: If ``strategy`` is unknown or ``"grid"`` is requested
            for anything other than binary classification.
    """
    y_true = np.asarray(y_true)
    y_proba = np.asarray(y_proba, dtype=float)
    classes = _resolve_classes(y_true, classes)

    if strategy is None:
        strategy = "grid" if len(classes) == 2 else "nelder-mead"
    if strategy not in ("grid", "nelder-mead"):
        raise ValueError(f"Unknown strategy {strategy!r}; expected 'grid' or 'nelder-mead'")

    if strategy == "grid":
        if len(classes) != 2:
            raise ValueError(
                "The 'grid' strategy requires binary classification (exactly two classes)."
            )
        return _grid_search_binary(y_true, y_proba, metric, classes, grid_points)
    return _nelder_mead_multiclass(y_true, y_proba, metric, classes)

perform_cross_validation(calculator, applier, X, y, config, n_folds=5, cv_type='k_fold', shuffle=True, random_state=DEFAULT_RANDOM_STATE, time_column=None, progress_callback=None, log_callback=None, preprocessing=None, *, cv_nested_type='auto', group_column=None, gap=0, test_size=None, max_train_size=None, inner_folds=None, sample_weight=None)

Performs K-Fold cross-validation.

Parameters:

Name	Type	Description	Default
sample_weight	Any	Optional positional training weights; held-out metrics stay unweighted.	None
calculator	BaseModelCalculator	The model calculator (fit logic).	required
applier	BaseModelApplier	The model applier (predict logic).	required
X	DataFrame | SkyulfDataFrame	Features.	required
y	Series | Any	Target.	required
config	dict[str, Any]	Model configuration.	required
n_folds	int	Number of folds.	5
cv_type	str	Type of CV.	'k_fold'
shuffle	bool	Whether to shuffle data before splitting (for KFold/Stratified).	True
random_state	int	Random seed for shuffling.	DEFAULT_RANDOM_STATE
time_column	str | None	Optional column name for sorting when using time_series_split.	None
cv_nested_type	str	Explicit nested split policy; auto retains task defaults.	'auto'
group_column	str | None	Split-only entity identifier excluded from model features.	None
gap	int	Number of omitted rows between temporal training and validation.	0
test_size	int | None	Optional row count per temporal validation fold.	None
max_train_size	int | None	Optional rolling training window row limit.	None
inner_folds	int | None	Optional fold count for explicit nested policies.	None
progress_callback	Callable[[int, int], None] | None	Optional callback(current_fold, total_folds).	None
log_callback	Callable[[str], None] | None	Optional callback for logging messages.	None
preprocessing	FoldPreprocessor | None	Optional per-fold preprocessor (F-15). When given, it is re-fit on each fold's training rows and applied to the held-out rows, so preprocessing statistics never see held-out data. None means the caller already transformed the data before splitting.	None

Returns:

Type	Description
dict[str, Any]	Dict containing aggregated metrics and per-fold details.

Source code in skyulf-core/skyulf/modeling/cross_validation.py

def perform_cross_validation(
    calculator: BaseModelCalculator,
    applier: BaseModelApplier,
    X: pd.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    config: dict[str, Any],
    n_folds: int = 5,
    cv_type: str = "k_fold",  # k_fold, stratified_k_fold, time_series_split, shuffle_split, nested_cv
    shuffle: bool = True,
    random_state: int = DEFAULT_RANDOM_STATE,
    time_column: str | None = None,
    progress_callback: Callable[[int, int], None] | None = None,
    log_callback: Callable[[str], None] | None = None,
    preprocessing: FoldPreprocessor | None = None,
    *,
    cv_nested_type: str = "auto",
    group_column: str | None = None,
    gap: int = 0,
    test_size: int | None = None,
    max_train_size: int | None = None,
    inner_folds: int | None = None,
    sample_weight: Any = None,
) -> dict[str, Any]:
    """Performs K-Fold cross-validation.

    Args:
        sample_weight: Optional positional training weights; held-out metrics stay unweighted.
        calculator: The model calculator (fit logic).
        applier: The model applier (predict logic).
        X: Features.
        y: Target.
        config: Model configuration.
        n_folds: Number of folds.
        cv_type: Type of CV.
        shuffle: Whether to shuffle data before splitting (for KFold/Stratified).
        random_state: Random seed for shuffling.
        time_column: Optional column name for sorting when using time_series_split.
        cv_nested_type: Explicit nested split policy; auto retains task defaults.
        group_column: Split-only entity identifier excluded from model features.
        gap: Number of omitted rows between temporal training and validation.
        test_size: Optional row count per temporal validation fold.
        max_train_size: Optional rolling training window row limit.
        inner_folds: Optional fold count for explicit nested policies.
        progress_callback: Optional callback(current_fold, total_folds).
        log_callback: Optional callback for logging messages.
        preprocessing: Optional per-fold preprocessor (F-15). When given, it is
            re-fit on each fold's training rows and applied to the held-out rows,
            so preprocessing statistics never see held-out data. ``None`` means
            the caller already transformed the data before splitting.

    Returns:
        Dict containing aggregated metrics and per-fold details.
    """
    from ._policy_cv import (  # noqa: PLC0415 - avoid tuning import cycle
        perform_policy_cv,
        policy_config,
    )
    from ._tuning.cv_policy import (  # noqa: PLC0415 - avoid tuning import cycle
        uses_explicit_policy,
    )
    from ._tuning.history_policy import has_temporal_history  # noqa: PLC0415 - tuning import cycle

    sample_weight = prepare_weights(sample_weight, len(X), preprocessing)
    y = _indexable_target(y)
    policy = policy_config(
        cv_type,
        n_folds,
        shuffle,
        random_state,
        time_column,
        {
            "cv_nested_type": cv_nested_type,
            "group_column": group_column,
            "gap": gap,
            "test_size": test_size,
            "max_train_size": max_train_size,
            "inner_folds": inner_folds,
        },
    )
    if uses_explicit_policy(policy) or has_temporal_history(preprocessing):
        return perform_policy_cv(
            calculator,
            X,
            y,
            config,
            policy,
            preprocessing,
            log_callback,
            progress_callback,
            sample_weight,
        )
    problem_type = calculator.problem_type

    if log_callback:
        log_callback(f"Starting Cross-Validation (Folds: {n_folds}, Type: {cv_type})")

    # For Time Series Split, sort data chronologically. Only applies to
    # DataFrame-like X (pandas or Polars); a plain array has no columns to
    # sort/drop by, so time_series_split relies on the caller's row order.
    if cv_type == "time_series_split" and hasattr(X, "columns"):
        X, y, positions = _sort_by_time(X, y, time_column, log_callback, logger, True)
        sample_weight = take_weights(sample_weight, positions)

    # Handle nested CV separately
    if cv_type == "nested_cv":
        return _perform_nested_cv(
            calculator=calculator,
            applier=applier,
            X=X,
            y=y,
            config=config,
            n_folds=n_folds,
            shuffle=shuffle,
            random_state=random_state,
            progress_callback=progress_callback,
            log_callback=log_callback,
            preprocessing=preprocessing,
            sample_weight=sample_weight,
        )

    # 1. Setup Splitter (delegates to _build_splitter so unknown cv_type
    # values get the same warning/fallback behavior in both call paths).
    splitter = _build_splitter(
        cv_type=cv_type,
        n_folds=n_folds,
        problem_type=problem_type,
        shuffle=shuffle,
        random_state=random_state,
    )

    fold_results = []

    # Ensure numpy for splitting using the Bridge
    X_arr, y_arr = SklearnBridge.to_sklearn((X, y))

    preflight_weights(sample_weight, splitter, X_arr, y_arr)
    # 2. Iterate Folds
    for fold_idx, (train_idx, val_idx) in enumerate(splitter.split(X_arr, y_arr)):
        fold_results.append(
            _run_cv_fold(
                calculator=calculator,
                X=X,
                y=y,
                train_idx=train_idx,
                val_idx=val_idx,
                config=config,
                problem_type=problem_type,
                fold_idx=fold_idx,
                n_folds=n_folds,
                progress_callback=progress_callback,
                log_callback=log_callback,
                preprocessing=preprocessing,
                sample_weight=sample_weight,
            )
        )

    # 3. Aggregate
    fold_metrics = [cast(dict[str, float], r["metrics"]) for r in fold_results]
    aggregated = _aggregate_metrics(fold_metrics)

    if log_callback:
        log_callback(f"Cross-Validation Completed. Aggregated Metrics: {aggregated}")

    return {
        "aggregated_metrics": aggregated,
        "folds": fold_results,
        "cv_config": {
            "n_folds": n_folds,
            "cv_type": cv_type,
            "shuffle": shuffle,
            "random_state": random_state,
        },
    }