API: preprocessing.split

skyulf.preprocessing.split

Train/test/val splitter and feature/target splitter nodes.

These nodes return a :class:SplitDataset (not the canonical (X, y) shape), so they cannot use :func:apply_dual_engine — the dispatcher expects appliers that return (X_out, y_out). Instead, we centralise the engine handling in two small helpers:

• :func:_to_pandas_remember_engine converts to pandas and records whether a conversion happened.
• :func:_back_to_engine converts the result back to polars when needed.

This removes the inline if engine.name == EngineName.POLARS branches from both :class:DataSplitter methods and :class:FeatureTargetSplitApplier.

DataSplitter

Split a DataFrame (or X/y pair) into Train, Test, and optional Validation.

Source code in skyulf-core/skyulf/preprocessing/split.py

class DataSplitter:
    """Split a DataFrame (or X/y pair) into Train, Test, and optional Validation."""

    def __init__(
        self,
        test_size: float = 0.2,
        validation_size: float = 0.0,
        random_state: int = DEFAULT_RANDOM_STATE,
        shuffle: bool = True,
        stratify_col: str | None = None,
    ):
        """Validate and store the split proportions and stratification column.

        ``stratify_col`` names the column to balance class proportions on;
        ``None`` disables stratification. ``random_state`` and ``shuffle`` are
        reused for both the test carve and the validation carve, so the
        partitioning is reproducible as a whole.

        Raises:
            ValueError: If ``test_size`` is not strictly between 0 and 1, if
                ``validation_size`` is outside ``[0, 1)``, or if the two
                together reach 1 and leave no rows for training.
        """
        if not 0 < test_size < 1:
            raise ValueError(f"test_size must be between 0 and 1 (exclusive), got {test_size!r}.")
        if not 0 <= validation_size < 1:
            raise ValueError(
                f"validation_size must be between 0 (inclusive) and 1 (exclusive), "
                f"got {validation_size!r}."
            )
        if test_size + validation_size >= 1:
            raise ValueError(
                f"test_size ({test_size!r}) + validation_size ({validation_size!r}) must be "
                f"less than 1, otherwise there is no data left for training."
            )
        self.test_size = test_size
        self.validation_size = validation_size
        self.random_state = random_state
        self.shuffle = shuffle
        self.stratify_col = stratify_col

    # ---- public API ---------------------------------------------------------

    def split_indices(self, n: int, stratify: Any = None, *, groups: Any = None) -> tuple[Any, Any]:
        """Split row positions for an initial train/test or calibration partition.

        Callers gather features, targets and weights with these same positions.
        Optional groups remain whole; group splits require shuffle and cannot
        simultaneously stratify. This method does not create a third validation
        partition; ``split`` and ``split_xy`` own that optional second split.
        """
        if groups is not None:
            if stratify is not None or not self.shuffle:
                raise ValueError("Group splitting requires shuffle=True and stratify=None.")
            check_consistent_length(np.arange(n), groups)
            if pd.isna(np.asarray(groups)).any():
                raise ValueError("Group metadata must be nonnull.")
            splitter = GroupShuffleSplit(
                n_splits=1, test_size=self.test_size, random_state=self.random_state
            )
            return next(splitter.split(np.arange(n), groups=groups))
        return self._split_indices(n, stratify)

    def split_xy(
        self,
        X: pd.DataFrame | pl.DataFrame | SkyulfDataFrame,
        y: pd.Series | Any,
        *,
        sample_weight: Any = None,
    ) -> SplitDataset:
        """Split an ``(X, y)`` pair into train/test and optional validation.

        Polars input is partitioned by index gather and never converted; other
        engines round-trip through pandas and are converted back, so the result
        stays on the input engine. Requested stratification is downgraded to an
        unstratified split, with a warning, when the rarest class has fewer than
        two members.
        """
        check_consistent_length(X, y)
        if sample_weight is not None:
            return self._split_weighted(X, y, sample_weight, paired=True)
        if is_polars(X):
            return self._split_xy_polars(cast(Any, X), y)

        X_pd, was_polars = _to_pandas_remember_engine(X)
        y_pd, _ = _to_pandas_remember_engine(y)

        stratify = _safe_stratify(y_pd, "Stratified split") if self.stratify_col else None

        X_tv, X_test, y_tv, y_test = train_test_split(
            X_pd,
            y_pd,
            test_size=self.test_size,
            random_state=self.random_state,
            shuffle=self.shuffle,
            stratify=stratify,
        )

        validation, X_train, y_train = self._maybe_split_validation_xy(X_tv, y_tv)

        train = (_back_to_engine(X_train, was_polars), _back_to_engine(y_train, was_polars))
        test = (_back_to_engine(X_test, was_polars), _back_to_engine(y_test, was_polars))
        if validation is not None:
            validation = (
                _back_to_engine(validation[0], was_polars),
                _back_to_engine(validation[1], was_polars),
            )
        return SplitDataset(train=train, test=test, validation=validation)

    def split(
        self, df: pd.DataFrame | pl.DataFrame | SkyulfDataFrame, *, sample_weight: Any = None
    ) -> SplitDataset:
        """Split a whole frame row-wise, leaving any target column in place.

        Shares :meth:`split_xy`'s engine handling. Stratification here reads
        ``stratify_col`` out of the frame itself, so a configured column that is
        absent from ``df`` disables stratification with a warning instead of
        raising.
        """
        if sample_weight is not None:
            return self._split_weighted(df, None, sample_weight, paired=False)
        if is_polars(df):
            return self._split_polars(cast(Any, df))

        df_pd, was_polars = _to_pandas_remember_engine(df)
        stratify = self._frame_stratify(df_pd, label="Stratified split")

        train_val, test = train_test_split(
            df_pd,
            test_size=self.test_size,
            random_state=self.random_state,
            shuffle=self.shuffle,
            stratify=stratify,
        )

        validation, train = self._maybe_split_validation_frame(train_val)
        return SplitDataset(
            train=_back_to_engine(train, was_polars),
            test=_back_to_engine(test, was_polars),
            validation=_back_to_engine(validation, was_polars),
        )

    # ---- polars-native paths (index split + gather, no frame conversion) ----

    def _split_weighted(self, X: Any, y: Any, weights: Any, *, paired: bool) -> SplitDataset:
        """Partition features, targets and weights with the same actual row positions."""
        check_consistent_length(X, y)
        weights = validate_sample_weight(weights, len(X))
        assert weights is not None
        labels = y if paired else X[self.stratify_col] if self.stratify_col in X.columns else None
        stratify = self._target_stratify(labels, "Stratified split")
        train_idx, test_idx = self._split_indices(len(X), stratify)
        val_idx = None
        if self.validation_size > 0:
            val_labels = _take_weighted_rows(labels, train_idx)
            train_idx, val_idx = train_test_split(
                train_idx,
                test_size=self.validation_size / (1 - self.test_size),
                random_state=self.random_state,
                shuffle=self.shuffle,
                stratify=self._target_stratify(val_labels, "Stratified validation split"),
            )

        def payload(indices: Any) -> Any:
            """Gather one partition without relying on pandas index labels."""
            if indices is None:
                return None
            features = _take_weighted_rows(X, indices)
            return (features, _take_weighted_rows(y, indices)) if paired else features

        return SplitDataset(
            train=payload(train_idx),
            test=payload(test_idx),
            validation=payload(val_idx),
            train_sample_weight=validate_sample_weight(weights[train_idx], len(train_idx)),
        )

    def _target_stratify(self, labels: Any, label: str) -> Any:
        """Retain the rare-class fallback for each supported target container."""
        if self.stratify_col is None or labels is None:
            return None
        if is_polars(labels):
            return _safe_stratify_polars(labels, label)
        return _safe_stratify(labels, label)

    def _split_indices(self, n: int, stratify: Any) -> tuple[Any, Any]:
        """Split row positions ``0..n-1``; same partitioning as splitting rows."""
        return train_test_split(
            np.arange(n),
            test_size=self.test_size,
            random_state=self.random_state,
            shuffle=self.shuffle,
            stratify=stratify,
        )

    def _split_xy_polars(self, X: Any, y: Any) -> SplitDataset:
        stratify = self._target_stratify(y, "Stratified split")

        tv_idx, test_idx = self._split_indices(X.height, stratify)

        validation = None
        train_idx = tv_idx
        if self.validation_size > 0:
            relative_val_size = self.validation_size / (1 - self.test_size)
            stratify_val = self._target_stratify(
                _take_weighted_rows(y, tv_idx), "Stratified validation split"
            )
            train_idx, val_idx = train_test_split(
                tv_idx,
                test_size=relative_val_size,
                random_state=self.random_state,
                shuffle=self.shuffle,
                stratify=stratify_val,
            )
            validation = (X.gather(val_idx), _take_weighted_rows(y, val_idx))

        return SplitDataset(
            train=(X.gather(train_idx), _take_weighted_rows(y, train_idx)),
            test=(X.gather(test_idx), _take_weighted_rows(y, test_idx)),
            validation=validation,
        )

    def _split_polars(self, df: Any) -> SplitDataset:
        stratify = self._frame_stratify_polars(df, label="Stratified split")

        tv_idx, test_idx = self._split_indices(df.height, stratify)

        validation = None
        train_idx = tv_idx
        if self.validation_size > 0:
            relative_val_size = self.validation_size / (1 - self.test_size)
            stratify_val = self._frame_stratify_polars(
                df.gather(tv_idx), label="Stratified validation split"
            )
            train_idx, val_idx = train_test_split(
                tv_idx,
                test_size=relative_val_size,
                random_state=self.random_state,
                shuffle=self.shuffle,
                stratify=stratify_val,
            )
            validation = df.gather(val_idx)

        return SplitDataset(
            train=df.gather(train_idx),
            test=df.gather(test_idx),
            validation=validation,
        )

    def _frame_stratify_polars(self, df: Any, label: str) -> Any:
        """Polars counterpart of :meth:`_frame_stratify`."""
        if not (self.stratify_col and self.stratify_col in df.columns):
            if self.stratify_col:
                logger.warning(
                    "%s requested but no target_column is configured for this "
                    "plain-DataFrame input, so there is no column to stratify on. "
                    "Stratification will be disabled.",
                    label,
                )
            return None
        return _safe_stratify_polars(df.get_column(self.stratify_col), label)

    # ---- private helpers ----------------------------------------------------

    def _frame_stratify(self, df_pd: Any, label: str) -> Any:
        """Pick + sanity-check the stratify column on a frame split."""
        if not (self.stratify_col and self.stratify_col in df_pd.columns):
            if self.stratify_col:
                logger.warning(
                    "%s requested but no target_column is configured for this "
                    "plain-DataFrame input, so there is no column to stratify on. "
                    "Stratification will be disabled.",
                    label,
                )
            return None
        return _safe_stratify(df_pd[self.stratify_col], label)

    def _maybe_split_validation_xy(self, X_tv: Any, y_tv: Any) -> tuple[Any, Any, Any]:
        """Carve a validation set off of (X_tv, y_tv); returns (val, X_train, y_train)."""
        if self.validation_size <= 0:
            return None, X_tv, y_tv

        relative_val_size = self.validation_size / (1 - self.test_size)
        stratify_val = (
            _safe_stratify(y_tv, "Stratified validation split") if self.stratify_col else None
        )
        X_train, X_val, y_train, y_val = train_test_split(
            X_tv,
            y_tv,
            test_size=relative_val_size,
            random_state=self.random_state,
            shuffle=self.shuffle,
            stratify=stratify_val,
        )
        return (X_val, y_val), X_train, y_train

    def _maybe_split_validation_frame(self, train_val: Any) -> tuple[Any, Any]:
        """Carve a validation set off of ``train_val`` (frame mode); returns (val, train)."""
        if self.validation_size <= 0:
            return None, train_val

        relative_val_size = self.validation_size / (1 - self.test_size)
        stratify_val = self._frame_stratify(train_val, label="Stratified validation split")
        train, val = train_test_split(
            train_val,
            test_size=relative_val_size,
            random_state=self.random_state,
            shuffle=self.shuffle,
            stratify=stratify_val,
        )
        return val, train

