API: modeling.sklearn_wrapper

skyulf.modeling.sklearn_wrapper

Wrapper for Scikit-Learn models.

SklearnApplier

Bases: BaseModelApplier

Base applier for Scikit-Learn models.

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

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

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

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

        preds = model_artifact.predict(X_np)

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

        return pd.Series(preds, index=index)

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

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

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

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

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

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

predict(df, model_artifact)

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

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

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

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

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

    preds = model_artifact.predict(X_np)

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

    return pd.Series(preds, index=index)

predict_proba(df, model_artifact)

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

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

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

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

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

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

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

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

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

SklearnCalculator

Bases: BaseModelCalculator

Base calculator for Scikit-Learn models.

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

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

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

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

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

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

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

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

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

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

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

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

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

        return model

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

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

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

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

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

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

        if overrides:
            params.update(overrides)

        return self._inject_default_seed(params)

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

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

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

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

        if accepts_kwargs:
            return params

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

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

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

default_params property

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

problem_type property

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

__init__(model_class, default_params, problem_type)

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

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

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

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

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

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

Fit the Scikit-Learn model.

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

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

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

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

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

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

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

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

    return model