Skip to content

API: modeling

skyulf.modeling

Modeling module for Skyulf.

BaseModelApplier

Bases: ABC

Source code in skyulf-core/skyulf/modeling/base.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
class BaseModelApplier(ABC):
    @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 
165
166
167
168
169
@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 
171
172
173
174
175
176
177
178
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

Source code in skyulf-core/skyulf/modeling/base.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
class BaseModelCalculator(ABC):
    @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 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,
    ) -> Any:
        """Trains the model and returns the fitted model artifact.

        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 
110
111
112
113
114
115
116
117
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) abstractmethod

Trains the model and returns the fitted model artifact.

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 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
@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,
) -> Any:
    """Trains the model and returns the fitted model artifact.

    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 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 
100
101
102
103
104
105
106
107
108
def prepare_tuning_params(self, config: dict[str, Any]) -> None:
    """Hook for structural models (e.g. ensembles) to absorb their
    sub-estimator selection 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 
65
66
class BernoulliNBApplier(SklearnApplier):
    """Bernoulli Naive Bayes Applier."""

BernoulliNBCalculator

Bases: SklearnCalculator

Bernoulli Naive Bayes Calculator.

Source code in skyulf-core/skyulf/modeling/naive_bayes.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@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):
        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 "classification"

CalibratedClassifierApplier

Bases: SklearnApplier

Calibrated Classifier Applier (well-calibrated predict_proba).

Source code in skyulf-core/skyulf/modeling/classification.py 
189
190
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.

Source code in skyulf-core/skyulf/modeling/classification.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
@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.
    """

    # 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]]] = {
        "logistic_regression": lambda: LogisticRegression(max_iter=1000),
        "random_forest": lambda: RandomForestClassifier(
            n_estimators=100, random_state=DEFAULT_RANDOM_STATE
        ),
        "gradient_boosting": lambda: GradientBoostingClassifier(random_state=DEFAULT_RANDOM_STATE),
        "decision_tree": lambda: DecisionTreeClassifier(random_state=DEFAULT_RANDOM_STATE),
        "gaussian_nb": GaussianNB,
        "svc": lambda: SVC(probability=True, random_state=DEFAULT_RANDOM_STATE),
    }

    def __init__(self):
        super().__init__(
            model_class=CalibratedClassifierCV,
            default_params={
                "estimator": LogisticRegression(max_iter=1000),
                "method": "sigmoid",
                "cv": 5,
            },
            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,
    ) -> Any:
        config = self._resolve_base_estimator(config)
        return super().fit(
            X,
            y,
            config,
            progress_callback,
            log_callback,
            validation_data,
            iteration_callback=iteration_callback,
        )

    @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):
            factory = cls.BASE_ESTIMATORS.get(key)
            if factory is None:
                logger.warning(
                    "Unknown base_estimator '%s'; falling back to logistic_regression.", key
                )
                factory = cls.BASE_ESTIMATORS["logistic_regression"]
            bucket["estimator"] = factory()
        if nested:
            resolved["params"] = bucket
            return resolved
        return bucket

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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@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 
27
28
29
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 
31
32
33
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 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@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]:
        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 
134
135
136
137
138
139
140
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 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@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):
        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 
62
63
class LogisticRegressionApplier(SklearnApplier):
    """Logistic Regression Applier."""

LogisticRegressionCalculator

Bases: SklearnCalculator

Logistic Regression Calculator.