__init__(test_size=0.2, validation_size=0.0, random_state=DEFAULT_RANDOM_STATE, shuffle=True, stratify_col=None)

Validate and store the split proportions and stratification column.

stratify_col names the column to balance class proportions on; None disables stratification. random_state and shuffle are reused for both the test carve and the validation carve, so the partitioning is reproducible as a whole.

Raises:

Type	Description
ValueError	If test_size is not strictly between 0 and 1, if validation_size is outside [0, 1), or if the two together reach 1 and leave no rows for training.

Source code in skyulf-core/skyulf/preprocessing/split.py

def __init__(
    self,
    test_size: float = 0.2,
    validation_size: float = 0.0,
    random_state: int = DEFAULT_RANDOM_STATE,
    shuffle: bool = True,
    stratify_col: str | None = None,
):
    """Validate and store the split proportions and stratification column.

    ``stratify_col`` names the column to balance class proportions on;
    ``None`` disables stratification. ``random_state`` and ``shuffle`` are
    reused for both the test carve and the validation carve, so the
    partitioning is reproducible as a whole.

    Raises:
        ValueError: If ``test_size`` is not strictly between 0 and 1, if
            ``validation_size`` is outside ``[0, 1)``, or if the two
            together reach 1 and leave no rows for training.
    """
    if not 0 < test_size < 1:
        raise ValueError(f"test_size must be between 0 and 1 (exclusive), got {test_size!r}.")
    if not 0 <= validation_size < 1:
        raise ValueError(
            f"validation_size must be between 0 (inclusive) and 1 (exclusive), "
            f"got {validation_size!r}."
        )
    if test_size + validation_size >= 1:
        raise ValueError(
            f"test_size ({test_size!r}) + validation_size ({validation_size!r}) must be "
            f"less than 1, otherwise there is no data left for training."
        )
    self.test_size = test_size
    self.validation_size = validation_size
    self.random_state = random_state
    self.shuffle = shuffle
    self.stratify_col = stratify_col

