API: preprocessing

The preprocessing package contains Calculator/Applier nodes and the FeatureEngineer orchestrator.

skyulf.preprocessing

AuditedFoldPreprocessor

Decorates a :class:FoldPreprocessor to record per-fold row counts.

Every fit_transform/transform call records the number of input rows it received. The isolation invariant a leak-free run must satisfy is max(fit_rows) <= train_rows — a leaked fit would see the train split plus held-out rows. Exposed via :meth:summary so the app can log it and persist it in node metrics for post-hoc audit (findings 2026-08-26 §3/B).

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

class AuditedFoldPreprocessor:
    """Decorates a :class:`FoldPreprocessor` to record per-fold row counts.

    Every ``fit_transform``/``transform`` call records the number of input
    rows it received. The isolation invariant a leak-free run must satisfy is
    ``max(fit_rows) <= train_rows`` — a leaked fit would see the train split
    plus held-out rows. Exposed via :meth:`summary` so the app can log it and
    persist it in node metrics for post-hoc audit (findings 2026-08-26 §3/B).
    """

    def __init__(self, inner: Any):
        self._inner = inner
        self.fit_rows: list[int] = []
        self.transform_rows: list[int] = []
        self.changes_row_count = getattr(inner, "changes_row_count", False)

    @property
    def inner(self) -> Any:
        return self._inner

    def fit_transform(self, X: Any, y: Any) -> tuple[Any, Any]:
        self.fit_rows.append(frame_rows(X))
        return self._inner.fit_transform(X, y)

    def transform(self, X: Any, y: Any) -> tuple[Any, Any]:
        self.transform_rows.append(frame_rows(X))
        return self._inner.transform(X, y)

    def summary(self, train_rows: int | None = None) -> dict[str, Any]:
        result: dict[str, Any] = {
            "fit_calls": len(self.fit_rows),
            "max_fit_rows": max(self.fit_rows, default=0),
            "transform_calls": len(self.transform_rows),
        }
        if train_rows is not None:
            result["train_rows"] = train_rows
            result["isolation_ok"] = result["max_fit_rows"] <= train_rows
        return result

BaseApplier

Bases: ABC

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

class BaseApplier(ABC):
    @abstractmethod
    def apply(self, df: pd.DataFrame | SkyulfDataFrame | tuple, params: dict[str, Any]) -> Any:
        """
        Applies the transformation using fitted parameters.

        The return type is intentionally `Any` because the concrete shape
        depends on the input: passing a `DataFrame` returns a `DataFrame`;
        passing an `(X, y)` tuple returns a tuple; splitters return
        `SplitDataset`. Encoding every case as a union forces callers to
        defensively narrow on every use, which is worse than `Any` here.
        """

apply(df, params) abstractmethod

Applies the transformation using fitted parameters.

The return type is intentionally Any because the concrete shape depends on the input: passing a DataFrame returns a DataFrame; passing an (X, y) tuple returns a tuple; splitters return SplitDataset. Encoding every case as a union forces callers to defensively narrow on every use, which is worse than Any here.

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

@abstractmethod
def apply(self, df: pd.DataFrame | SkyulfDataFrame | tuple, params: dict[str, Any]) -> Any:
    """
    Applies the transformation using fitted parameters.

    The return type is intentionally `Any` because the concrete shape
    depends on the input: passing a `DataFrame` returns a `DataFrame`;
    passing an `(X, y)` tuple returns a tuple; splitters return
    `SplitDataset`. Encoding every case as a union forces callers to
    defensively narrow on every use, which is worse than `Any` here.
    """

BaseCalculator

Bases: ABC

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

class BaseCalculator(ABC):
    @abstractmethod
    def fit(
        self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
    ) -> Mapping[str, Any]:
        """
        Calculates parameters from the training data.
        Returns a Mapping of fitted parameters (typically a TypedDict
        ``*Artifact`` declared in ``preprocessing._artifacts``). The return
        type is ``Mapping`` rather than ``Dict`` so concrete TypedDict
        subclasses are valid LSP-substitutable returns.
        """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema | None:
        """Best-effort prediction of the output schema from config alone.

        Override this in concrete Calculators when the output columns/dtypes
        can be derived purely from ``input_schema`` and ``config`` (i.e.
        without seeing data). Examples:

        * Scalers — pass through (output == input).
        * Drop columns by name — drop the configured names.
        * One-hot — adds K columns per categorical (K is data-dependent →
          return ``None``).

        Default returns ``None`` to signal "unknown / data-dependent";
        callers should fall back to runtime introspection.
        """
        return None

fit(df, config) abstractmethod

Calculates parameters from the training data. Returns a Mapping of fitted parameters (typically a TypedDict *Artifact declared in preprocessing._artifacts). The return type is Mapping rather than Dict so concrete TypedDict subclasses are valid LSP-substitutable returns.

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

@abstractmethod
def fit(
    self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
) -> Mapping[str, Any]:
    """
    Calculates parameters from the training data.
    Returns a Mapping of fitted parameters (typically a TypedDict
    ``*Artifact`` declared in ``preprocessing._artifacts``). The return
    type is ``Mapping`` rather than ``Dict`` so concrete TypedDict
    subclasses are valid LSP-substitutable returns.
    """

infer_output_schema(input_schema, config)

Best-effort prediction of the output schema from config alone.

Override this in concrete Calculators when the output columns/dtypes can be derived purely from input_schema and config (i.e. without seeing data). Examples:

• Scalers — pass through (output == input).
• Drop columns by name — drop the configured names.
• One-hot — adds K columns per categorical (K is data-dependent → return None).

Default returns None to signal "unknown / data-dependent"; callers should fall back to runtime introspection.

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

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema | None:
    """Best-effort prediction of the output schema from config alone.

    Override this in concrete Calculators when the output columns/dtypes
    can be derived purely from ``input_schema`` and ``config`` (i.e.
    without seeing data). Examples:

    * Scalers — pass through (output == input).
    * Drop columns by name — drop the configured names.
    * One-hot — adds K columns per categorical (K is data-dependent →
      return ``None``).

    Default returns ``None`` to signal "unknown / data-dependent";
    callers should fall back to runtime introspection.
    """
    return None

CustomBinningCalculator

Bases: BaseCalculator

Apply user-supplied bin edges to selected columns.

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

@NodeRegistry.register("CustomBinning", CustomBinningApplier)
@node_meta(
    id="CustomBinning",
    name="Custom Binning",
    category="Preprocessing",
    description="Bin data using custom edges.",
    params={"bins": [], "columns": []},
    learns_from_data=False,
)
class CustomBinningCalculator(BaseCalculator):
    """Apply user-supplied bin edges to selected columns."""

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> GeneralBinningArtifact:
        if user_picked_no_columns(config):
            return cast(GeneralBinningArtifact, {})

        X, columns = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
        bins = config.get("bins")

        bin_edges_map: dict[str, list[float]] = {}
        if bins:
            sorted_bins = sorted(bins)
            for col in columns:
                if col in X.columns:
                    bin_edges_map[col] = sorted_bins

        artifact: dict[str, Any] = {
            "type": "general_binning",  # Reuses GeneralBinningApplier.
            "bin_edges": bin_edges_map,
        }
        artifact.update(_passthrough_artifact_options(config))
        return cast(GeneralBinningArtifact, artifact)

FeatureEngineer

Orchestrates a sequence of feature engineering steps.

Examples:

>>> engineer = FeatureEngineer([])
>>> transformed, metrics = engineer.fit_transform(data)

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

