API: preprocessing.pipeline

skyulf.preprocessing.pipeline

Feature Engineering Pipeline Orchestrator.

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"}

    # Explicit filters apply to training and evaluation. Running them at
    # unkeyed 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",
        "RowFilterFunction",
    }

    def __init__(
        self,
        steps_config: Sequence[PreprocessingStepConfig | dict[str, Any]],
        *,
        _validated: bool = False,
        frame_spec: FrameSpec | None = None,
        execution_options: ExecutionOptions | None = None,
    ):
        """Store the ordered step configuration, validating it unless the caller already did.

        Args:
            steps_config: Pipeline steps in execution order, as dicts or
                ``PreprocessingStepConfig`` objects.
            _validated: Skip ``validate_preprocessing_steps`` when ``True``; set by
                callers such as ``SkyulfPipeline`` that already validated the same
                structural rules via ``validate_pipeline_config``.
            frame_spec: Explicit keys and optional target for a single Spark frame.
            execution_options: Opt-in engine selection; must match the input engine.
        """
        # `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]] = []
        if frame_spec is not None and not isinstance(frame_spec, FrameSpec):
            raise TypeError("frame_spec must be FrameSpec.")
        if execution_options is not None and not isinstance(execution_options, ExecutionOptions):
            raise TypeError("execution_options must be ExecutionOptions.")
        self.frame_spec = frame_spec
        self.execution_options = execution_options
        self._spark_fitted = False
        self._portable_fitted = False

    def export_state(self) -> bytes:
        """Export supported fitted steps as bounded, versioned UTF-8 JSON bytes.

        SimpleImputer mean/constant and StandardScaler are supported. Learned
        column selection is frozen; row keys, target and runtime objects are
        supplied again when loading. The complete payload must fit the execution
        state byte budget (8 MiB by default). Callers own storage and transport.
        """
        return export_feature_state(self)

    @classmethod
    def from_state(
        cls,
        payload: bytes,
        *,
        frame_spec: FrameSpec | None = None,
        execution_options: ExecutionOptions | None = None,
    ) -> FeatureEngineer:
        """Restore supported fitted steps without fitting, I/O or creating a Spark session.

        Spark requires explicit frame_spec and execution_options. Local callers
        may omit both and let input frames select pandas or Polars. Malformed,
        oversized, unsupported or corrupt envelopes are rejected before execution.
        """
        configs, records = restore_feature_state(payload, execution_options, frame_spec)
        engineer = cls(configs, frame_spec=frame_spec, execution_options=execution_options)
        engineer.fitted_steps = records
        engineer._portable_fitted = True
        engineer._spark_fitted = (
            execution_options is not None and execution_options.engine == "spark"
        )
        return engineer

    def transform(
        self, data: pd.DataFrame | SkyulfDataFrame | Any, *, preserve_rows: bool = False
    ) -> Any:
        """Apply transformations, optionally rejecting row-count changes and temporal sorting.

        Local prediction callers set ``preserve_rows=True`` because their responses
        have no input-row provenance. Each applied step is checked immediately;
        a later step cannot conceal filtering, expansion, or a built-in temporal
        permutation. Ordinary transform and fold scoring retain their configured
        filtering and sorting behavior. Custom appliers receive count checks only.
        Spark execution always preserves key identity and optional target values;
        physical row order is not guaranteed. Consumers must align by row keys.
        """
        spec = getattr(self, "frame_spec", None)
        if use_spark(data, getattr(self, "execution_options", None), spec):
            if not getattr(self, "_spark_fitted", False):
                raise ValueError("Spark FeatureEngineer must be fitted before transform.")
            assert spec is not None
            assert self.execution_options is not None
            return transform_spark(
                data,
                self.fitted_steps,
                spec,
                state_max_bytes=self.execution_options.state_max_bytes,
            )
        current_data = data
        input_rows = prediction_row_count(unpack_pipeline_input(data)[0])
        coverage_steps: list[dict[str, Any]] = []

        for step in self._transform_steps(preserve_rows=preserve_rows):
            name = step["name"]
            transformer_type = step["type"]
            applier = step["applier"]
            artifact = step["artifact"]

            logger.debug(f"Applying step: {name} ({transformer_type})")
            rows_before = prediction_row_count(unpack_pipeline_input(current_data)[0])
            current_data = (
                _apply_prediction_step(current_data, step)
                if preserve_rows
                else applier.apply(current_data, artifact)
            )
            if not preserve_rows:
                coverage_steps.append(
                    {
                        "name": name,
                        "transformer": transformer_type,
                        **record_coverage(
                            rows_before,
                            prediction_row_count(unpack_pipeline_input(current_data)[0]),
                            step_name=f"{name} ({transformer_type})",
                        ),
                    }
                )

        if not preserve_rows:
            self.last_transform_coverage_ = record_coverage(
                input_rows,
                prediction_row_count(unpack_pipeline_input(current_data)[0]),
                coverage_steps,
            )
        return current_data

    def _transform_steps(self, *, preserve_rows: bool = True) -> list[dict[str, Any]]:
        """Apply explicit evaluation filters while retaining unkeyed prediction skips."""
        skipped = {
            "TrainTestSplitter",
            "Split",
            "feature_target_split",
            *self._RESAMPLING_TYPES,
        }
        if preserve_rows:
            skipped.update(self._ROW_DROPPING_TYPES)
        return [step for step in self.fitted_steps if step["type"] not in skipped]

    def transform_tracking_order(self, data: Any) -> tuple[Any, np.ndarray]:
        """Transform local features and report each output row's original input position.

        Only built-in temporal permutations are tracked. Row-count changes fail;
        ordinary serving continues to reject permutations via preserve_rows=True.
        """
        if use_spark(
            data, getattr(self, "execution_options", None), getattr(self, "frame_spec", None)
        ):
            raise ValueError("Positional prediction tracking requires a local frame.")
        positions = np.arange(prediction_row_count(data))
        for step in self._transform_steps():
            data, local_positions = _apply_tracked_step(data, step)
            if local_positions is not None:
                positions = positions[local_positions]
        return data, positions

    def fit_transform(
        self,
        data: pd.DataFrame | SkyulfDataFrame | Any,
        node_id_prefix="",
        *,
        target_column: str | None = None,
        on_split: Callable[[SplitDataset], None] | None = None,
        sample_weight: Any = None,
    ) -> Any:
        """Runs the pipeline on data.

        Args:
            data: Input frame, feature-target pair, or existing split dataset.
            node_id_prefix (str): Prefix for the per-step transformer identifiers.
            target_column: Execution target excluded from automatic feature selection.
            sample_weight: Optional raw positional weights, sliced by the actual
                splitter and stored in the resulting training split.
            on_split: Optional synchronous callback receiving the first actual row
                split before later transformations run. Skipped splitters and
                feature-target separation do not invoke it. The callback is not
                retained; consumers must copy or persist the payload immediately
                if they need a snapshot. Callback errors propagate to the caller.

        Returns:
            A pair containing the transformed data and per-step metrics.

        Spark execution validates all capabilities before any data action. Key
        uniqueness and per-step identity checks run distributed queries with
        bounded results; local row counts and memory metrics remain unknown.
        """
        raw_weights = prepare_pipeline_weights(data, sample_weight, self.steps_config)
        spec = getattr(self, "frame_spec", None)
        if use_spark(data, getattr(self, "execution_options", None), spec):
            return self._fit_transform_spark(data, spec, target_column, on_split)
        self._portable_fitted = False
        self.fitted_steps = []  # Reset fitted steps
        current_data = data
        split_captured = False
        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 = dict(step.get("params", {}))
            if target_column is not None:
                # A raw-frame target must not become an auto-selected feature.
                # Keep the caller's config untouched and use pipeline context.
                params["target_column"] = target_column
            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,
                sample_weight=raw_weights,
            )

            raw_weights = _step_weights(current_data, transformer_inst, raw_weights)
            record_target_labels(
                data_before, current_data, target_column, transformer_type, fitted_params
            )

            if (
                on_split is not None
                and not split_captured
                and transformer_type in {"TrainTestSplitter", "Split"}
                and not isinstance(data_before, SplitDataset)
                and isinstance(current_data, SplitDataset)
            ):
                split_captured = True
                on_split(current_data)

            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 = self._step_metric_record(
                name, transformer_type, transformer_inst, rows_before, rows_after, 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"]

        self._portable_fitted = True
        self.train_sample_weight_ = raw_weights
        self.evaluation_coverage_ = deepcopy(getattr(current_data, "evaluation_coverage", {}))
        return current_data, metrics

    def _fit_transform_spark(
        self,
        data: Any,
        spec: FrameSpec | None,
        target_column: str | None,
        on_split: Callable[[SplitDataset], None] | None,
    ) -> tuple[Any, dict[str, Any]]:
        """Fit one Spark frame and publish learned records after native execution succeeds."""
        assert spec is not None
        if target_column is not None and target_column != spec.target:
            raise ValueError("target_column conflicts with frame_spec.target.")
        if on_split is not None:
            raise ValueError("Spark FE uses one frame; on_split is unsupported.")
        assert self.execution_options is not None
        result, metrics, records = fit_spark(
            data,
            self.steps_config,
            spec,
            state_max_bytes=self.execution_options.state_max_bytes,
        )
        self.fitted_steps = records
        self._spark_fitted = True
        self._portable_fitted = True
        return result, metrics

    @staticmethod
    def _step_metric_record(
        name: str,
        transformer_type: str,
        transformer_inst: Any,
        rows_before: int,
        rows_after: int,
        details: dict[str, Any],
    ) -> dict[str, Any]:
        """Capture a step's measurements, falling back to observed row counts."""
        return {
            "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": details,
        }

    # ------------------------------------------------------------------
    # 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],
        sample_weight: Any = None,
    ) -> 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 in {"TrainTestSplitter", "Split"}:
            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)
                weight_kwargs = {} if sample_weight is None else {"sample_weight": sample_weight}
                current_data = applier.apply(current_data, params, **weight_kwargs)
            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")
        if sample_weight is not None and not isinstance(current_data, SplitDataset):
            current_data, sample_weight = transformer.fit_transform_weighted(
                current_data, params, sample_weight
            )
        else:
            current_data = transformer.fit_transform(current_data, params)
        transformer.train_sample_weight_ = sample_weight
        fitted_params = transformer.params
        self.fitted_steps.append(
            {
                "name": name,
                "type": transformer_type,
                "applier": applier,
                "artifact": fitted_params,
                "params": dict(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

__init__(steps_config, *, _validated=False, frame_spec=None, execution_options=None)

Store the ordered step configuration, validating it unless the caller already did.

Parameters:

Name	Type	Description	Default
steps_config	Sequence[PreprocessingStepConfig | dict[str, Any]]	Pipeline steps in execution order, as dicts or PreprocessingStepConfig objects.	required
_validated	bool	Skip validate_preprocessing_steps when True; set by callers such as SkyulfPipeline that already validated the same structural rules via validate_pipeline_config.	False
frame_spec	FrameSpec | None	Explicit keys and optional target for a single Spark frame.	None
execution_options	ExecutionOptions | None	Opt-in engine selection; must match the input engine.	None

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

def __init__(
    self,
    steps_config: Sequence[PreprocessingStepConfig | dict[str, Any]],
    *,
    _validated: bool = False,
    frame_spec: FrameSpec | None = None,
    execution_options: ExecutionOptions | None = None,
):
    """Store the ordered step configuration, validating it unless the caller already did.

    Args:
        steps_config: Pipeline steps in execution order, as dicts or
            ``PreprocessingStepConfig`` objects.
        _validated: Skip ``validate_preprocessing_steps`` when ``True``; set by
            callers such as ``SkyulfPipeline`` that already validated the same
            structural rules via ``validate_pipeline_config``.
        frame_spec: Explicit keys and optional target for a single Spark frame.
        execution_options: Opt-in engine selection; must match the input engine.
    """
    # `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]] = []
    if frame_spec is not None and not isinstance(frame_spec, FrameSpec):
        raise TypeError("frame_spec must be FrameSpec.")
    if execution_options is not None and not isinstance(execution_options, ExecutionOptions):
        raise TypeError("execution_options must be ExecutionOptions.")
    self.frame_spec = frame_spec
    self.execution_options = execution_options
    self._spark_fitted = False
    self._portable_fitted = False

export_state()

Export supported fitted steps as bounded, versioned UTF-8 JSON bytes.

SimpleImputer mean/constant and StandardScaler are supported. Learned column selection is frozen; row keys, target and runtime objects are supplied again when loading. The complete payload must fit the execution state byte budget (8 MiB by default). Callers own storage and transport.

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

def export_state(self) -> bytes:
    """Export supported fitted steps as bounded, versioned UTF-8 JSON bytes.

    SimpleImputer mean/constant and StandardScaler are supported. Learned
    column selection is frozen; row keys, target and runtime objects are
    supplied again when loading. The complete payload must fit the execution
    state byte budget (8 MiB by default). Callers own storage and transport.
    """
    return export_feature_state(self)

fit_transform(data, node_id_prefix='', *, target_column=None, on_split=None, sample_weight=None)

Runs the pipeline on data.

Parameters:

Name	Type	Description	Default
data	DataFrame | SkyulfDataFrame | Any	Input frame, feature-target pair, or existing split dataset.	required
node_id_prefix	str	Prefix for the per-step transformer identifiers.	''
target_column	str | None	Execution target excluded from automatic feature selection.	None
sample_weight	Any	Optional raw positional weights, sliced by the actual splitter and stored in the resulting training split.	None
on_split	Callable[[SplitDataset], None] | None	Optional synchronous callback receiving the first actual row split before later transformations run. Skipped splitters and feature-target separation do not invoke it. The callback is not retained; consumers must copy or persist the payload immediately if they need a snapshot. Callback errors propagate to the caller.	None

Returns:

Type	Description
Any	A pair containing the transformed data and per-step metrics.

Spark execution validates all capabilities before any data action. Key uniqueness and per-step identity checks run distributed queries with bounded results; local row counts and memory metrics remain unknown.

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

def fit_transform(
    self,
    data: pd.DataFrame | SkyulfDataFrame | Any,
    node_id_prefix="",
    *,
    target_column: str | None = None,
    on_split: Callable[[SplitDataset], None] | None = None,
    sample_weight: Any = None,
) -> Any:
    """Runs the pipeline on data.

    Args:
        data: Input frame, feature-target pair, or existing split dataset.
        node_id_prefix (str): Prefix for the per-step transformer identifiers.
        target_column: Execution target excluded from automatic feature selection.
        sample_weight: Optional raw positional weights, sliced by the actual
            splitter and stored in the resulting training split.
        on_split: Optional synchronous callback receiving the first actual row
            split before later transformations run. Skipped splitters and
            feature-target separation do not invoke it. The callback is not
            retained; consumers must copy or persist the payload immediately
            if they need a snapshot. Callback errors propagate to the caller.

    Returns:
        A pair containing the transformed data and per-step metrics.

    Spark execution validates all capabilities before any data action. Key
    uniqueness and per-step identity checks run distributed queries with
    bounded results; local row counts and memory metrics remain unknown.
    """
    raw_weights = prepare_pipeline_weights(data, sample_weight, self.steps_config)
    spec = getattr(self, "frame_spec", None)
    if use_spark(data, getattr(self, "execution_options", None), spec):
        return self._fit_transform_spark(data, spec, target_column, on_split)
    self._portable_fitted = False
    self.fitted_steps = []  # Reset fitted steps
    current_data = data
    split_captured = False
    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 = dict(step.get("params", {}))
        if target_column is not None:
            # A raw-frame target must not become an auto-selected feature.
            # Keep the caller's config untouched and use pipeline context.
            params["target_column"] = target_column
        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,
            sample_weight=raw_weights,
        )

        raw_weights = _step_weights(current_data, transformer_inst, raw_weights)
        record_target_labels(
            data_before, current_data, target_column, transformer_type, fitted_params
        )

        if (
            on_split is not None
            and not split_captured
            and transformer_type in {"TrainTestSplitter", "Split"}
            and not isinstance(data_before, SplitDataset)
            and isinstance(current_data, SplitDataset)
        ):
            split_captured = True
            on_split(current_data)

        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 = self._step_metric_record(
            name, transformer_type, transformer_inst, rows_before, rows_after, 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"]

    self._portable_fitted = True
    self.train_sample_weight_ = raw_weights
    self.evaluation_coverage_ = deepcopy(getattr(current_data, "evaluation_coverage", {}))
    return current_data, metrics

from_state(payload, *, frame_spec=None, execution_options=None) classmethod

Restore supported fitted steps without fitting, I/O or creating a Spark session.

Spark requires explicit frame_spec and execution_options. Local callers may omit both and let input frames select pandas or Polars. Malformed, oversized, unsupported or corrupt envelopes are rejected before execution.

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

@classmethod
def from_state(
    cls,
    payload: bytes,
    *,
    frame_spec: FrameSpec | None = None,
    execution_options: ExecutionOptions | None = None,
) -> FeatureEngineer:
    """Restore supported fitted steps without fitting, I/O or creating a Spark session.

    Spark requires explicit frame_spec and execution_options. Local callers
    may omit both and let input frames select pandas or Polars. Malformed,
    oversized, unsupported or corrupt envelopes are rejected before execution.
    """
    configs, records = restore_feature_state(payload, execution_options, frame_spec)
    engineer = cls(configs, frame_spec=frame_spec, execution_options=execution_options)
    engineer.fitted_steps = records
    engineer._portable_fitted = True
    engineer._spark_fitted = (
        execution_options is not None and execution_options.engine == "spark"
    )
    return engineer

transform(data, *, preserve_rows=False)

Apply transformations, optionally rejecting row-count changes and temporal sorting.

Local prediction callers set preserve_rows=True because their responses have no input-row provenance. Each applied step is checked immediately; a later step cannot conceal filtering, expansion, or a built-in temporal permutation. Ordinary transform and fold scoring retain their configured filtering and sorting behavior. Custom appliers receive count checks only. Spark execution always preserves key identity and optional target values; physical row order is not guaranteed. Consumers must align by row keys.

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

def transform(
    self, data: pd.DataFrame | SkyulfDataFrame | Any, *, preserve_rows: bool = False
) -> Any:
    """Apply transformations, optionally rejecting row-count changes and temporal sorting.

    Local prediction callers set ``preserve_rows=True`` because their responses
    have no input-row provenance. Each applied step is checked immediately;
    a later step cannot conceal filtering, expansion, or a built-in temporal
    permutation. Ordinary transform and fold scoring retain their configured
    filtering and sorting behavior. Custom appliers receive count checks only.
    Spark execution always preserves key identity and optional target values;
    physical row order is not guaranteed. Consumers must align by row keys.
    """
    spec = getattr(self, "frame_spec", None)
    if use_spark(data, getattr(self, "execution_options", None), spec):
        if not getattr(self, "_spark_fitted", False):
            raise ValueError("Spark FeatureEngineer must be fitted before transform.")
        assert spec is not None
        assert self.execution_options is not None
        return transform_spark(
            data,
            self.fitted_steps,
            spec,
            state_max_bytes=self.execution_options.state_max_bytes,
        )
    current_data = data
    input_rows = prediction_row_count(unpack_pipeline_input(data)[0])
    coverage_steps: list[dict[str, Any]] = []

    for step in self._transform_steps(preserve_rows=preserve_rows):
        name = step["name"]
        transformer_type = step["type"]
        applier = step["applier"]
        artifact = step["artifact"]

        logger.debug(f"Applying step: {name} ({transformer_type})")
        rows_before = prediction_row_count(unpack_pipeline_input(current_data)[0])
        current_data = (
            _apply_prediction_step(current_data, step)
            if preserve_rows
            else applier.apply(current_data, artifact)
        )
        if not preserve_rows:
            coverage_steps.append(
                {
                    "name": name,
                    "transformer": transformer_type,
                    **record_coverage(
                        rows_before,
                        prediction_row_count(unpack_pipeline_input(current_data)[0]),
                        step_name=f"{name} ({transformer_type})",
                    ),
                }
            )

    if not preserve_rows:
        self.last_transform_coverage_ = record_coverage(
            input_rows,
            prediction_row_count(unpack_pipeline_input(current_data)[0]),
            coverage_steps,
        )
    return current_data

transform_tracking_order(data)

Transform local features and report each output row's original input position.

Only built-in temporal permutations are tracked. Row-count changes fail; ordinary serving continues to reject permutations via preserve_rows=True.

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

def transform_tracking_order(self, data: Any) -> tuple[Any, np.ndarray]:
    """Transform local features and report each output row's original input position.

    Only built-in temporal permutations are tracked. Row-count changes fail;
    ordinary serving continues to reject permutations via preserve_rows=True.
    """
    if use_spark(
        data, getattr(self, "execution_options", None), getattr(self, "frame_spec", None)
    ):
        raise ValueError("Positional prediction tracking requires a local frame.")
    positions = np.arange(prediction_row_count(data))
    for step in self._transform_steps():
        data, local_positions = _apply_tracked_step(data, step)
        if local_positions is not None:
            positions = positions[local_positions]
    return data, positions