split(df, *, sample_weight=None)

Split a whole frame row-wise, leaving any target column in place.

Shares :meth:split_xy's engine handling. Stratification here reads stratify_col out of the frame itself, so a configured column that is absent from df disables stratification with a warning instead of raising.

Source code in skyulf-core/skyulf/preprocessing/split.py

def split(
    self, df: pd.DataFrame | pl.DataFrame | SkyulfDataFrame, *, sample_weight: Any = None
) -> SplitDataset:
    """Split a whole frame row-wise, leaving any target column in place.

    Shares :meth:`split_xy`'s engine handling. Stratification here reads
    ``stratify_col`` out of the frame itself, so a configured column that is
    absent from ``df`` disables stratification with a warning instead of
    raising.
    """
    if sample_weight is not None:
        return self._split_weighted(df, None, sample_weight, paired=False)
    if is_polars(df):
        return self._split_polars(cast(Any, df))

    df_pd, was_polars = _to_pandas_remember_engine(df)
    stratify = self._frame_stratify(df_pd, label="Stratified split")

    train_val, test = train_test_split(
        df_pd,
        test_size=self.test_size,
        random_state=self.random_state,
        shuffle=self.shuffle,
        stratify=stratify,
    )

    validation, train = self._maybe_split_validation_frame(train_val)
    return SplitDataset(
        train=_back_to_engine(train, was_polars),
        test=_back_to_engine(test, was_polars),
        validation=_back_to_engine(validation, was_polars),
    )