class FeatureEngineer:
    """
    Orchestrates a sequence of feature engineering steps.

    Examples:
        >>> engineer = FeatureEngineer([])
        >>> transformed, metrics = engineer.fit_transform(data)
    """

    # Resampling steps (SMOTE/undersampling) must only ever run on the train
    # split -- applying them to test/validation would fabricate synthetic rows
    # or delete real held-out rows purely to balance classes, corrupting any
    # metrics later computed on that "held-out" data. Kept as a single shared
    # constant (rather than duplicated literals) so `transform()`, `_run_step`,
    # and `_collect_step_metrics` can't drift out of sync with each other.
    _RESAMPLING_TYPES: ClassVar[set[str]] = {"Oversampling", "Undersampling"}

    # Row-dropping steps are train-time cleaning only (F-18). Running them at
    # inference would silently vanish requested input rows -- prediction
    # responses carry no row keys, so callers could never tell which inputs
    # lost their prediction. Skipping them means a null row surfaces as a
    # visible model error instead of a silent misalignment.
    _ROW_DROPPING_TYPES: ClassVar[set[str]] = {"Deduplicate", "DropMissingRows"}

    def __init__(
        self,
        steps_config: Sequence[PreprocessingStepConfig | dict[str, Any]],
        *,
        _validated: bool = False,
    ):
        # `Sequence` (covariant) accepts list[dict] or list[PreprocessingStepConfig].
        if not _validated:
            validate_preprocessing_steps(steps_config)
        self.steps_config = steps_config
        self.fitted_steps: list[dict[str, Any]] = []

    def transform(self, data: pd.DataFrame | SkyulfDataFrame | Any) -> Any:
        """
        Apply fitted transformations to new data.
        """
        current_data = data

        for step in self.fitted_steps:
            name = step["name"]
            transformer_type = step["type"]
            applier = step["applier"]
            artifact = step["artifact"]

            # Skip splitters during inference/transform
            if transformer_type in [
                "TrainTestSplitter",
                "feature_target_split",
                *self._RESAMPLING_TYPES,
                *self._ROW_DROPPING_TYPES,
            ]:
                continue

            logger.debug(f"Applying step: {name} ({transformer_type})")
            current_data = applier.apply(current_data, artifact)

        return current_data

    def fit_transform(self, data: pd.DataFrame | SkyulfDataFrame | Any, node_id_prefix="") -> Any:
        """
        Runs the pipeline on data.
        Returns: (transformed_data, metrics_dict)
        """
        self.fitted_steps = []  # Reset fitted steps
        current_data = data
        metrics: dict[str, Any] = {
            "summary": {
                "fit_time": 0.0,
                "peak_memory_bytes": 0,
                "rows_in": 0,
                "rows_out": 0,
            },
            "steps": {},
        }

        for i, step in enumerate(self.steps_config):
            name = step["name"]
            transformer_type = step["transformer"]
            params = step.get("params", {})
            step_metrics: dict[str, Any] = {}
            step_key = f"{i}:{name}"

            logger.info(f"Running step {i}: {name} ({transformer_type})")
            logger.debug(f"FeatureEngineer running step {i}: {name} ({transformer_type})")
            logger.debug(f"current_data type: {type(current_data)}")

            # Snapshot before for shape-delta + Winsorize value-clipping metrics
            rows_before, cols_before = get_data_stats(current_data)
            data_before = current_data

            calculator, applier = self._get_transformer_components(transformer_type)
            step_node_id = f"{node_id_prefix}_{name}"

            current_data, fitted_params, transformer_inst = self._run_step(
                transformer_type=transformer_type,
                name=name,
                calculator=calculator,
                applier=applier,
                step_node_id=step_node_id,
                current_data=current_data,
                params=params,
            )

            logger.debug(f"Step {i} complete. New data type: {type(current_data)}")

            rows_after, cols_after = get_data_stats(current_data)
            self._collect_step_metrics(
                transformer_type=transformer_type,
                fitted_params=fitted_params,
                data_before=data_before,
                current_data=current_data,
                params=params,
                rows_before=rows_before,
                cols_before=cols_before,
                rows_after=rows_after,
                cols_after=cols_after,
                name=name,
                metrics=step_metrics,
            )

            step_record = {
                "name": name,
                "transformer": transformer_type,
                "fit_time": (
                    getattr(transformer_inst, "fit_time", 0.0)
                    if transformer_inst is not None
                    else 0.0
                ),
                "peak_memory_bytes": (
                    getattr(transformer_inst, "peak_memory_bytes", 0)
                    if transformer_inst is not None
                    else 0
                ),
                "rows_in": (
                    getattr(transformer_inst, "rows_in", rows_before)
                    if transformer_inst is not None
                    else rows_before
                ),
                "rows_out": (
                    getattr(transformer_inst, "rows_out", rows_after)
                    if transformer_inst is not None
                    else rows_after
                ),
                "details": step_metrics,
            }
            metrics["steps"][step_key] = step_record

            summary = metrics["summary"]
            summary["fit_time"] += step_record["fit_time"]
            summary["peak_memory_bytes"] = max(
                summary["peak_memory_bytes"], step_record["peak_memory_bytes"]
            )
            if i == 0:
                summary["rows_in"] = step_record["rows_in"]
            summary["rows_out"] = step_record["rows_out"]

        metrics["fit_time"] = metrics["summary"]["fit_time"]
        metrics["peak_memory_bytes"] = metrics["summary"]["peak_memory_bytes"]
        metrics["rows_in"] = metrics["summary"]["rows_in"]
        metrics["rows_out"] = metrics["summary"]["rows_out"]

        return current_data, metrics

    # ------------------------------------------------------------------
    # Step execution
    # ------------------------------------------------------------------

    def _run_step(
        self,
        *,
        transformer_type: str,
        name: str,
        calculator: Any,
        applier: Any,
        step_node_id: str,
        current_data: Any,
        params: dict[str, Any],
    ) -> tuple:  # Returns (data, params, transformer)
        """Execute one pipeline step. Returns (new_data, fitted_params).

        Splitters change the data structure (DataFrame -> SplitDataset / (X, y)),
        so they bypass StatefulTransformer; everything else goes through the
        standard fit_transform wrapper and is appended to fitted_steps.
        """
        transformer = StatefulTransformer(
            calculator,
            applier,
            step_node_id,
            apply_on_test=transformer_type not in self._RESAMPLING_TYPES,
            apply_on_validation=transformer_type not in self._RESAMPLING_TYPES,
        )
        fitted_params: dict[str, Any] = {}

        if transformer_type == "TrainTestSplitter":
            logger.debug("Handling TrainTestSplitter")
            # A raw (unwrapped) polars DataFrame satisfies neither `pd.DataFrame`
            # nor the `SkyulfDataFrame` protocol (it has no `.copy()`, uses
            # `.clone()` instead) even though it's a fully-supported "not yet
            # split" input -- `SplitCalculator`/`SplitApplier` already handle it
            # correctly via their own polars round-trip. Without this extra
            # check, a raw polars DataFrame passed straight to
            # `SkyulfPipeline.fit()` (the advertised polars-native usage) would
            # silently skip the split entirely and fit/evaluate on the whole
            # dataset with no held-out test set.
            if isinstance(current_data, pd.DataFrame | SkyulfDataFrame | tuple | pl.DataFrame):
                params = calculator.fit(current_data, params)
                current_data = applier.apply(current_data, params)
            else:
                logger.debug(f"Skipping TrainTestSplitter. current_data is {type(current_data)}")
                logger.warning(
                    "Attempting to split an already split dataset. Skipping TrainTestSplitter."
                )
            return current_data, fitted_params, None

        if transformer_type == "feature_target_split":
            logger.debug("Handling feature_target_split")
            params = calculator.fit(current_data, params)
            current_data = applier.apply(current_data, params)
            return current_data, fitted_params, None

        logger.debug("Handling standard transformer via StatefulTransformer")
        current_data = transformer.fit_transform(current_data, params)
        fitted_params = transformer.params
        self.fitted_steps.append(
            {
                "name": name,
                "type": transformer_type,
                "applier": applier,
                "artifact": fitted_params,
            }
        )
        return current_data, fitted_params, transformer

    # ------------------------------------------------------------------
    # Metrics collection
    # ------------------------------------------------------------------

    # Transformer-type groups, kept as class constants so dispatch is data-driven.
    _IMPUTATION_TYPES: ClassVar[set[str]] = {"SimpleImputer", "KNNImputer", "IterativeImputer"}
    _FEATURE_SELECTION_TYPES: ClassVar[set[str]] = {
        "feature_selection",
        "UnivariateSelection",
        "ModelBasedSelection",
        "VarianceThreshold",
    }
    _SCALING_TYPES: ClassVar[set[str]] = {
        "StandardScaler",
        "MinMaxScaler",
        "RobustScaler",
        "MaxAbsScaler",
    }
    _OUTLIER_TYPES: ClassVar[set[str]] = {"IQR", "Winsorize", "ZScore", "EllipticEnvelope"}
    _BUCKETING_TYPES: ClassVar[set[str]] = {
        "GeneralBinning",
        "EqualWidthBinning",
        "EqualFrequencyBinning",
        "CustomBinning",
        "KBinsDiscretizer",
    }
    _FEATURE_GEN_TYPES: ClassVar[set[str]] = {"FeatureMath", "FeatureGenerationNode"}
    _ROW_DROP_TYPES: ClassVar[set[str]] = {
        "DropMissingRows",
        "Deduplicate",
        "IQR",
        "ZScore",
        "EllipticEnvelope",
        "Winsorize",
    }
    _ENCODER_TYPES: ClassVar[set[str]] = {
        "OneHotEncoder",
        "LabelEncoder",
        "OrdinalEncoder",
        "TargetEncoder",
        "HashEncoder",
        "DummyEncoder",
    }

    def _collect_step_metrics(
        self,
        *,
        transformer_type: str,
        fitted_params: dict[str, Any],
        data_before: Any,
        current_data: Any,
        params: dict[str, Any],
        rows_before: int,
        cols_before: Any,
        rows_after: int,
        cols_after: Any,
        name: str,
        metrics: dict[str, Any],
    ) -> None:
        """Populate one step record's details dict with node-specific metrics."""
        try:
            if fitted_params:
                self._metrics_from_fitted_params(
                    transformer_type, fitted_params, data_before, current_data, metrics
                )
        except Exception as e:  # noqa: BLE001 - step metrics are best-effort; logged
            logger.warning(f"Failed to retrieve metrics for step {name}: {e}")

        if transformer_type in self._RESAMPLING_TYPES:
            self._metrics_resampling(current_data, params, metrics)

        if rows_after > 0 or cols_after:
            self._metrics_shape_change(
                transformer_type,
                data_before,
                current_data,
                params,
                rows_before,
                cols_before,
                rows_after,
                cols_after,
                metrics,
            )

    @staticmethod
    def _copy_present_keys(
        fitted_params: dict[str, Any], metrics: dict[str, Any], keys: tuple[str, ...]
    ) -> None:
        """Copy fitted params into a single step's details dict when present."""
        for key in keys:
            if key in fitted_params:
                metrics[key] = fitted_params[key]

    # Each rule maps a set of transformer types to the fitted_params/metrics key
    # to copy over when that transformer type produced it. Keys are the same on
    # both sides for every current rule.
    _OUTLIER_METRIC_RULES: tuple[tuple[frozenset[str], str], ...] = (
        (frozenset({"IQR", "Winsorize"}), "bounds"),
        (frozenset({"ZScore"}), "stats"),
        (frozenset({"EllipticEnvelope"}), "contamination"),
    )

    def _apply_outlier_metrics(
        self, transformer_type: str, fitted_params: dict[str, Any], metrics: dict[str, Any]
    ) -> None:
        """Populate one step's outlier details (warnings/bounds/stats/contamination)."""
        if transformer_type in self._OUTLIER_TYPES and "warnings" in fitted_params:
            metrics["warnings"] = fitted_params["warnings"]
        for types, key in self._OUTLIER_METRIC_RULES:
            if transformer_type in types and key in fitted_params:
                metrics[key] = fitted_params[key]

    def _apply_feature_gen_metrics(
        self,
        fitted_params: dict[str, Any],
        data_before: Any,
        current_data: Any,
        metrics: dict[str, Any],
    ) -> None:
        """Populate one step's generated-feature details and operation metadata."""
        if "operations" in fitted_params:
            metrics["operations_count"] = len(fitted_params["operations"])
            metrics["operations"] = fitted_params["operations"]
        new_cols = self._diff_generated_columns(data_before, current_data)
        if new_cols is not None:
            metrics["generated_features"] = new_cols

    def _metrics_from_fitted_params(
        self,
        transformer_type: str,
        fitted_params: dict[str, Any],
        data_before: Any,
        current_data: Any,
        metrics: dict[str, Any],
    ) -> None:
        """Populate one step's details from fitted parameters and derived artifacts."""
        if transformer_type in self._IMPUTATION_TYPES:
            self._copy_present_keys(
                fitted_params, metrics, ("missing_counts", "total_missing", "fill_values")
            )

        if transformer_type in self._FEATURE_SELECTION_TYPES:
            self._copy_present_keys(
                fitted_params,
                metrics,
                (
                    "feature_scores",
                    "p_values",
                    "feature_importances",
                    "variances",
                    "ranking",
                    "selected_columns",
                ),
            )

        if transformer_type in self._SCALING_TYPES:
            self._copy_present_keys(
                fitted_params,
                metrics,
                (
                    "mean",
                    "scale",
                    "var",
                    "min",
                    "data_min",
                    "data_max",
                    "center",
                    "max_abs",
                    "columns",
                ),
            )

        self._apply_outlier_metrics(transformer_type, fitted_params, metrics)

        if transformer_type in self._BUCKETING_TYPES:
            self._copy_present_keys(fitted_params, metrics, ("bin_edges", "n_bins"))

        if transformer_type in self._FEATURE_GEN_TYPES:
            self._apply_feature_gen_metrics(fitted_params, data_before, current_data, metrics)

    @staticmethod
    def _columns_diff_if_dataframes(before: Any, after: Any):
        """Return the column-set difference if both objects are DataFrames, else None."""
        if isinstance(before, pd.DataFrame | SkyulfDataFrame) and isinstance(
            after, pd.DataFrame | SkyulfDataFrame
        ):
            return list(set(after.columns) - set(before.columns))
        return None

    @classmethod
    def _diff_generated_columns_split_dataset(
        cls, data_before: SplitDataset, current_data: SplitDataset
    ):
        """Diff columns for the SplitDataset case, handling both DataFrame and (X, y) train shapes."""
        before_train, after_train = data_before.train, current_data.train
        diff = cls._columns_diff_if_dataframes(before_train, after_train)
        if diff is not None:
            return diff
        if isinstance(before_train, tuple) and isinstance(after_train, tuple):
            x_before, _ = before_train
            x_after, _ = after_train
            return cls._columns_diff_if_dataframes(x_before, x_after)
        return None

    @classmethod
    def _diff_generated_columns(cls, data_before: Any, current_data: Any):
        """Return the set of newly added columns between two pipeline data objects.

        Handles plain DataFrames, SplitDatasets of DataFrames, and (X, y) tuple variants.
        Returns None if the structures don't allow a meaningful diff.
        """
        diff = cls._columns_diff_if_dataframes(data_before, current_data)
        if diff is not None:
            return diff

        if isinstance(data_before, SplitDataset) and isinstance(current_data, SplitDataset):
            return cls._diff_generated_columns_split_dataset(data_before, current_data)
        return None

    @staticmethod
    def _extract_y_from_split_dataset(current_data: SplitDataset, params: dict[str, Any]):
        """Pull the target Series out of a SplitDataset's train split (tuple or DataFrame form)."""
        if isinstance(current_data.train, tuple):
            _, y_res = current_data.train
            return y_res
        if isinstance(current_data.train, (pd.DataFrame, SkyulfDataFrame)):
            target_col = params.get("target_column")
            if target_col and target_col in current_data.train.columns:
                return current_data.train[target_col]
        return None

    @classmethod
    def _extract_y_for_resampling(cls, current_data: Any, params: dict[str, Any]):
        """Pull the target Series out of whatever shape the resampler produced."""
        if isinstance(current_data, SplitDataset):
            return cls._extract_y_from_split_dataset(current_data, params)
        if isinstance(current_data, tuple):
            _, y_res = current_data
            return y_res
        if isinstance(current_data, pd.DataFrame | SkyulfDataFrame):
            target_col = params.get("target_column")
            if target_col and target_col in current_data.columns:
                return current_data[target_col]
        return None

    def _metrics_resampling(
        self, current_data: Any, params: dict[str, Any], metrics: dict[str, Any]
    ) -> None:
        """Populate one resampling step's class-balance details."""
        try:
            y_res: Any = self._extract_y_for_resampling(current_data, params)
            if y_res is None:
                return
            if hasattr(y_res, "to_pandas"):
                y_res = y_res.to_pandas()
            counts = y_res.value_counts().to_dict()
            metrics["class_counts"] = {str(k): int(v) for k, v in counts.items()}
            metrics["total_samples"] = int(len(y_res))
        except Exception as e:  # noqa: BLE001 - resampling metrics are best-effort; logged
            logger.warning(f"Failed to calculate resampling metrics: {e}")

    @staticmethod
    def _to_pandas_if_needed(obj: Any) -> Any:
        """Convert obj to pandas via to_pandas() if it supports that, otherwise return as-is."""
        return obj.to_pandas() if hasattr(obj, "to_pandas") else obj

    @classmethod
    def _count_diff_cells(cls, a: Any, b: Any, types: tuple[type, ...]) -> int:
        """Count differing cells between two objects of the given types with matching shape."""
        a = cls._to_pandas_if_needed(a)
        b = cls._to_pandas_if_needed(b)
        if isinstance(a, types) and isinstance(b, types) and a.shape == b.shape:
            return int(a.ne(b).sum().sum())
        return 0

    @classmethod
    def _count_winsorize_tuple_diffs(cls, d1: tuple, d2: tuple) -> int:
        """Count differing cells across the X/y halves of two (X, y) tuple pairs."""
        diffs = cls._count_diff_cells(d1[0], d2[0], (pd.DataFrame,))
        diffs += cls._count_diff_cells(d1[1], d2[1], (pd.DataFrame, pd.Series))
        return diffs

    @classmethod
    def _count_winsorize_diffs(cls, d1: Any, d2: Any) -> int:
        """Count cells that differ between two data objects, for Winsorize clipping metric."""
        d1 = cls._to_pandas_if_needed(d1)
        d2 = cls._to_pandas_if_needed(d2)

        if isinstance(d1, pd.DataFrame) and isinstance(d2, pd.DataFrame):
            return cls._count_diff_cells(d1, d2, (pd.DataFrame,))

        if isinstance(d1, tuple) and isinstance(d2, tuple) and len(d1) == 2 and len(d2) == 2:
            return cls._count_winsorize_tuple_diffs(d1, d2)
        return 0

    def _metrics_winsorize_clipped(
        self, data_before: Any, current_data: Any, metrics: dict[str, Any]
    ) -> None:
        """Populate one Winsorize step's clipped-value count detail."""
        try:
            clipped_count = 0
            if isinstance(data_before, (pd.DataFrame, SkyulfDataFrame)) and isinstance(
                current_data, (pd.DataFrame, SkyulfDataFrame)
            ):
                clipped_count = self._count_winsorize_diffs(data_before, current_data)
            elif isinstance(data_before, SplitDataset) and isinstance(current_data, SplitDataset):
                clipped_count += self._count_winsorize_diffs(data_before.train, current_data.train)
                clipped_count += self._count_winsorize_diffs(data_before.test, current_data.test)
                clipped_count += self._count_winsorize_diffs(
                    data_before.validation, current_data.validation
                )
            metrics["values_clipped"] = clipped_count
        except Exception as e:  # noqa: BLE001 - winsorize metrics are best-effort; logged
            logger.warning(f"Failed to calculate values_clipped for Winsorize: {e}")

    def _metrics_shape_change(
        self,
        transformer_type: str,
        data_before: Any,
        current_data: Any,
        params: dict[str, Any],
        rows_before: int,
        cols_before: Any,
        rows_after: int,
        cols_after: Any,
        metrics: dict[str, Any],
    ) -> None:
        """Populate one step's shape-change details such as dropped rows/columns."""
        if transformer_type in self._ROW_DROP_TYPES:
            dropped = rows_before - rows_after
            metrics[f"{transformer_type}_rows_removed"] = dropped
            metrics[f"{transformer_type}_rows_remaining"] = rows_after
            metrics[f"{transformer_type}_rows_total"] = rows_before
            metrics["rows_removed"] = dropped
            metrics["rows_total"] = rows_before
            if transformer_type == "Winsorize":
                self._metrics_winsorize_clipped(data_before, current_data, metrics)

        if transformer_type == "MissingIndicator":
            new_cols_set = cols_after - cols_before
            metrics["missing_indicators_created"] = len(new_cols_set)
            metrics["missing_indicators_columns"] = list(new_cols_set)

        if transformer_type in {"DropMissingColumns", "feature_selection"}:
            dropped_cols_set = cols_before - cols_after
            metrics["dropped_columns"] = list(dropped_cols_set)
            metrics["dropped_columns_count"] = len(dropped_cols_set)

        if transformer_type in self._ENCODER_TYPES:
            new_cols_set = cols_after - cols_before
            metrics["new_features_count"] = len(new_cols_set)
            metrics["encoded_columns_count"] = len(params.get("columns", []))
            if "categories_count" in params:
                metrics["categories_count"] = params["categories_count"]
            if "classes_count" in params:
                metrics["classes_count"] = params["classes_count"]

    def _get_transformer_components(self, type_name: str):
        try:
            return (
                NodeRegistry.get_calculator(type_name)(),
                NodeRegistry.get_applier(type_name)(),
            )
        except ValueError as exc:
            raise ValueError(f"Unknown transformer type: {type_name}. {exc}") from exc