Source code in skyulf-core/skyulf/modeling/classification.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@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):
        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,
    ) -> Any:
        self._validate_solver_penalty(config)
        return super().fit(
            X,
            y,
            config,
            progress_callback,
            log_callback,
            validation_data,
            iteration_callback=iteration_callback,
        )

    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 — it's translated to the newer kwargs here,
        right before the sklearn estimator is constructed, so we never pass a
        bare ``penalty=`` to sklearn regardless of installed sklearn version.
        """
        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)

MultinomialNBApplier

Bases: SklearnApplier

Multinomial Naive Bayes Applier.

Source code in skyulf-core/skyulf/modeling/naive_bayes.py 
29
30
class MultinomialNBApplier(SklearnApplier):
    """Multinomial Naive Bayes Applier."""

MultinomialNBCalculator

Bases: SklearnCalculator

Multinomial Naive Bayes Calculator.

Source code in skyulf-core/skyulf/modeling/naive_bayes.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@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):
        super().__init__(
            model_class=MultinomialNB,
            default_params={"alpha": 1.0, "fit_prior": True},
            problem_type="classification",
        )

    @property
    def problem_type(self) -> str:
        return "classification"

RandomForestClassifierApplier

Bases: SklearnApplier

Random Forest Classifier Applier.

Source code in skyulf-core/skyulf/modeling/classification.py 
289
290
class RandomForestClassifierApplier(SklearnApplier):
    """Random Forest Classifier Applier."""

RandomForestClassifierCalculator

Bases: SklearnCalculator

Random Forest Classifier Calculator.

Source code in skyulf-core/skyulf/modeling/classification.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
@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):
        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 
111
112
class RandomForestRegressorApplier(SklearnApplier):
    """Random Forest Regressor Applier."""

RandomForestRegressorCalculator

Bases: SklearnCalculator

Random Forest Regressor Calculator.

Source code in skyulf-core/skyulf/modeling/regression.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@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):
        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 
82
83
class RidgeRegressionApplier(SklearnApplier):
    """Ridge Regression Applier."""

RidgeRegressionCalculator

Bases: SklearnCalculator

Ridge Regression Calculator.

Source code in skyulf-core/skyulf/modeling/regression.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@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):
        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 
705
706
class SGDClassifierApplier(SklearnApplier):
    """Stochastic Gradient Descent Classifier Applier."""

SGDClassifierCalculator

Bases: SklearnCalculator

SGD Classifier Calculator.

Source code in skyulf-core/skyulf/modeling/classification.py 
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
@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):
        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 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
class SklearnApplier(BaseModelApplier):
    """Base applier for Scikit-Learn models."""

    def predict(self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any) -> Any:
        # Convert to Numpy
        X_np, _ = SklearnBridge.to_sklearn(df)

        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 = 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:
        if not hasattr(model_artifact, "predict_proba"):
            return None

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

        # Return as DataFrame
        index = None
        if hasattr(df, "index"):
            index = 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 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
class SklearnCalculator(BaseModelCalculator):
    """Base calculator for Scikit-Learn models."""

    def __init__(
        self,
        model_class: type[BaseEstimator],
        default_params: dict[str, Any],
        problem_type: str,
    ):
        # `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]:
        return self._default_params

    @property
    def problem_type(self) -> str:
        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,
    ) -> Any:
        """Fit the Scikit-Learn model."""
        # 1. Merge Config with Defaults
        params = self._resolve_fit_params(config)

        # A generic <select> UI element always submits its option value as a
        # string, so a "None" option (e.g. "no class weighting") arrives here
        # as the literal string "None", not Python None. Normalize that back
        # to None before anything below decides whether class weighting was
        # actually requested.
        if params.get("class_weight") in ("None", "none", ""):
            params["class_weight"] = None

        # Some estimators (e.g. XGBoost's sklearn wrapper) accept arbitrary
        # **kwargs in their constructor but have no built-in notion of class
        # weighting: a `class_weight` kwarg is silently stored and ignored at
        # fit time (no error — just a native warning). Detect that case up
        # front (by checking whether `class_weight` is an explicitly named
        # constructor parameter, not just swallowed by **kwargs) and, if the
        # value isn't None, translate it into a `sample_weight` array passed
        # to `.fit()` instead, so "balanced"/dict class weighting behaves the
        # same regardless of whether the underlying library supports it
        # natively.
        class_weight_to_apply = None
        if "class_weight" in params and not self._constructor_accepts_class_weight():
            class_weight_to_apply = params.pop("class_weight")

        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))

        sample_weight = None
        if class_weight_to_apply is not None:
            sample_weight = self._compute_sample_weight_for_fit(model, class_weight_to_apply, y_np)

        # 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
        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 _constructor_accepts_class_weight(self) -> bool:
        """True if the wrapped model's constructor explicitly declares a
        `class_weight` parameter (e.g. RandomForestClassifier, LGBMClassifier,
        LogisticRegression) — as opposed to merely accepting arbitrary
        **kwargs (e.g. XGBoost's sklearn wrapper) that silently swallow it."""
        sig = inspect.signature(self.model_class)
        return "class_weight" in sig.parameters

    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."""
        fit_sig = inspect.signature(model.fit)
        if "sample_weight" not in fit_sig.parameters:
            raise ValueError(
                f"{self.model_class.__name__} does not support 'class_weight' natively "
                "and its fit() method does not accept 'sample_weight' either, so "
                "class weighting cannot be applied to this model."
            )
        return compute_sample_weight(class_weight, y_np)

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

Fit the Scikit-Learn model.

Source code in skyulf-core/skyulf/modeling/sklearn_wrapper.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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,
) -> Any:
    """Fit the Scikit-Learn model."""
    # 1. Merge Config with Defaults
    params = self._resolve_fit_params(config)

    # A generic <select> UI element always submits its option value as a
    # string, so a "None" option (e.g. "no class weighting") arrives here
    # as the literal string "None", not Python None. Normalize that back
    # to None before anything below decides whether class weighting was
    # actually requested.
    if params.get("class_weight") in ("None", "none", ""):
        params["class_weight"] = None

    # Some estimators (e.g. XGBoost's sklearn wrapper) accept arbitrary
    # **kwargs in their constructor but have no built-in notion of class
    # weighting: a `class_weight` kwarg is silently stored and ignored at
    # fit time (no error — just a native warning). Detect that case up
    # front (by checking whether `class_weight` is an explicitly named
    # constructor parameter, not just swallowed by **kwargs) and, if the
    # value isn't None, translate it into a `sample_weight` array passed
    # to `.fit()` instead, so "balanced"/dict class weighting behaves the
    # same regardless of whether the underlying library supports it
    # natively.
    class_weight_to_apply = None
    if "class_weight" in params and not self._constructor_accepts_class_weight():
        class_weight_to_apply = params.pop("class_weight")

    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))

    sample_weight = None
    if class_weight_to_apply is not None:
        sample_weight = self._compute_sample_weight_for_fit(model, class_weight_to_apply, y_np)

    # 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
    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 
553
554
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 
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
@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):
        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 
627
628
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 
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
@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):
        super().__init__(
            model_class=StackingRegressor,
            default_params={"cv": 5},
            problem_type="regression",
        )

StatefulEstimator

Source code in skyulf-core/skyulf/modeling/base.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
class StatefulEstimator:
    def __init__(self, calculator: BaseModelCalculator, applier: BaseModelApplier, node_id: str):
        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,
    ) -> dict[str, Any]:
        """
        Performs cross-validation on the training split.
        """
        X_train, y_train = self._extract_xy(dataset.train, target_column)

        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,
        )

    @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):
            X, y_split = split_data
            X = cast(Any, X)
            # If y is None, the target may still be in X — drop it
            if y_split is None and hasattr(X, "columns") and target_column in X.columns:
                X = self._drop_target_column(X, target_column)
            return X

        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,
    ) -> 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.
        """
        # Handle raw DataFrame or Tuple input by wrapping it in a dummy SplitDataset
        dataset = self._normalize_fit_predict_dataset(dataset, target_column, log_callback)

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

        validation_data = None
        if dataset.validation is not None:
            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,
            )
        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,
            )

        # 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: pandas uses `.empty`, Polars uses `.is_empty()`
        is_test_empty = test_df.empty if hasattr(test_df, "empty") else test_df.is_empty()

        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 dataset.validation is not None:
            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,
                )

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

    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``
        and returning 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

        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)

    @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

    @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
        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 = "",
    ) -> 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
            )
        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"
            )

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)