split_indices(n, stratify=None, *, groups=None)

Split row positions for an initial train/test or calibration partition.

Callers gather features, targets and weights with these same positions. Optional groups remain whole; group splits require shuffle and cannot simultaneously stratify. This method does not create a third validation partition; split and split_xy own that optional second split.

Source code in skyulf-core/skyulf/preprocessing/split.py

def split_indices(self, n: int, stratify: Any = None, *, groups: Any = None) -> tuple[Any, Any]:
    """Split row positions for an initial train/test or calibration partition.

    Callers gather features, targets and weights with these same positions.
    Optional groups remain whole; group splits require shuffle and cannot
    simultaneously stratify. This method does not create a third validation
    partition; ``split`` and ``split_xy`` own that optional second split.
    """
    if groups is not None:
        if stratify is not None or not self.shuffle:
            raise ValueError("Group splitting requires shuffle=True and stratify=None.")
        check_consistent_length(np.arange(n), groups)
        if pd.isna(np.asarray(groups)).any():
            raise ValueError("Group metadata must be nonnull.")
        splitter = GroupShuffleSplit(
            n_splits=1, test_size=self.test_size, random_state=self.random_state
        )
        return next(splitter.split(np.arange(n), groups=groups))
    return self._split_indices(n, stratify)

split_xy(X, y, *, sample_weight=None)

Split an (X, y) pair into train/test and optional validation.

Polars input is partitioned by index gather and never converted; other engines round-trip through pandas and are converted back, so the result stays on the input engine. Requested stratification is downgraded to an unstratified split, with a warning, when the rarest class has fewer than two members.

Source code in skyulf-core/skyulf/preprocessing/split.py