fit_transform(data, node_id_prefix='')

Runs the pipeline on data. Returns: (transformed_data, metrics_dict)

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

def fit_transform(self, data: pd.DataFrame | SkyulfDataFrame | Any, node_id_prefix="") -> Any:
    """
    Runs the pipeline on data.
    Returns: (transformed_data, metrics_dict)
    """
    self.fitted_steps = []  # Reset fitted steps
    current_data = data
    metrics: dict[str, Any] = {
        "summary": {
            "fit_time": 0.0,
            "peak_memory_bytes": 0,
            "rows_in": 0,
            "rows_out": 0,
        },
        "steps": {},
    }

    for i, step in enumerate(self.steps_config):
        name = step["name"]
        transformer_type = step["transformer"]
        params = step.get("params", {})
        step_metrics: dict[str, Any] = {}
        step_key = f"{i}:{name}"

        logger.info(f"Running step {i}: {name} ({transformer_type})")
        logger.debug(f"FeatureEngineer running step {i}: {name} ({transformer_type})")
        logger.debug(f"current_data type: {type(current_data)}")

        # Snapshot before for shape-delta + Winsorize value-clipping metrics
        rows_before, cols_before = get_data_stats(current_data)
        data_before = current_data

        calculator, applier = self._get_transformer_components(transformer_type)
        step_node_id = f"{node_id_prefix}_{name}"

        current_data, fitted_params, transformer_inst = self._run_step(
            transformer_type=transformer_type,
            name=name,
            calculator=calculator,
            applier=applier,
            step_node_id=step_node_id,
            current_data=current_data,
            params=params,
        )

        logger.debug(f"Step {i} complete. New data type: {type(current_data)}")

        rows_after, cols_after = get_data_stats(current_data)
        self._collect_step_metrics(
            transformer_type=transformer_type,
            fitted_params=fitted_params,
            data_before=data_before,
            current_data=current_data,
            params=params,
            rows_before=rows_before,
            cols_before=cols_before,
            rows_after=rows_after,
            cols_after=cols_after,
            name=name,
            metrics=step_metrics,
        )

        step_record = {
            "name": name,
            "transformer": transformer_type,
            "fit_time": (
                getattr(transformer_inst, "fit_time", 0.0)
                if transformer_inst is not None
                else 0.0
            ),
            "peak_memory_bytes": (
                getattr(transformer_inst, "peak_memory_bytes", 0)
                if transformer_inst is not None
                else 0
            ),
            "rows_in": (
                getattr(transformer_inst, "rows_in", rows_before)
                if transformer_inst is not None
                else rows_before
            ),
            "rows_out": (
                getattr(transformer_inst, "rows_out", rows_after)
                if transformer_inst is not None
                else rows_after
            ),
            "details": step_metrics,
        }
        metrics["steps"][step_key] = step_record

        summary = metrics["summary"]
        summary["fit_time"] += step_record["fit_time"]
        summary["peak_memory_bytes"] = max(
            summary["peak_memory_bytes"], step_record["peak_memory_bytes"]
        )
        if i == 0:
            summary["rows_in"] = step_record["rows_in"]
        summary["rows_out"] = step_record["rows_out"]

    metrics["fit_time"] = metrics["summary"]["fit_time"]
    metrics["peak_memory_bytes"] = metrics["summary"]["peak_memory_bytes"]
    metrics["rows_in"] = metrics["summary"]["rows_in"]
    metrics["rows_out"] = metrics["summary"]["rows_out"]

    return current_data, metrics

