API: modeling.base

skyulf.modeling.base

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

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

@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

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

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]:
        """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

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

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

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

extract_xy(data, target_column)

Extract (X, y) from a DataFrame (pandas or Polars) or an (X, y) tuple, given the target column name.

An empty/falsy target_column is the established "no target" sentinel (see _node_runners.py's target_col="" for data-preview-only inputs): unsupervised calculators (e.g. clustering) rely on this to get the whole frame back as X with y=None.

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

def extract_xy(data: Any, target_column: str) -> tuple[Any, Any]:
    """Extract ``(X, y)`` from a DataFrame (pandas or Polars) or an ``(X, y)``
    tuple, given the target column name.

    An empty/falsy ``target_column`` is the established "no target" sentinel
    (see ``_node_runners.py``'s ``target_col=""`` for data-preview-only
    inputs): unsupervised calculators (e.g. clustering) rely on this to get
    the whole frame back as ``X`` with ``y=None``.
    """
    if not target_column:
        X = data[0] if isinstance(data, tuple) else data
        return X, None

    if isinstance(data, tuple) and len(data) == 2:
        return _extract_xy_from_tuple(data, target_column)

    engine = get_engine(data)

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

    return _extract_xy_pandas_like(data, target_column)