API: modeling

skyulf.modeling

Modeling module for Skyulf.

BaseModelApplier

Bases: ABC

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

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

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

predict_proba(df, model_artifact)

Generates prediction probabilities if supported. Returns DataFrame where columns are classes.

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

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

BaseModelCalculator

Bases: ABC

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

class BaseModelCalculator(ABC):
    @property
    @abstractmethod
    def problem_type(self) -> str:
        """Returns 'classification' or 'regression'."""

    @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 {}

    @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,
    ) -> 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' or 'regression'.

build_tuning_search_space(config, strategy)

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

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

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

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

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

fit(X, y, config, progress_callback=None, log_callback=None, validation_data=None) 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

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

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

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

BernoulliNBCalculator

Bases: SklearnCalculator

Bernoulli Naive Bayes Calculator.

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

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

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

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

    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,
    ) -> Any:
        config = self._resolve_base_estimator(config)
        return super().fit(X, y, config, progress_callback, log_callback, validation_data)

    @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

HyperparameterField dataclass

Describe a single tunable hyperparameter.

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

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

    name: str
    label: str
    type: str  # "number", "select", "boolean"
    default: Any
    description: str = ""
    min: float | None = None
    max: float | None = None
    step: float | None = None
    options: list[dict[str, Any]] | None = (
        None  # For 'select' type: [{"label": "L1", "value": "l1"}]
    )

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)

LogisticRegressionApplier

Bases: SklearnApplier

Logistic Regression Applier.

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

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

LogisticRegressionCalculator

Bases: SklearnCalculator

Logistic Regression Calculator.

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