transform(data)

Apply fitted transformations to new data.

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

def transform(self, data: pd.DataFrame | SkyulfDataFrame | Any) -> Any:
    """
    Apply fitted transformations to new data.
    """
    current_data = data

    for step in self.fitted_steps:
        name = step["name"]
        transformer_type = step["type"]
        applier = step["applier"]
        artifact = step["artifact"]

        # Skip splitters during inference/transform
        if transformer_type in [
            "TrainTestSplitter",
            "feature_target_split",
            *self._RESAMPLING_TYPES,
            *self._ROW_DROPPING_TYPES,
        ]:
            continue

        logger.debug(f"Applying step: {name} ({transformer_type})")
        current_data = applier.apply(current_data, artifact)

    return current_data

FeatureEngineerFoldAdapter

Re-runs a preprocessing step chain inside each CV/tuning fold.

Parameters:

Name	Type	Description	Default
steps_config	list[dict[str, Any]]	The upstream Feature-Engineering node's step list (plain dicts, validated by validate_preprocessing_steps). Splitter steps are filtered out automatically.	required
target_column	str	Name of the target, kept to reject payloads where X still embeds it — target-aware steps (target encoders, resampling, label encoding of the target) receive y through the (X, y) payload the same way they do downstream of a real splitter.	required

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

class FeatureEngineerFoldAdapter:
    """Re-runs a preprocessing step chain inside each CV/tuning fold.

    Args:
        steps_config: The upstream Feature-Engineering node's step list
            (plain dicts, validated by ``validate_preprocessing_steps``).
            Splitter steps are filtered out automatically.
        target_column: Name of the target, kept to reject payloads where X
            still embeds it — target-aware steps (target encoders, resampling,
            label encoding of the target) receive ``y`` through the ``(X, y)``
            payload the same way they do downstream of a real splitter.
    """

    def __init__(self, steps_config: list[dict[str, Any]], target_column: str):
        self._steps_config = [
            step for step in steps_config if step.get("transformer") not in SPLITTER_STEP_TYPES
        ]
        self._target_column = target_column
        # True when any step reshapes the rows/target (resampling, row
        # drops); documents the chain's nature for callers that need it.
        self.changes_row_count = any(
            step.get("transformer") in ROW_COUNT_CHANGING_STEP_TYPES for step in self._steps_config
        )
        # Validate eagerly (unknown transformer names, bad params) so a
        # misconfigured chain fails at construction, not mid-fold.
        # validate_preprocessing_steps does not check registry membership,
        # so resolve each step's calculator explicitly.
        FeatureEngineer(self._steps_config)
        for step in self._steps_config:
            NodeRegistry.get_calculator(step["transformer"])
        self._engineer: FeatureEngineer | None = None

    def fit_transform(self, X: Any, y: Any) -> tuple[Any, Any]:
        self._validate_payload(X)
        engineer = FeatureEngineer(self._steps_config)
        transformed, _metrics = engineer.fit_transform((X, y))
        self._engineer = engineer
        return transformed

    def transform(self, X: Any, y: Any) -> tuple[Any, Any]:
        if self._engineer is None:
            raise RuntimeError("transform() called before fit_transform()")
        self._validate_payload(X)
        # At transform/inference time no step needs the target (target-aware
        # encoders use their fitted artifact; splitters/resampling are skipped).
        # Feed a bare frame when y is absent — a ``(X, None)`` tuple would trip
        # ``pack_pipeline_output``'s tuple-shape-lost diagnostic on every step.
        payload = (X, y) if y is not None else X
        transformed = self._engineer.transform(payload)
        # Some appliers return a bare frame instead of the ``(X, y)`` payload;
        # re-pair so callers always get the FoldPreprocessor protocol shape.
        if not (isinstance(transformed, tuple) and len(transformed) == 2):
            transformed = (transformed, y)
        return transformed

    def _validate_payload(self, X: Any) -> None:
        if hasattr(X, "columns") and self._target_column in X.columns:
            raise ValueError(f"target column '{self._target_column}' already present in X")