def split_xy(
    self,
    X: pd.DataFrame | pl.DataFrame | SkyulfDataFrame,
    y: pd.Series | Any,
    *,
    sample_weight: Any = None,
) -> SplitDataset:
    """Split an ``(X, y)`` pair into train/test and optional validation.

    Polars input is partitioned by index gather and never converted; other
    engines round-trip through pandas and are converted back, so the result
    stays on the input engine. Requested stratification is downgraded to an
    unstratified split, with a warning, when the rarest class has fewer than
    two members.
    """
    check_consistent_length(X, y)
    if sample_weight is not None:
        return self._split_weighted(X, y, sample_weight, paired=True)
    if is_polars(X):
        return self._split_xy_polars(cast(Any, X), y)

    X_pd, was_polars = _to_pandas_remember_engine(X)
    y_pd, _ = _to_pandas_remember_engine(y)

    stratify = _safe_stratify(y_pd, "Stratified split") if self.stratify_col else None

    X_tv, X_test, y_tv, y_test = train_test_split(
        X_pd,
        y_pd,
        test_size=self.test_size,
        random_state=self.random_state,
        shuffle=self.shuffle,
        stratify=stratify,
    )

    validation, X_train, y_train = self._maybe_split_validation_xy(X_tv, y_tv)

    train = (_back_to_engine(X_train, was_polars), _back_to_engine(y_train, was_polars))
    test = (_back_to_engine(X_test, was_polars), _back_to_engine(y_test, was_polars))
    if validation is not None:
        validation = (
            _back_to_engine(validation[0], was_polars),
            _back_to_engine(validation[1], was_polars),
        )
    return SplitDataset(train=train, test=test, validation=validation)

FeatureTargetSplitApplier

Bases: BaseApplier

Separate the target column from the features without moving any rows.

Source code in skyulf-core/skyulf/preprocessing/split.py

class FeatureTargetSplitApplier(BaseApplier):
    """Separate the target column from the features without moving any rows."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the saved target name without executing feature/target extraction."""
        local_state_fields(raw, "feature_target_split", {"type", "target_column"})
        if not isinstance(raw["target_column"], str) or not raw["target_column"]:
            raise ValueError("Fitted target_column must be a nonempty string.")
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe native pair extraction; pipelines omit this training-only marker."""
        if engine not in {"pandas", "polars"}:
            return None
        FeatureTargetSplitApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    def apply(
        self,
        df: pd.DataFrame | SkyulfDataFrame | SplitDataset | tuple[Any, ...],
        params: dict[str, Any],
    ) -> tuple[pd.DataFrame, pd.Series] | SplitDataset:
        """Return ``(X, y)`` with ``target_column`` lifted out of the features.

        A ``SplitDataset`` has each train/test/validation member split on its
        own, and members that already are ``(X, y)`` pairs pass through
        untouched; a bare tuple is returned as-is.

        Raises:
            ValueError: If no ``target_column`` is configured.
        """
        target_col = params.get("target_column")
        if not target_col:
            raise ValueError("Target column must be specified for FeatureTargetSplitter")

        if isinstance(df, SplitDataset):
            train = _maybe_split_xy_member(df.train, target_col)
            test = _maybe_split_xy_member(df.test, target_col)
            validation = (
                _maybe_split_xy_member(df.validation, target_col)
                if df.validation is not None
                else None
            )
            return SplitDataset(
                train=train,
                test=test,
                validation=validation,
                train_sample_weight=df.train_sample_weight,
                evaluation_coverage=deepcopy(df.evaluation_coverage),
            )

        if isinstance(df, tuple):
            return cast(tuple[pd.DataFrame, pd.Series], df)

        return _split_xy_one(df, target_col)

apply(df, params)

Return (X, y) with target_column lifted out of the features.

A SplitDataset has each train/test/validation member split on its own, and members that already are (X, y) pairs pass through untouched; a bare tuple is returned as-is.

Raises:

Type	Description
ValueError	If no target_column is configured.

Source code in skyulf-core/skyulf/preprocessing/split.py

def apply(
    self,
    df: pd.DataFrame | SkyulfDataFrame | SplitDataset | tuple[Any, ...],
    params: dict[str, Any],
) -> tuple[pd.DataFrame, pd.Series] | SplitDataset:
    """Return ``(X, y)`` with ``target_column`` lifted out of the features.

    A ``SplitDataset`` has each train/test/validation member split on its
    own, and members that already are ``(X, y)`` pairs pass through
    untouched; a bare tuple is returned as-is.

    Raises:
        ValueError: If no ``target_column`` is configured.
    """
    target_col = params.get("target_column")
    if not target_col:
        raise ValueError("Target column must be specified for FeatureTargetSplitter")

    if isinstance(df, SplitDataset):
        train = _maybe_split_xy_member(df.train, target_col)
        test = _maybe_split_xy_member(df.test, target_col)
        validation = (
            _maybe_split_xy_member(df.validation, target_col)
            if df.validation is not None
            else None
        )
        return SplitDataset(
            train=train,
            test=test,
            validation=validation,
            train_sample_weight=df.train_sample_weight,
            evaluation_coverage=deepcopy(df.evaluation_coverage),
        )

    if isinstance(df, tuple):
        return cast(tuple[pd.DataFrame, pd.Series], df)

    return _split_xy_one(df, target_col)

inference_capability(state, *, engine) staticmethod

Describe native pair extraction; pipelines omit this training-only marker.

Source code in skyulf-core/skyulf/preprocessing/split.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe native pair extraction; pipelines omit this training-only marker."""
    if engine not in {"pandas", "polars"}:
        return None
    FeatureTargetSplitApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect the saved target name without executing feature/target extraction.