Performs cross-validation on the training split.

Source code in skyulf-core/skyulf/modeling/base.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
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,
) -> dict[str, Any]:
    """
    Performs cross-validation on the training split.
    """
    X_train, y_train = self._extract_xy(dataset.train, target_column)

    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,
    )

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 
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
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,
            )

    # 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)

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.

Source code in skyulf-core/skyulf/modeling/base.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
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,
) -> 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.
    """
    # Handle raw DataFrame or Tuple input by wrapping it in a dummy SplitDataset
    dataset = self._normalize_fit_predict_dataset(dataset, target_column, log_callback)

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

    validation_data = None
    if dataset.validation is not None:
        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,
        )
    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,
        )

    # 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: pandas uses `.empty`, Polars uses `.is_empty()`
    is_test_empty = test_df.empty if hasattr(test_df, "empty") else test_df.is_empty()

    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 dataset.validation is not None:
        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 
516
517
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 
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
@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):
        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 
592
593
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 
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
@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):
        super().__init__(
            model_class=VotingRegressor,
            default_params={},
            problem_type="regression",
        )

apply_thresholds(y_proba, thresholds, classes=None)

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

Binary (thresholds is a single float, or a one-entry dict): predicts the positive (second) class when y_proba[:, 1] >= threshold, else the first class.

Multiclass (thresholds is a dict covering every class): scaled argmax — classes[argmax(y_proba / thresholds, axis=1)]. Equal thresholds across all classes reduce to plain argmax.

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

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 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def apply_thresholds(
    y_proba: Any,
    thresholds: dict[Any, float] | float,
    classes: Any = None,
) -> np.ndarray:
    """Convert predicted probabilities into class predictions using per-class thresholds.

    Binary (``thresholds`` is a single float, or a one-entry dict): predicts
    the positive (second) class when ``y_proba[:, 1] >= threshold``, else the
    first class.

    Multiclass (``thresholds`` is a dict covering every class): scaled
    argmax — ``classes[argmax(y_proba / thresholds, axis=1)]``. Equal
    thresholds across all classes reduce to plain argmax.

    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).

    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 = np.asarray(y_proba, dtype=float)
    if y_proba.ndim != 2:
        raise ValueError(f"y_proba must be 2D (n_samples, n_classes); got shape {y_proba.shape}")

    n_classes = y_proba.shape[1]
    if classes is None:
        classes = np.arange(n_classes)
    classes = np.asarray(classes)
    if len(classes) != n_classes:
        raise ValueError(f"classes has {len(classes)} entries but y_proba has {n_classes} columns")

    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:
        (threshold,) = thresholds.values()
        threshold = float(threshold)
        return np.where(y_proba[:, 1] >= threshold, classes[1], classes[0])

    missing = [c for c in classes if c not in thresholds]
    if missing:
        raise ValueError(
            f"thresholds is missing entries for classes: {missing}. "
            "apply_thresholds() requires a threshold for every class."
        )

    thresholds_array = np.array([float(thresholds[c]) for c in 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)

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.

Source code in skyulf-core/skyulf/modeling/_evaluation/metrics.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def calculate_classification_metrics(
    model: Any,
    X: pd.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    *,
    X_np: Any = None,
    y_np: Any = None,
    predictions: Any = None,
    proba: Any = 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.
    """

    # 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))

    # 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)

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

    _add_probability_based_metrics(metrics, model, X_np, y_arr, proba=proba)

    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 
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
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))
    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")

    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 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
def calculate_regression_metrics(
    model: Any,
    X: pd.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))

    # 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: a global summary plus 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 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
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: a global summary plus
    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)
        try:
            explanation = explainer(sample)
        except Exception as exc:
            # Only catch additivity failures here — everything else
            # propagates to the outer handler. When `_shap_exceptions` is
            # available we check via `isinstance` against `ExplainerError`;
            # otherwise we fall back to a message-based heuristic so
            # explainability isn't silently disabled on SHAP versions that
            # don't expose the private `_exceptions` module.
            _is_additivity = (
                _shap_exceptions is not None and isinstance(exc, _shap_exceptions.ExplainerError)
            ) or (_shap_exceptions is None and "additivity" in str(exc).lower())
            if not _is_additivity or not is_exact_tree:
                raise
            # `tree_path_dependent` computes Shapley values directly from
            # each tree's own path/sample-weight structure with no
            # background approximation involved, so this additivity
            # mismatch can only be a floating-point tolerance artefact in
            # `shap`'s own re-check (a known upstream issue for some
            # scikit-learn/shap version combinations), not evidence that we
            # fed the explainer inconsistent data. Retry without the
            # (redundant, in this case) re-verification rather than
            # dropping the explanation entirely.
            logger.warning(
                "SHAP additivity re-check failed for model_type=%s despite using the "
                "exact tree_path_dependent algorithm; retrying with check_additivity=False "
                "(see https://github.com/shap/shap/issues/2777)",
                type(model).__name__,
            )
            explanation = explainer(sample, check_additivity=False)
        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: list[dict[str, Any]] = []
        if resolved is not None:
            shap_rows, base_values = resolved
            display_sample = sample.iloc[:max_display_samples]
            for i in range(len(display_sample)):
                row = display_sample.iloc[i]
                samples.append(
                    {
                        "base_value": round(float(base_values[i]), 6),
                        "feature_values": {
                            name: round(float(row[name]), 6) for name in feature_names
                        },
                        "shap_values": {
                            name: round(float(shap_rows[i, j]), 6)
                            for j, name in enumerate(feature_names)
                        },
                    }
                )

        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 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.

Source code in skyulf-core/skyulf/modeling/hyperparameters/_registry.py 
503
504
505
506
507
508
509
510
511
512
513
def get_default_search_space(model_key: str, strategy: str = "random") -> dict[str, Any]:
    """Return 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``.
    """
    if strategy in _GRID_STRATEGIES:
        return GRID_SEARCH_SPACES.get(model_key, DEFAULT_SEARCH_SPACES.get(model_key, {}))
    return DEFAULT_SEARCH_SPACES.get(model_key, {})

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.

 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 not one of "grid"/"nelder-mead"/None.
Source code in skyulf-core/skyulf/modeling/_evaluation/thresholds.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
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.
        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 not one of ``"grid"``/``"nelder-mead"``/``None``.
    """
    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":
        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)

Performs K-Fold cross-validation.

Parameters:

Name Type Description Default
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
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 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
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,
) -> dict[str, Any]:
    """
    Performs K-Fold cross-validation.

    Args:
        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.
        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.
    """
    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 = _sort_by_time(X, y, time_column, log_callback, logger)

    # 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,
        )

    # 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))

    # 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,
            )
        )

    # 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,
        },
    }