GeneralBinningCalculator

Bases: BaseCalculator

Master calculator that handles mixed strategies and per-column overrides.

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

@NodeRegistry.register("GeneralBinning", GeneralBinningApplier)
@node_meta(
    id="GeneralBinning",
    name="General Binning",
    category="Preprocessing",
    description="Bin continuous data into intervals.",
    params={"n_bins": 5, "strategy": "uniform", "columns": []},
    learns_from_data=True,
)
class GeneralBinningCalculator(BaseCalculator):
    """Master calculator that handles mixed strategies and per-column overrides."""

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> GeneralBinningArtifact:  # pylint: disable=arguments-differ
        if user_picked_no_columns(config):
            return cast(GeneralBinningArtifact, {})

        X, columns = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)

        defaults = {
            "default_n_bins": config.get("n_bins", 5),
            "n_bins": config.get("equal_width_bins", config.get("n_bins", 5)),
            "q_bins": config.get("equal_frequency_bins", config.get("n_bins", 5)),
            "duplicates": config.get("duplicates", "drop"),
        }

        valid_cols = [c for c in columns if c in X.columns]
        bin_edges_map: dict[str, list[float]] = {}
        custom_labels_map: dict[str, Any] = {}

        for col in valid_cols:
            _fit_one_column_into_maps(X, col, config, defaults, bin_edges_map, custom_labels_map)

        artifact: dict[str, Any] = {
            "type": "general_binning",
            "bin_edges": bin_edges_map,
            "custom_labels": custom_labels_map,
        }
        artifact.update(_passthrough_artifact_options(config))
        return cast(GeneralBinningArtifact, artifact)

KBinsDiscretizerCalculator

Bases: GeneralBinningCalculator

Thin wrapper around :class:GeneralBinningCalculator with kbins strategy.

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

@NodeRegistry.register("KBinsDiscretizer", KBinsDiscretizerApplier)
@node_meta(
    id="KBinsDiscretizer",
    name="K-Bins Discretizer",
    category="Preprocessing",
    description="Bin continuous data into intervals using sklearn KBinsDiscretizer.",
    params={"n_bins": 5, "encode": "ordinal", "strategy": "quantile", "columns": []},
    learns_from_data=True,
)
class KBinsDiscretizerCalculator(GeneralBinningCalculator):
    """Thin wrapper around :class:`GeneralBinningCalculator` with ``kbins`` strategy."""

    def fit(  # pylint: disable=arguments-differ
        self,
        df: pd.DataFrame | SkyulfDataFrame | tuple[Any, ...] | Any,
        config: dict[str, Any],
    ) -> GeneralBinningArtifact:
        new_config = config.copy()
        new_config["strategy"] = "kbins"
        if "n_bins" in config:
            new_config["kbins_n_bins"] = config["n_bins"]
        if "strategy" in config and config["strategy"] != "kbins":
            new_config["kbins_strategy"] = config["strategy"]
        return super().fit(df, new_config)  # pylint: disable=no-value-for-parameter

MergedBranchFoldAdapter

Re-runs fork-join preprocessing branches inside each CV/tuning fold.

Built for graphs where a shared trunk ends in a splitter (fork point) and N parallel transformer branches fan back into one training node. Per fold it re-runs every branch step list on the fold-train payload and merges the branch frames exactly like the engine's pure-strategy column-wise merge, so the fold sees the same columns the full run produces — without any pre-fit statistics leaking from outside the fold.

Parameters:

Name	Type	Description	Default
branch_step_lists	list[list[dict[str, Any]]]	One unfitted step list per branch, in the engine's merge input order.	required
merge_strategy	str	"last_wins" (default engine behaviour) or "first_wins" — must match the training node's configured strategy so fold columns match the full-run merge.	required
target_column	str	Target name, kept to reject payloads that still embed it; target-aware branch steps receive y through the payload.	required
drop_columns	list[str] | tuple[str, ...]	Columns removed upstream (e.g. by Drop Columns nodes); stripped again after each merge so a branch cannot resurrect them.	()

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

class MergedBranchFoldAdapter:
    """Re-runs fork-join preprocessing branches inside each CV/tuning fold.

    Built for graphs where a shared trunk ends in a splitter (fork point) and
    N parallel transformer branches fan back into one training node. Per fold
    it re-runs every branch step list on the fold-train payload and merges the
    branch frames exactly like the engine's pure-strategy column-wise merge,
    so the fold sees the same columns the full run produces — without any
    pre-fit statistics leaking from outside the fold.

    Args:
        branch_step_lists: One unfitted step list per branch, in the engine's
            merge input order.
        merge_strategy: ``"last_wins"`` (default engine behaviour) or
            ``"first_wins"`` — must match the training node's configured
            strategy so fold columns match the full-run merge.
        target_column: Target name, kept to reject payloads that still embed
            it; target-aware branch steps receive ``y`` through the payload.
        drop_columns: Columns removed upstream (e.g. by Drop Columns nodes);
            stripped again after each merge so a branch cannot resurrect them.
    """

    def __init__(
        self,
        branch_step_lists: list[list[dict[str, Any]]],
        merge_strategy: str,
        target_column: str,
        drop_columns: list[str] | tuple[str, ...] = (),
    ):
        if merge_strategy not in ("last_wins", "first_wins"):
            raise ValueError(f"unknown merge strategy '{merge_strategy}'")
        if not branch_step_lists:
            raise ValueError("at least one branch step list is required")
        for steps in branch_step_lists:
            if not steps:
                raise ValueError("branch step list must not be empty")
            FeatureEngineer(list(steps))
            for step in steps:
                transformer = step["transformer"]
                if transformer in UNSAFE_BRANCH_STEP_TYPES:
                    raise ValueError(
                        f"branch step '{transformer}' cannot run inside a fold: "
                        "it splits the data or changes row counts"
                    )
                NodeRegistry.get_calculator(transformer)
        self._branch_step_lists = [list(steps) for steps in branch_step_lists]
        self._merge_strategy = merge_strategy
        self._target_column = target_column
        self._drop_columns = list(drop_columns)
        self._engineers: list[FeatureEngineer] | None = None
        # Branch steps are screened against UNSAFE_BRANCH_STEP_TYPES, so the
        # merge keeps every row.
        self.changes_row_count = False

    def fit_transform(self, X: Any, y: Any) -> tuple[Any, Any]:
        self._validate_payload(X)
        engineers = [FeatureEngineer(list(steps)) for steps in self._branch_step_lists]
        frames, ys = self._run_branches(engineers, (X, y), fit=True)
        self._engineers = engineers
        return self._finalize(frames, ys)

    def transform(self, X: Any, y: Any) -> tuple[Any, Any]:
        if self._engineers is None:
            raise RuntimeError("transform() called before fit_transform()")
        self._validate_payload(X)
        payload = (X, y) if y is not None else X
        frames, ys = self._run_branches(self._engineers, payload, fit=False)
        return self._finalize(frames, ys)

    def _run_branches(
        self, engineers: list[FeatureEngineer], payload: Any, *, fit: bool
    ) -> tuple[list[pd.DataFrame], list[Any]]:
        frames: list[pd.DataFrame] = []
        ys: list[Any] = []
        input_y = payload[1] if isinstance(payload, tuple) else None
        for engineer in engineers:
            out = engineer.fit_transform(payload)[0] if fit else engineer.transform(payload)
            if isinstance(out, tuple) and len(out) == 2:
                frame, y_out = out
            else:
                # Appliers that return a bare frame pass the target through.
                frame, y_out = out, input_y
            if isinstance(frame, pl.DataFrame):
                frame = frame.to_pandas()
            frames.append(frame)
            ys.append(y_out)
        return frames, ys

    def _finalize(self, frames: list[pd.DataFrame], ys: list[Any]) -> tuple[Any, Any]:
        merged = _merge_branch_frames_columnwise(frames, self._merge_strategy)
        drop = [col for col in self._drop_columns if col in merged.columns]
        if drop:
            merged = merged.drop(columns=drop)
        # Mirrors the engine's SplitDataset merge: the first branch supplies
        # the target (branches are screened to never change row counts).
        return merged, ys[0]

    def _validate_payload(self, X: Any) -> None:
        if hasattr(X, "columns") and self._target_column in X.columns:
            raise ValueError(f"target column '{self._target_column}' already present in X")