@NodeRegistry.register("logistic_regression", LogisticRegressionApplier)
@node_meta(
    id="logistic_regression",
    name="Logistic Regression",
    category="Modeling",
    description="Linear model for classification.",
    params={"max_iter": 1000, "solver": "lbfgs", "random_state": 42},
    tags=["requires_scaling"],
)
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: 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",
                "random_state": 42,
            },
            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,
    ) -> Any:
        self._validate_solver_penalty(config)
        return super().fit(X, y, config, progress_callback, log_callback, validation_data)

    @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'}."
        )

    @classmethod
    def _validate_solver_penalty(cls, 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.
        """
        params = cls._extract_solver_penalty_params(config)
        if params is None:
            return
        solver = params.get("solver")
        penalty = params.get("penalty")
        if solver is None or "penalty" not in params:
            return
        compatible = cls._SOLVER_PENALTIES.get(solver)
        if compatible is not None and penalty not in compatible:
            cls._raise_incompatible_solver_penalty(solver, penalty)

MultinomialNBApplier

Bases: SklearnApplier

Multinomial Naive Bayes Applier.

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

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

MultinomialNBCalculator

Bases: SklearnCalculator

Multinomial Naive Bayes Calculator.

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

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

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

RandomForestClassifierCalculator

Bases: SklearnCalculator

Random Forest Classifier Calculator.

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

@NodeRegistry.register("random_forest_classifier", RandomForestClassifierApplier)
@node_meta(
    id="random_forest_classifier",
    name="Random Forest Classifier",
    category="Modeling",
    description="Ensemble of decision trees.",
    params={"n_estimators": 50, "max_depth": 10, "min_samples_split": 5},
)
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,
                "random_state": 42,
            },
            problem_type="classification",
        )

RandomForestRegressorApplier

Bases: SklearnApplier

Random Forest Regressor Applier.

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

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

RandomForestRegressorCalculator

Bases: SklearnCalculator

Random Forest Regressor Calculator.

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

@NodeRegistry.register("random_forest_regressor", RandomForestRegressorApplier)
@node_meta(
    id="random_forest_regressor",
    name="Random Forest Regressor",
    category="Modeling",
    description="Ensemble of decision trees for regression.",
    params={"n_estimators": 50, "max_depth": 10, "min_samples_split": 5},
)
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,
                "random_state": 42,
            },
            problem_type="regression",
        )

RidgeRegressionApplier

Bases: SklearnApplier

Ridge Regression Applier.

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

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

RidgeRegressionCalculator

Bases: SklearnCalculator

Ridge Regression Calculator.

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

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

    def __init__(self):
        super().__init__(
            model_class=Ridge,
            default_params={
                "alpha": 1.0,
                "solver": "auto",
                "random_state": 42,
            },
            problem_type="regression",
        )

SGDClassifierApplier

Bases: SklearnApplier

Stochastic Gradient Descent Classifier Applier.

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

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

SGDClassifierCalculator

Bases: SklearnCalculator

SGD Classifier Calculator.

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

@NodeRegistry.register("sgd_classifier", SGDClassifierApplier)
@node_meta(
    id="sgd_classifier",
    name="SGD Classifier (text / linear)",
    category="Modeling",
    description=(
        "Linear classifiers (SVM, logistic regression, etc.) with SGD training. "
        "Highly efficient for high-dimensional sparse/dense text representations "
        "and large datasets."
    ),
    params={
        "loss": "log_loss",
        "penalty": "l2",
        "alpha": 0.0001,
        "l1_ratio": 0.15,
        "max_iter": 1000,
        "random_state": 42,
    },
    tags=["text", "nlp", "classification", "linear", "requires_scaling"],
)
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,
                "random_state": 42,
            },
            problem_type="classification",
        )

SklearnApplier

Bases: BaseModelApplier

Base applier for Scikit-Learn models.

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

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

    def predict(self, df: pd.DataFrame | SkyulfDataFrame, model_artifact: Any) -> Any:
        # 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 = [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,
    ):
        # `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,
    ) -> Any:
        """Fit the Scikit-Learn model."""
        # 1. Merge Config with Defaults
        params = self._resolve_fit_params(config)

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

        model.fit(X_np, y_np)

        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 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 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.
        """
        import inspect

        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

fit(X, y, config, progress_callback=None, log_callback=None, validation_data=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,
) -> Any:
    """Fit the Scikit-Learn model."""
    # 1. Merge Config with Defaults
    params = self._resolve_fit_params(config)

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

    model.fit(X_np, y_np)

    return model

StackingClassifierApplier

Bases: SklearnApplier

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

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

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

StackingClassifierCalculator

Bases: _BaseEnsembleCalculator

Stacking Classifier Calculator with selectable base + final learners.

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

@NodeRegistry.register("stacking_classifier", StackingClassifierApplier)
@node_meta(
    id="stacking_classifier",
    name="Stacking Classifier",
    category="Ensemble",
    description=(
        "Trains a final classifier on the out-of-fold predictions of several "
        "base classifiers. Uses internal CV folds to avoid leakage."
    ),
    params={
        "base_estimators": ["random_forest", "gradient_boosting", "svc"],
        "final_estimator": "logistic_regression",
        "cv": 5,
    },
    tags=["requires_scaling"],
)
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

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

StackingRegressorCalculator

Bases: _BaseEnsembleCalculator

Stacking Regressor Calculator with selectable base + final learners.

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

@NodeRegistry.register("stacking_regressor", StackingRegressorApplier)
@node_meta(
    id="stacking_regressor",
    name="Stacking Regressor",
    category="Ensemble",
    description=(
        "Trains a final regressor on the out-of-fold predictions of several "
        "base regressors. Uses internal CV folds to avoid leakage."
    ),
    params={
        "base_estimators": ["random_forest", "gradient_boosting", "ridge"],
        "final_estimator": "ridge",
        "cv": 5,
    },
    tags=["requires_scaling"],
)
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

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]:
        """Helper to extract X and y from DataFrame or Tuple."""
        if isinstance(data, tuple) and len(data) == 2:
            return self._extract_xy_from_tuple(data, target_column)

        engine = get_engine(data)

        if engine.name == EngineName.POLARS:
            return self._extract_xy_polars(data, target_column)

        return self._extract_xy_pandas_like(data, target_column)

    def _extract_xy_from_tuple(self, data: tuple[Any, Any], target_column: str) -> tuple[Any, Any]:
        """Extracts X/y from a ``(X, y)`` tuple, pulling ``y`` out of ``X`` if it's missing."""
        X, y = data[0], data[1]
        # If y is None but X is a DataFrame containing the target, extract it
        if y is None and hasattr(X, "columns") and target_column in X.columns:
            return self._extract_xy(X, target_column)
        return X, y

    @staticmethod
    def _extract_xy_polars(data: Any, target_column: str) -> tuple[Any, Any]:
        """Extracts X/y from a Polars DataFrame by dropping/selecting ``target_column``."""
        if target_column not in data.columns:
            raise ValueError(f"Target column '{target_column}' not found in data")
        X = data.drop([target_column])
        y = data.select(target_column).to_series()
        return X, y

    @staticmethod
    def _extract_xy_pandas_like(data: Any, target_column: str) -> tuple[Any, Any]:
        """Extracts X/y from a pandas or generic DataFrame-like object."""
        # Check for DataFrame-like
        if hasattr(data, "columns"):
            if target_column not in data.columns:
                raise ValueError(f"Target column '{target_column}' not found in data")

            # Fallback for pure Pandas or Generic DataFrame
            # If we reached here without matching Polars explicitly, treat as generic/pandas
            # Try generic drop if available
            if hasattr(data, "drop"):
                # Handle pandas-like drop
                try:
                    return data.drop(columns=[target_column]), data[target_column]
                except TypeError:
                    # Maybe it doesn't support columns= kwarg, try position or list
                    pass

            # Simple attribute access fallback
            if hasattr(data, target_column):
                return data, getattr(data, target_column)

        raise ValueError(f"Unexpected data type: {type(data)}")

    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,
    ) -> dict[str, Any]:
        """
        Performs cross-validation on the training split.
        """
        # Import here to avoid circular dependency if any
        from .cross_validation import perform_cross_validation

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

    @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
        | 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."""
        if isinstance(dataset, pd.DataFrame):
            return SplitDataset(train=dataset, test=pd.DataFrame(), validation=None)

        if isinstance(dataset, tuple):
            # Check if it's (train_df, test_df) or (X, y)
            elem0 = dataset[0]
            if isinstance(elem0, pd.DataFrame) and target_column in elem0.columns:
                # It's (train_df, test_df)
                train_df, test_df = dataset
                return SplitDataset(train=train_df, test=test_df, validation=None)  # type: ignore

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

        return dataset

    def fit_predict(
        self,
        dataset: SplitDataset
        | pd.DataFrame
        | 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",
    ) -> dict[str, pd.Series]:
        """
        Fits the model on training data and returns predictions for all splits.
        """
        # 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
        self.model = self.calculator.fit(
            X_train,
            y_train,
            config,
            progress_callback=progress_callback,
            log_callback=log_callback,
            validation_data=validation_data,
        )

        # 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 refit(
        self,
        dataset: SplitDataset,
        target_column: str,
        config: dict[str, Any],
        job_id: str = "unknown",
    ) -> None:
        """
        Refits the model on Train + Validation data and updates the artifact.
        """
        if dataset.validation is None:
            # Fallback to normal fit if no validation set
            self.fit_predict(dataset, target_column, config, job_id=job_id)
            return

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

        if get_engine(X_train).name == EngineName.POLARS:
            import polars as pl

            X_combined = pl.concat([X_train, X_val])
            y_combined = pl.concat([y_train, y_val])
        else:
            X_combined = pd.concat([X_train, X_val], axis=0)
            y_combined = pd.concat([y_train, y_val], axis=0)

        # 2. Train Model
        self.model = self.calculator.fit(X_combined, y_combined, config)

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

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

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

        # 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],
    ) -> 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 or y is None:
            return None

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

        model_to_evaluate = self._unwrap_tuned_model()
        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

    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.values.tolist(),
        }

    @staticmethod
    def _evaluate_split_with_model(
        model_to_evaluate: Any, split_name: str, X: Any, y: Any, problem_type: str
    ) -> Any:
        """Dispatches to the classification or regression evaluator based on ``problem_type``."""
        # Import here to avoid circular dependency
        from ._evaluation.classification import evaluate_classification_model
        from ._evaluation.regression import evaluate_regression_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
            )
        else:
            raise ValueError(f"Unknown problem type: {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)

Performs cross-validation on the training split.

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

def cross_validate(
    self,
    dataset: SplitDataset,
    target_column: str,
    config: dict[str, Any],
    n_folds: int = 5,
    cv_type: str = "k_fold",
    shuffle: bool = True,
    random_state: int = 42,
    time_column: str | None = None,
    progress_callback: Callable[[int, int], None] | None = None,
    log_callback: Callable[[str], None] | None = None,
) -> dict[str, Any]:
    """
    Performs cross-validation on the training split.
    """
    # Import here to avoid circular dependency if any
    from .cross_validation import perform_cross_validation

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

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

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

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

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

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

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

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

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

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

def fit_predict(
    self,
    dataset: SplitDataset
    | pd.DataFrame
    | 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",
) -> dict[str, pd.Series]:
    """
    Fits the model on training data and returns predictions for all splits.
    """
    # 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
    self.model = self.calculator.fit(
        X_train,
        y_train,
        config,
        progress_callback=progress_callback,
        log_callback=log_callback,
        validation_data=validation_data,
    )

    # 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

refit(dataset, target_column, config, job_id='unknown')

Refits the model on Train + Validation data and updates the artifact.

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

def refit(
    self,
    dataset: SplitDataset,
    target_column: str,
    config: dict[str, Any],
    job_id: str = "unknown",
) -> None:
    """
    Refits the model on Train + Validation data and updates the artifact.
    """
    if dataset.validation is None:
        # Fallback to normal fit if no validation set
        self.fit_predict(dataset, target_column, config, job_id=job_id)
        return

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

    if get_engine(X_train).name == EngineName.POLARS:
        import polars as pl

        X_combined = pl.concat([X_train, X_val])
        y_combined = pl.concat([y_train, y_val])
    else:
        X_combined = pd.concat([X_train, X_val], axis=0)
        y_combined = pd.concat([y_train, y_val], axis=0)

    # 2. Train Model
    self.model = self.calculator.fit(X_combined, y_combined, config)

VotingClassifierApplier

Bases: SklearnApplier

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

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

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

VotingClassifierCalculator

Bases: _BaseEnsembleCalculator

Voting Classifier Calculator with selectable base learners.

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

@NodeRegistry.register("voting_classifier", VotingClassifierApplier)
@node_meta(
    id="voting_classifier",
    name="Voting Classifier",
    category="Ensemble",
    description=(
        "Combines several classifiers by majority vote (hard) or averaged "
        "probabilities (soft). Fits each base model once; no internal CV."
    ),
    params={
        "base_estimators": ["random_forest", "logistic_regression", "gradient_boosting"],
        "voting": "soft",
    },
    tags=["requires_scaling"],
)
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

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

VotingRegressorCalculator

Bases: _BaseEnsembleCalculator

Voting Regressor Calculator with selectable base learners.

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

@NodeRegistry.register("voting_regressor", VotingRegressorApplier)
@node_meta(
    id="voting_regressor",
    name="Voting Regressor",
    category="Ensemble",
    description=(
        "Averages the predictions of several regressors (optionally weighted). "
        "Fits each base model once; no internal CV."
    ),
    params={
        "base_estimators": ["linear_regression", "random_forest", "gradient_boosting"],
    },
    tags=["requires_scaling"],
)
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",
        )

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

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

perform_cross_validation(calculator, applier, X, y, config, n_folds=5, cv_type='k_fold', shuffle=True, random_state=42, time_column=None, progress_callback=None, log_callback=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.	42
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

Returns:

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

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

def perform_cross_validation(
    calculator: "BaseModelCalculator",
    applier: "BaseModelApplier",
    X: pd.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    config: dict[str, Any],
    n_folds: int = 5,
    cv_type: str = "k_fold",  # k_fold, stratified_k_fold, time_series_split, shuffle_split, nested_cv
    shuffle: bool = True,
    random_state: int = 42,
    time_column: str | None = None,
    progress_callback: Callable[[int, int], None] | None = None,
    log_callback: Callable[[str], None] | 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.

    Returns:
        Dict containing aggregated metrics and per-fold details.
    """
    import logging

    logger = logging.getLogger(__name__)
    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,
        )

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

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