Source code in skyulf-core/skyulf/preprocessing/split.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect the saved target name without executing feature/target extraction."""
    local_state_fields(raw, "feature_target_split", {"type", "target_column"})
    if not isinstance(raw["target_column"], str) or not raw["target_column"]:
        raise ValueError("Fitted target_column must be a nonempty string.")
    return raw

FeatureTargetSplitCalculator

Bases: BaseCalculator

Carry the configured target column into the artifact.

Source code in skyulf-core/skyulf/preprocessing/split.py

@NodeRegistry.register("feature_target_split", FeatureTargetSplitApplier)
@node_meta(
    id="feature_target_split",
    name="Feature/Target Split",
    category="Data Operations",
    description="Split the dataset into features (X) and target (y).",
    params={"target_column": "target"},
    learns_from_data=False,
)
class FeatureTargetSplitCalculator(BaseCalculator):
    """Carry the configured target column into the artifact."""

    def fit(
        self,
        df: pd.DataFrame | SkyulfDataFrame | SplitDataset | tuple[Any, ...] | Any,
        config: dict[str, Any],
    ) -> FeatureTargetSplitArtifact:
        """Return a typed artifact holding ``target_column`` as a string.

        The key is omitted entirely when unconfigured or ``None``, which is what
        makes the applier raise rather than split on an empty name.
        """
        # Only ``target_column`` is consumed downstream by the Applier.
        # Build a typed artifact rather than echoing the raw config dict.
        artifact: FeatureTargetSplitArtifact = {"type": "feature_target_split"}
        if "target_column" in config and config["target_column"] is not None:
            artifact["target_column"] = str(config["target_column"])
        return artifact

    def infer_output_schema(
        self,
        input_schema: SkyulfSchema,
        config: dict[str, Any],
    ) -> SkyulfSchema | None:
        """Return ``input_schema``, target column included, so pickers can still select it."""
        # Output is (X, y). The target column lives in the y slot but is still
        # part of the dataset, so we keep it in the downstream schema. This
        # lets downstream column pickers (Encoder, Scaler, etc.) see and select
        # the target column. Runtime Appliers still receive the (X, y) tuple
        # via the dispatcher's ``unpack_pipeline_input`` and operate on X.
        return input_schema

fit(df, config)

Return a typed artifact holding target_column as a string.

The key is omitted entirely when unconfigured or None, which is what makes the applier raise rather than split on an empty name.

Source code in skyulf-core/skyulf/preprocessing/split.py

def fit(
    self,
    df: pd.DataFrame | SkyulfDataFrame | SplitDataset | tuple[Any, ...] | Any,
    config: dict[str, Any],
) -> FeatureTargetSplitArtifact:
    """Return a typed artifact holding ``target_column`` as a string.

    The key is omitted entirely when unconfigured or ``None``, which is what
    makes the applier raise rather than split on an empty name.
    """
    # Only ``target_column`` is consumed downstream by the Applier.
    # Build a typed artifact rather than echoing the raw config dict.
    artifact: FeatureTargetSplitArtifact = {"type": "feature_target_split"}
    if "target_column" in config and config["target_column"] is not None:
        artifact["target_column"] = str(config["target_column"])
    return artifact

infer_output_schema(input_schema, config)

Return input_schema, target column included, so pickers can still select it.

Source code in skyulf-core/skyulf/preprocessing/split.py

def infer_output_schema(
    self,
    input_schema: SkyulfSchema,
    config: dict[str, Any],
) -> SkyulfSchema | None:
    """Return ``input_schema``, target column included, so pickers can still select it."""
    # Output is (X, y). The target column lives in the y slot but is still
    # part of the dataset, so we keep it in the downstream schema. This
    # lets downstream column pickers (Encoder, Scaler, etc.) see and select
    # the target column. Runtime Appliers still receive the (X, y) tuple
    # via the dispatcher's ``unpack_pipeline_input`` and operate on X.
    return input_schema

SplitApplier

Bases: BaseApplier

Route the input to the right :class:DataSplitter entry point.

Source code in skyulf-core/skyulf/preprocessing/split.py

class SplitApplier(BaseApplier):
    """Route the input to the right :class:`DataSplitter` entry point."""

    def apply(
        self,
        df: pd.DataFrame | SkyulfDataFrame | tuple[Any, ...] | Any,
        params: dict[str, Any],
        *,
        sample_weight: Any = None,
    ) -> SplitDataset:
        """Split ``df`` into train/test and optional validation.

        Routes three ways: an ``(X, y)`` tuple is split as a pair; a plain frame
        carrying a configured ``target_column`` has that column lifted into ``y``
        first; anything else is split row-wise with the target left in place.
        """
        target_col = params.get("target_column")
        splitter = _build_splitter(params)

        if isinstance(df, tuple) and len(df) == 2:
            X, y = df
            return splitter.split_xy(cast(Any, X), y, sample_weight=sample_weight)

        # If a target column is configured and present, split features from target
        # so downstream nodes see real (X, y) tuples — not the placeholder
        # (df, None) shape that gets silently collapsed by transformers.
        if target_col and hasattr(df, "columns") and target_col in list(cast(Any, df).columns):
            frame = cast(Any, df)
            # `.drop(columns=[...])` is pandas-only syntax; polars' `.drop()`
            # takes column names positionally and has no `columns=` kwarg.
            X = frame.drop(target_col) if is_polars(frame) else frame.drop(columns=[target_col])
            y = frame[target_col]
            return splitter.split_xy(X, y, sample_weight=sample_weight)

        return splitter.split(cast(Any, df), sample_weight=sample_weight)

apply(df, params, *, sample_weight=None)

Split df into train/test and optional validation.

Routes three ways: an (X, y) tuple is split as a pair; a plain frame carrying a configured target_column has that column lifted into y first; anything else is split row-wise with the target left in place.

Source code in skyulf-core/skyulf/preprocessing/split.py

def apply(
    self,
    df: pd.DataFrame | SkyulfDataFrame | tuple[Any, ...] | Any,
    params: dict[str, Any],
    *,
    sample_weight: Any = None,
) -> SplitDataset:
    """Split ``df`` into train/test and optional validation.

    Routes three ways: an ``(X, y)`` tuple is split as a pair; a plain frame
    carrying a configured ``target_column`` has that column lifted into ``y``
    first; anything else is split row-wise with the target left in place.
    """
    target_col = params.get("target_column")
    splitter = _build_splitter(params)

    if isinstance(df, tuple) and len(df) == 2:
        X, y = df
        return splitter.split_xy(cast(Any, X), y, sample_weight=sample_weight)

    # If a target column is configured and present, split features from target
    # so downstream nodes see real (X, y) tuples — not the placeholder
    # (df, None) shape that gets silently collapsed by transformers.
    if target_col and hasattr(df, "columns") and target_col in list(cast(Any, df).columns):
        frame = cast(Any, df)
        # `.drop(columns=[...])` is pandas-only syntax; polars' `.drop()`
        # takes column names positionally and has no `columns=` kwarg.
        X = frame.drop(target_col) if is_polars(frame) else frame.drop(columns=[target_col])
        y = frame[target_col]
        return splitter.split_xy(X, y, sample_weight=sample_weight)

    return splitter.split(cast(Any, df), sample_weight=sample_weight)

SplitCalculator

Bases: BaseCalculator

Carry the split configuration into the artifact; splitting happens at apply time.

Source code in skyulf-core/skyulf/preprocessing/split.py

@NodeRegistry.register("Split", SplitApplier)
@NodeRegistry.register("TrainTestSplitter", SplitApplier)
@node_meta(
    id="TrainTestSplitter",
    name="Train/Test Split",
    category="Data Operations",
    description="Split the dataset into training and testing sets.",
    params={
        "test_size": 0.2,
        "validation_size": 0.0,
        "random_state": DEFAULT_RANDOM_STATE,
        "shuffle": True,
        "stratify": False,
        "target_column": "target",
    },
    learns_from_data=False,
    is_splitter=True,
)
class SplitCalculator(BaseCalculator):
    """Carry the split configuration into the artifact; splitting happens at apply time."""

    def fit(
        self, df: pd.DataFrame | SkyulfDataFrame | tuple[Any, ...] | Any, config: dict[str, Any]
    ) -> SplitArtifact:
        """Return a typed artifact and warn about ignored public configuration keys.

        ``type`` and underscore-prefixed routing metadata are ignored quietly.
        Use ``stratify=True`` with ``target_column`` to request stratification;
        ``stratify_col`` belongs to ``DataSplitter``, not this node's config.
        """
        # No learning from data; pass through known split params from config.
        # Constructed explicitly so the artifact shape matches SplitArtifact
        # rather than echoing arbitrary user keys back into the params dict.
        unknown = sorted(
            key
            for key in config
            if key not in _SPLIT_CONFIG_KEYS and key != "type" and not key.startswith("_")
        )
        if unknown:
            logger.warning(
                "TrainTestSplitter ignored unrecognized config keys: %s. Supported keys: %s.",
                unknown,
                ", ".join(_SPLIT_CONFIG_KEYS),
            )
        artifact: SplitArtifact = {"type": "split"}
        for key in _SPLIT_CONFIG_KEYS:
            if key in config:
                artifact[key] = config[key]  # type: ignore[literal-required]
        return artifact

    def infer_output_schema(
        self,
        input_schema: SkyulfSchema,
        config: dict[str, Any],
    ) -> SkyulfSchema | None:
        """Return ``input_schema``: every split frame keeps the full column set."""
        # Split produces a SplitDataset whose train/test/val frames carry
        # the full input schema. Downstream consumers see the same columns.
        return input_schema

fit(df, config)

Return a typed artifact and warn about ignored public configuration keys.

type and underscore-prefixed routing metadata are ignored quietly. Use stratify=True with target_column to request stratification; stratify_col belongs to DataSplitter, not this node's config.

Source code in skyulf-core/skyulf/preprocessing/split.py

def fit(
    self, df: pd.DataFrame | SkyulfDataFrame | tuple[Any, ...] | Any, config: dict[str, Any]
) -> SplitArtifact:
    """Return a typed artifact and warn about ignored public configuration keys.

    ``type`` and underscore-prefixed routing metadata are ignored quietly.
    Use ``stratify=True`` with ``target_column`` to request stratification;
    ``stratify_col`` belongs to ``DataSplitter``, not this node's config.
    """
    # No learning from data; pass through known split params from config.
    # Constructed explicitly so the artifact shape matches SplitArtifact
    # rather than echoing arbitrary user keys back into the params dict.
    unknown = sorted(
        key
        for key in config
        if key not in _SPLIT_CONFIG_KEYS and key != "type" and not key.startswith("_")
    )
    if unknown:
        logger.warning(
            "TrainTestSplitter ignored unrecognized config keys: %s. Supported keys: %s.",
            unknown,
            ", ".join(_SPLIT_CONFIG_KEYS),
        )
    artifact: SplitArtifact = {"type": "split"}
    for key in _SPLIT_CONFIG_KEYS:
        if key in config:
            artifact[key] = config[key]  # type: ignore[literal-required]
    return artifact

infer_output_schema(input_schema, config)

Return input_schema: every split frame keeps the full column set.

Source code in skyulf-core/skyulf/preprocessing/split.py

def infer_output_schema(
    self,
    input_schema: SkyulfSchema,
    config: dict[str, Any],
) -> SkyulfSchema | None:
    """Return ``input_schema``: every split frame keeps the full column set."""
    # Split produces a SplitDataset whose train/test/val frames carry
    # the full input schema. Downstream consumers see the same columns.
    return input_schema