SchemaMismatchError

Bases: ValueError

Raised when an actual frame schema violates an expected SkyulfSchema.

Carries structured details so callers can render a precise message instead of a generic KeyError deep inside a transformer:

Attributes:

Name	Type	Description
missing		Expected columns absent from the actual frame.
unexpected		Actual columns not present in the expected schema.
dtype_mismatches		{column: (expected_dtype, actual_dtype)} for shared columns whose dtype labels differ (only when dtype checking is requested).
order_mismatch		True when the shared columns appear in a different relative order than expected (only when order checking is requested).

Source code in skyulf-core/skyulf/core/schema.py

class SchemaMismatchError(ValueError):
    """Raised when an actual frame schema violates an expected ``SkyulfSchema``.

    Carries structured details so callers can render a precise message instead
    of a generic ``KeyError`` deep inside a transformer:

    Attributes:
        missing: Expected columns absent from the actual frame.
        unexpected: Actual columns not present in the expected schema.
        dtype_mismatches: ``{column: (expected_dtype, actual_dtype)}`` for
            shared columns whose dtype labels differ (only when dtype checking
            is requested).
        order_mismatch: ``True`` when the shared columns appear in a different
            relative order than expected (only when order checking is requested).
    """

    def __init__(
        self,
        message: str,
        *,
        missing: list[str] | None = None,
        unexpected: list[str] | None = None,
        dtype_mismatches: dict[str, tuple[str, str]] | None = None,
        order_mismatch: bool = False,
    ) -> None:
        super().__init__(message)
        self.missing = missing or []
        self.unexpected = unexpected or []
        self.dtype_mismatches = dtype_mismatches or {}
        self.order_mismatch = order_mismatch

SimpleImputerApplier

Bases: BaseApplier

Apply fitted Simple Imputer fill values to missing values in selected columns.

The calculator artifact records per-column values for mean, median, most_frequent (also accepted as mode), or constant strategies. Missing columns seen during fitting are restored with their stored value.

Source code in skyulf-core/skyulf/preprocessing/imputation/simple.py

class SimpleImputerApplier(BaseApplier):
    """Apply fitted Simple Imputer fill values to missing values in selected columns.

    The calculator artifact records per-column values for ``mean``, ``median``,
    ``most_frequent`` (also accepted as ``mode``), or ``constant`` strategies.
    Missing columns seen during fitting are restored with their stored value.
    """

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        return apply_dual_engine(X, params, self._apply_polars, self._apply_pandas)

    @staticmethod
    def _apply_polars(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        fill_values = params.get("fill_values", {})
        if not cols:
            return X, _y

        exprs: list[Any] = []
        for col in X.columns:
            if col in cols and col in fill_values:
                expr = pl.col(col).fill_null(fill_values[col])
                if X.schema[col].is_float():
                    expr = expr.fill_nan(fill_values[col])
                exprs.append(expr.alias(col))
            else:
                exprs.append(pl.col(col))

        # Restore columns that were present at fit time but missing in input X.
        exprs.extend(
            pl.lit(fill_values[col]).alias(col)
            for col in cols
            if col not in X.columns and col in fill_values
        )

        return X.select(exprs), _y

    @staticmethod
    def _apply_pandas(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        fill_values = params.get("fill_values", {})
        if not cols:
            return X, _y

        X_out = X.copy()
        for col in cols:
            val = fill_values.get(col)
            if val is None:
                continue
            if col not in X_out.columns:
                X_out[col] = val
            else:
                series = X_out[col]
                # Nullable numeric extension dtypes (Int64...) refuse a float
                # fill value; upcast like the Polars fill does (F-10).
                if (
                    isinstance(val, float)
                    and isinstance(series.dtype, pd.api.extensions.ExtensionDtype)
                    and pd.api.types.is_numeric_dtype(series.dtype)
                ):
                    series = series.astype("float64")
                X_out[col] = series.fillna(val)
        return X_out, _y

SimpleImputerCalculator

Bases: BaseCalculator

Fit per-column values for filling missing data with sklearn-compatible strategies.

Supported strategy values are mean, median, most_frequent, and constant; mode is normalized to most_frequent. Use columns to select columns and fill_value with constant. Mean and median operate on numeric columns, while the other strategies can operate on all selected columns.

Source code in skyulf-core/skyulf/preprocessing/imputation/simple.py

@NodeRegistry.register("SimpleImputer", SimpleImputerApplier)
@node_meta(
    id="SimpleImputer",
    name="Simple Imputer",
    category="Preprocessing",
    description="Imputes missing values using mean, median, or constant.",
    params={"strategy": "mean", "fill_value": None, "columns": []},
    learns_from_data=True,
)
class SimpleImputerCalculator(BaseCalculator):
    """Fit per-column values for filling missing data with sklearn-compatible strategies.

    Supported ``strategy`` values are ``mean``, ``median``, ``most_frequent``,
    and ``constant``; ``mode`` is normalized to ``most_frequent``. Use
    ``columns`` to select columns and ``fill_value`` with ``constant``.
    Mean and median operate on numeric columns, while the other strategies can
    operate on all selected columns.
    """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        # Imputers fill NaNs in place; column set and order are preserved.
        return input_schema

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> SimpleImputerArtifact:  # pylint: disable=arguments-differ
        if user_picked_no_columns(config):
            return {}

        strategy = config.get("strategy", "mean")
        if strategy == "mode":
            strategy = "most_frequent"
        fill_value = config.get("fill_value")

        cols = _resolve_simple_columns(X, config, strategy)
        if not cols:
            return {}

        # Stash resolved-once values into params so dispatched fits don't redo work.
        merged = dict(config)
        merged["_resolved_strategy"] = strategy
        merged["_resolved_cols"] = cols
        merged["_resolved_fill_value"] = fill_value

        return cast(
            SimpleImputerArtifact,
            fit_dual_engine(X, merged, self._fit_polars, self._fit_pandas),
        )

    @staticmethod
    def _fit_polars(X: Any, _y: Any, params: dict[str, Any]) -> dict[str, Any]:
        cols: list[str] = params["_resolved_cols"]
        strategy: str = params["_resolved_strategy"]
        fill_value = params["_resolved_fill_value"]

        fill_values = _compute_polars_fill_values(X, cols, strategy, fill_value)
        missing_counts, total_missing = _polars_missing_counts(X, cols)

        return {
            "type": "simple_imputer",
            "strategy": strategy,
            "fill_values": fill_values,
            "columns": cols,
            "missing_counts": missing_counts,
            "total_missing": total_missing,
        }

    @staticmethod
    def _fit_pandas(X: Any, _y: Any, params: dict[str, Any]) -> dict[str, Any]:
        cols: list[str] = params["_resolved_cols"]
        strategy: str = params["_resolved_strategy"]
        fill_value = params["_resolved_fill_value"]

        # Mean/median: extra safety filter to numeric columns only.
        if strategy in ("mean", "median"):
            numeric = set(detect_numeric_columns(X))
            cols = [c for c in cols if c in numeric]
            if not cols:
                return {}

        imputer = SimpleImputer(strategy=strategy, fill_value=fill_value)
        imputer.fit(X[cols])

        statistics = imputer.statistics_.tolist()
        fill_values = dict(zip(cols, statistics, strict=True))
        missing_counts = X[cols].isnull().sum().to_dict()
        total_missing = int(sum(missing_counts.values()))

        return {
            "type": "simple_imputer",
            "strategy": strategy,
            "fill_values": fill_values,
            "columns": cols,
            "missing_counts": missing_counts,
            "total_missing": total_missing,
        }

SkyulfSchema dataclass

Immutable schema description.

Attributes:

Name	Type	Description
columns	tuple[str, ...]	Ordered list of column names.
dtypes	dict[str, str]	Mapping of column name → string dtype label (engine-agnostic; e.g. "int64", "float64", "string", "category", "datetime", "bool", or "unknown"). A column may be present in columns but absent from dtypes when its type is unknown.

Source code in skyulf-core/skyulf/core/schema.py

@dataclass(frozen=True)
class SkyulfSchema:
    """Immutable schema description.

    Attributes:
        columns: Ordered list of column names.
        dtypes: Mapping of column name → string dtype label
            (engine-agnostic; e.g. ``"int64"``, ``"float64"``, ``"string"``,
            ``"category"``, ``"datetime"``, ``"bool"``, or ``"unknown"``).
            A column may be present in ``columns`` but absent from
            ``dtypes`` when its type is unknown.
    """

    columns: tuple[str, ...]
    dtypes: dict[str, str] = field(default_factory=dict)

    # ---- Constructors -----------------------------------------------------

    @classmethod
    def from_columns(
        cls, columns: Iterable[str], dtypes: dict[str, str] | None = None
    ) -> "SkyulfSchema":
        cols = tuple(columns)
        return cls(columns=cols, dtypes=dict(dtypes or {}))

    @classmethod
    def from_dataframe(cls, df: Any) -> "SkyulfSchema":
        """Best-effort schema extraction from a Pandas/Polars/Wrapper frame."""
        raw_cols = getattr(df, "columns", None)
        cols = list(raw_cols) if raw_cols is not None else []
        dtypes = _extract_pandas_dtypes(df)
        if not dtypes:
            dtypes = _extract_polars_dtypes(df)
        return cls(columns=tuple(cols), dtypes=dtypes)

    # ---- Mutations (return new instances) ---------------------------------

    def drop(self, names: Iterable[str]) -> "SkyulfSchema":
        drop_set = set(names)
        new_cols = tuple(c for c in self.columns if c not in drop_set)
        new_dtypes = {k: v for k, v in self.dtypes.items() if k not in drop_set}
        return replace(self, columns=new_cols, dtypes=new_dtypes)

    def add(self, name: str, dtype: str = "unknown") -> "SkyulfSchema":
        if name in self.columns:
            return self
        new_dtypes = dict(self.dtypes)
        new_dtypes[name] = dtype
        return replace(self, columns=self.columns + (name,), dtypes=new_dtypes)

    def rename(self, mapping: dict[str, str]) -> "SkyulfSchema":
        new_cols = tuple(mapping.get(c, c) for c in self.columns)
        if len(set(new_cols)) != len(new_cols):
            seen: set[str] = set()
            collisions: set[str] = set()
            for c in new_cols:
                if c in seen:
                    collisions.add(c)
                seen.add(c)
            raise ValueError(
                "Schema rename() would produce duplicate column name(s): "
                f"{sorted(collisions)}. Rename mapping: {mapping}"
            )
        new_dtypes: dict[str, str] = {}
        for k, v in self.dtypes.items():
            new_dtypes[mapping.get(k, k)] = v
        return replace(self, columns=new_cols, dtypes=new_dtypes)

    def with_dtype(self, name: str, dtype: str) -> "SkyulfSchema":
        if name not in self.columns:
            return self
        new_dtypes = dict(self.dtypes)
        new_dtypes[name] = dtype
        return replace(self, dtypes=new_dtypes)

    # ---- Queries ----------------------------------------------------------

    def has(self, name: str) -> bool:
        return name in self.columns

    def column_list(self) -> list[str]:
        return list(self.columns)

    def __contains__(self, item: object) -> bool:
        return item in self.columns

    def __len__(self) -> int:
        return len(self.columns)

    # ---- Contract validation ---------------------------------------------

    def assert_compatible(
        self,
        actual: "SkyulfSchema",
        *,
        check_dtypes: bool = False,
        check_order: bool = False,
        where: str = "input",
    ) -> None:
        """Validate that ``actual`` satisfies this (expected) schema.

        ``self`` is the expected schema (e.g. what an Applier was fitted on);
        ``actual`` is the schema observed at apply time. Raises
        :class:`SchemaMismatchError` describing every discrepancy. Presence of
        the expected columns is always checked; dtype and column-order checks
        are opt-in to keep the default contract permissive and non-breaking.

        Args:
            actual: The schema observed at runtime.
            check_dtypes: Also compare dtype labels for shared columns.
            check_order: Also require shared columns in the same relative order.
            where: Label used in the error message (e.g. ``"input"``).
        """
        missing, unexpected = _presence_diff(self, actual)
        dtype_mismatches = _dtype_mismatches(self, actual) if check_dtypes else {}
        order_mismatch = _check_order(self, actual, check_order, missing)

        if missing or unexpected or dtype_mismatches or order_mismatch:
            raise SchemaMismatchError(
                _format_mismatch(where, missing, unexpected, dtype_mismatches, order_mismatch),
                missing=missing,
                unexpected=unexpected,
                dtype_mismatches=dtype_mismatches,
                order_mismatch=order_mismatch,
            )

assert_compatible(actual, *, check_dtypes=False, check_order=False, where='input')

Validate that actual satisfies this (expected) schema.

self is the expected schema (e.g. what an Applier was fitted on); actual is the schema observed at apply time. Raises :class:SchemaMismatchError describing every discrepancy. Presence of the expected columns is always checked; dtype and column-order checks are opt-in to keep the default contract permissive and non-breaking.

Parameters:

Name	Type	Description	Default
actual	SkyulfSchema	The schema observed at runtime.	required
check_dtypes	bool	Also compare dtype labels for shared columns.	False
check_order	bool	Also require shared columns in the same relative order.	False
where	str	Label used in the error message (e.g. "input").	'input'

Source code in skyulf-core/skyulf/core/schema.py

def assert_compatible(
    self,
    actual: "SkyulfSchema",
    *,
    check_dtypes: bool = False,
    check_order: bool = False,
    where: str = "input",
) -> None:
    """Validate that ``actual`` satisfies this (expected) schema.

    ``self`` is the expected schema (e.g. what an Applier was fitted on);
    ``actual`` is the schema observed at apply time. Raises
    :class:`SchemaMismatchError` describing every discrepancy. Presence of
    the expected columns is always checked; dtype and column-order checks
    are opt-in to keep the default contract permissive and non-breaking.

    Args:
        actual: The schema observed at runtime.
        check_dtypes: Also compare dtype labels for shared columns.
        check_order: Also require shared columns in the same relative order.
        where: Label used in the error message (e.g. ``"input"``).
    """
    missing, unexpected = _presence_diff(self, actual)
    dtype_mismatches = _dtype_mismatches(self, actual) if check_dtypes else {}
    order_mismatch = _check_order(self, actual, check_order, missing)

    if missing or unexpected or dtype_mismatches or order_mismatch:
        raise SchemaMismatchError(
            _format_mismatch(where, missing, unexpected, dtype_mismatches, order_mismatch),
            missing=missing,
            unexpected=unexpected,
            dtype_mismatches=dtype_mismatches,
            order_mismatch=order_mismatch,
        )

from_dataframe(df) classmethod

Best-effort schema extraction from a Pandas/Polars/Wrapper frame.

Source code in skyulf-core/skyulf/core/schema.py

@classmethod
def from_dataframe(cls, df: Any) -> "SkyulfSchema":
    """Best-effort schema extraction from a Pandas/Polars/Wrapper frame."""
    raw_cols = getattr(df, "columns", None)
    cols = list(raw_cols) if raw_cols is not None else []
    dtypes = _extract_pandas_dtypes(df)
    if not dtypes:
        dtypes = _extract_polars_dtypes(df)
    return cls(columns=tuple(cols), dtypes=dtypes)

StatefulTransformer

Fits + applies one pipeline step.

Accepts anything satisfying :class:~skyulf.core.protocols.CalculatorProtocol / :class:~skyulf.core.protocols.ApplierProtocol (structural typing) — a BaseCalculator/BaseApplier subclass, or any duck-typed object exposing matching fit/apply methods, works without subclassing.

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

class StatefulTransformer:
    """Fits + applies one pipeline step.

    Accepts anything satisfying :class:`~skyulf.core.protocols.CalculatorProtocol` /
    :class:`~skyulf.core.protocols.ApplierProtocol` (structural typing) — a
    ``BaseCalculator``/``BaseApplier`` subclass, or any duck-typed object
    exposing matching ``fit``/``apply`` methods, works without subclassing.
    """

    def __init__(
        self,
        calculator: CalculatorProtocol,
        applier: ApplierProtocol,
        node_id: str,
        apply_on_test: bool = True,
        apply_on_validation: bool = True,
    ):
        self.calculator = calculator
        self.applier = applier
        self.node_id = node_id
        self.apply_on_test = apply_on_test
        self.apply_on_validation = apply_on_validation
        self.params: dict[str, Any] = {}  # Store params in memory instead of ArtifactStore
        # Profiling metrics
        self.fit_time: float = 0.0
        self.peak_memory_bytes: int = 0
        self.rows_in: int = 0
        self.rows_out: int = 0

    def fit_transform(
        self,
        dataset: SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple,
        config: dict[str, Any],
    ) -> SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple:
        self.rows_in, _ = get_data_stats(dataset)
        tracing_was_active = tracemalloc.is_tracing()
        if not tracing_was_active:
            tracemalloc.start()
            tracemalloc.reset_peak()
        peak_baseline = tracemalloc.get_traced_memory()[1] if tracemalloc.is_tracing() else 0
        # With caller-owned tracing we can only report new global peak growth since entry.
        self.peak_memory_bytes = 0
        self.rows_out = 0
        start = time.time()

        try:
            result = self._fit_transform_inner(dataset, config)
            self.rows_out, _ = get_data_stats(result)
            return result
        finally:
            self.fit_time = time.time() - start
            if tracemalloc.is_tracing():
                _, peak = tracemalloc.get_traced_memory()
                self.peak_memory_bytes = max(0, peak - peak_baseline)
            if not tracing_was_active and tracemalloc.is_tracing():
                tracemalloc.stop()

    def _fit_and_apply_training_data(
        self,
        data: Any,
        config: dict[str, Any],
        *,
        guard_split_output: bool = True,
    ) -> Any:
        """Fit one training input and return its train-time representation."""
        if isinstance(self.calculator, TrainTransformCalculatorProtocol):
            params, transformed = self.calculator.fit_transform_train(data, config)
            self.params = cast(dict[str, Any], params)
            if isinstance(transformed, SplitDataset):
                raise TypeError(
                    "Calculator returned SplitDataset from fit_transform_train, which is not supported."
                )
            return transformed

        self.params = cast(dict[str, Any], self.calculator.fit(data, config))
        if guard_split_output:
            return self._apply_guarded(data, self.params)
        return self.applier.apply(data, self.params)

    def _fit_transform_inner(
        self,
        dataset: SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple,
        config: dict[str, Any],
    ) -> SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple:
        # Check for DataFrame-like (Pandas, Polars, Wrapper)
        if (
            hasattr(dataset, "shape")
            and hasattr(dataset, "columns")
            and not isinstance(dataset, tuple)
        ):
            # Fit on the whole dataframe (be careful about leakage!)
            # ty can't narrow a Union through hasattr — cast once for both calls.
            frame = cast(Any, dataset)
            return self._fit_and_apply_training_data(frame, config, guard_split_output=False)

        # If dataset is a tuple (e.g. from FeatureTargetSplitter), pass it through.
        # This allows nodes like TrainTestSplitter to accept (X, y) tuples.
        if isinstance(dataset, tuple):
            return self._fit_and_apply_training_data(dataset, config, guard_split_output=False)

        # 1. Calculate on Train
        new_train = self._fit_and_apply_training_data(dataset.train, config)

        # 2. Apply fitted params to held-out splits only
        new_test = dataset.test
        if self.apply_on_test:
            new_test = self._apply_guarded(dataset.test, self.params)

        new_val = dataset.validation
        if self.apply_on_validation and dataset.validation is not None:
            new_val = self._apply_guarded(dataset.validation, self.params)

        return SplitDataset(train=new_train, test=new_test, validation=new_val)

    def _apply_guarded(self, data: Any, params: dict[str, Any]) -> Any:
        """Apply the applier to `data` and raise if it produces a nested SplitDataset."""
        result = self.applier.apply(data, params)
        if isinstance(result, SplitDataset):
            raise TypeError(
                "Applier returned SplitDataset inside StatefulTransformer, which is not supported."
            )
        return result

    def _apply_to_split_dataset(
        self, dataset: SplitDataset, params: dict[str, Any]
    ) -> SplitDataset:
        """Apply the applier to each split (train/test/validation) of a SplitDataset."""
        new_train = self._apply_guarded(dataset.train, params)

        new_test = dataset.test
        if self.apply_on_test:
            new_test = self._apply_guarded(dataset.test, params)

        new_val = dataset.validation
        if self.apply_on_validation and dataset.validation is not None:
            new_val = self._apply_guarded(dataset.validation, params)

        return SplitDataset(train=new_train, test=new_test, validation=new_val)

    def transform(
        self, dataset: SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple
    ) -> SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple:
        # Use stored params
        params = self.params

        # Check for DataFrame-like (Pandas, Polars, Wrapper) input, mirroring
        # `_fit_transform_inner`'s detection above -- an `isinstance(dataset,
        # pd.DataFrame)`-only check here would misroute a raw (unwrapped)
        # `pl.DataFrame` into the SplitDataset branch below, crashing with
        # `AttributeError: 'DataFrame' object has no attribute 'train'`.
        if (
            hasattr(dataset, "shape")
            and hasattr(dataset, "columns")
            and not isinstance(dataset, tuple)
        ):
            return self.applier.apply(cast(Any, dataset), params)

        if isinstance(dataset, tuple):
            return self.applier.apply(dataset, params)

        # 2. Apply
        # ty can't narrow SplitDataset out of the SkyulfDataFrame branch of this
        # Union via isinstance alone (mirrors the `frame = cast(Any, dataset)`
        # note in `_fit_transform_inner` above).
        return self._apply_to_split_dataset(cast(SplitDataset, dataset), params)

TargetEncoderCalculator

Bases: BaseCalculator

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

@NodeRegistry.register("TargetEncoder", TargetEncoderApplier)
@node_meta(
    id="TargetEncoder",
    name="Target Encoder",
    category="Preprocessing",
    description="Encode categorical features using target statistics.",
    params={"smooth": "auto", "target_type": "auto", "columns": []},
    learns_from_data=True,
)
class TargetEncoderCalculator(BaseCalculator):
    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> TargetEncoderArtifact:  # pylint: disable=arguments-differ
        if user_picked_no_columns(config):
            return {}
        return cast(
            TargetEncoderArtifact,
            fit_dual_engine(
                (X, y) if y is not None else X,
                config,
                polars_func=_target_fit_polars,
                pandas_func=_target_fit_pandas,
            ),
        )

    def fit_transform_train(
        self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
    ) -> tuple[TargetEncoderArtifact, Any]:
        """Fit sklearn TargetEncoder and cross-fit the pipeline training rows."""
        if user_picked_no_columns(config):
            return {}, df

        artifact, transformed = fit_transform_train_dual_engine(
            df,
            config,
            polars_func=_target_fit_transform_train_polars,
            pandas_func=_target_fit_transform_train_pandas,
        )
        return cast(
            TargetEncoderArtifact,
            artifact,
        ), transformed

    def infer_output_schema(
        self,
        input_schema: SkyulfSchema,
        config: dict[str, Any],
    ) -> SkyulfSchema | None:
        # For binary/regression targets, the encoder replaces values in
        # source columns in place — same column names, dtype becomes float
        # (per-column dtype is best-effort so we don't bother rewriting it).
        #
        # For multiclass targets, the apply logic (see
        # ``_target_apply_polars``/``_target_apply_pandas``) drops the
        # original columns and creates ``{col}_cls{i}`` columns instead — the
        # number of classes is data-dependent and unknown here, so we can't
        # confidently predict the output columns. The default/"auto"
        # target_type is resolved to multiclass at fit time whenever y has
        # more than two classes, so we must also treat "auto" as unknown
        # rather than assuming binary/regression. Only the explicit
        # "binary"/"regression" config values are confidently in-place.
        if config.get("target_type", "auto") not in ("binary", "regression"):
            return None
        return input_schema

fit_transform_train(df, config)

Fit sklearn TargetEncoder and cross-fit the pipeline training rows.

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

def fit_transform_train(
    self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
) -> tuple[TargetEncoderArtifact, Any]:
    """Fit sklearn TargetEncoder and cross-fit the pipeline training rows."""
    if user_picked_no_columns(config):
        return {}, df

    artifact, transformed = fit_transform_train_dual_engine(
        df,
        config,
        polars_func=_target_fit_transform_train_polars,
        pandas_func=_target_fit_transform_train_pandas,
    )
    return cast(
        TargetEncoderArtifact,
        artifact,
    ), transformed

frame_rows(frame)

Row count for pandas/polars frames and numpy arrays (-1 if unknowable).

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

def frame_rows(frame: Any) -> int:
    """Row count for pandas/polars frames and numpy arrays (-1 if unknowable)."""
    height = getattr(frame, "height", None)
    if isinstance(height, int):
        return height
    try:
        return len(frame)
    except TypeError:
        return -1

validate_schema(expected, actual, *, check_dtypes=False, check_order=False, where='input')

Validate a live DataFrame against an expected schema.

Thin convenience wrapper: builds a :class:SkyulfSchema from actual (Pandas/Polars/wrapper frame) and delegates to :meth:SkyulfSchema.assert_compatible. Raises :class:SchemaMismatchError on any violation.

Source code in skyulf-core/skyulf/core/schema.py

def validate_schema(
    expected: SkyulfSchema,
    actual: Any,
    *,
    check_dtypes: bool = False,
    check_order: bool = False,
    where: str = "input",
) -> None:
    """Validate a live DataFrame against an ``expected`` schema.

    Thin convenience wrapper: builds a :class:`SkyulfSchema` from ``actual``
    (Pandas/Polars/wrapper frame) and delegates to
    :meth:`SkyulfSchema.assert_compatible`. Raises
    :class:`SchemaMismatchError` on any violation.
    """
    actual_schema = (
        actual if isinstance(actual, SkyulfSchema) else SkyulfSchema.from_dataframe(actual)
    )
    expected.assert_compatible(
        actual_schema,
        check_dtypes=check_dtypes,
        check_order=check_order,
        where=where,
    )