API: preprocessing.outliers

skyulf.preprocessing.outliers

Outlier-handling nodes package.

Split from a single 539-LOC module into per-node files

_common.py — shared mask helpers (y filter, pandas mask apply) iqr.py — IQR zscore.py — ZScore winsorize.py — Winsorize manual_bounds.py — ManualBounds elliptic.py — EllipticEnvelope clip_values.py — ClipValues (fixed bounds, clips instead of dropping rows)

All public names are re-exported here so existing imports such as from skyulf.preprocessing.outliers import IQRCalculator continue to work.

ClipValuesApplier

Bases: BaseApplier

Set values below lower to lower and above upper to upper.

Nulls stay null, other columns are untouched and every row is kept, so unlike ManualBounds this step also runs on scoring rows.

Source code in skyulf-core/skyulf/preprocessing/outliers/clip_values.py

class ClipValuesApplier(BaseApplier):
    """Set values below ``lower`` to ``lower`` and above ``upper`` to ``upper``.

    Nulls stay null, other columns are untouched and every row is kept, so
    unlike ManualBounds this step also runs on scoring rows.
    """

    @staticmethod
    def validate_fitted_state(raw: dict) -> dict:
        """Inspect this node's supported saved state without fitting or applying data."""
        return _clip_state(raw)

    @staticmethod
    def resolve_fitted_config(raw: dict, state: dict) -> dict:
        """Bind inference configuration to this node's inspected fitted artifact."""
        params = _normalize(raw)
        params.pop("target_column", None)
        params.pop("_auto_columns", None)
        _fields(params, {"bounds"})
        bounds = _bounds(params["bounds"])
        if bounds != state["bounds"]:
            raise ValueError("Configured bounds disagree with fitted clipping bounds.")
        return {"bounds": bounds}

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Clip the configured columns on the active engine; ``y`` passes through."""
        input_data = (X, y) if y is not None else X
        return apply_dual_engine(
            input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

    @staticmethod
    def _apply_polars(X: Any, y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        exprs = [
            _clip_polars_column(X, column, bound)
            for column, bound in params.get("bounds", {}).items()
            if column in X.columns
        ]
        return X.with_columns(exprs), y

    @staticmethod
    def _apply_pandas(X: Any, y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        out = X.copy()
        for column, bound in params.get("bounds", {}).items():
            if column not in out.columns:
                continue
            series = out[column]
            if is_decimal_series(series) or _fractional_bound(bound):
                series = pd.to_numeric(series).astype("float64")
            out[column] = series.clip(lower=bound.get("lower"), upper=bound.get("upper"))
        return out, y

apply(X, y, params)

Clip the configured columns on the active engine; y passes through.

Source code in skyulf-core/skyulf/preprocessing/outliers/clip_values.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Clip the configured columns on the active engine; ``y`` passes through."""
    input_data = (X, y) if y is not None else X
    return apply_dual_engine(
        input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

resolve_fitted_config(raw, state) staticmethod

Bind inference configuration to this node's inspected fitted artifact.

Source code in skyulf-core/skyulf/preprocessing/outliers/clip_values.py

@staticmethod
def resolve_fitted_config(raw: dict, state: dict) -> dict:
    """Bind inference configuration to this node's inspected fitted artifact."""
    params = _normalize(raw)
    params.pop("target_column", None)
    params.pop("_auto_columns", None)
    _fields(params, {"bounds"})
    bounds = _bounds(params["bounds"])
    if bounds != state["bounds"]:
        raise ValueError("Configured bounds disagree with fitted clipping bounds.")
    return {"bounds": bounds}

validate_fitted_state(raw) staticmethod

Inspect this node's supported saved state without fitting or applying data.

Source code in skyulf-core/skyulf/preprocessing/outliers/clip_values.py

@staticmethod
def validate_fitted_state(raw: dict) -> dict:
    """Inspect this node's supported saved state without fitting or applying data."""
    return _clip_state(raw)

ClipValuesCalculator

Bases: BaseCalculator

Validate the configured bounds; nothing is learned from the data.

bounds maps a column to {"lower": ..., "upper": ...}; either side may be omitted. Use Winsorize for bounds learned from percentiles.

Source code in skyulf-core/skyulf/preprocessing/outliers/clip_values.py

@NodeRegistry.register(
    "ClipValues",
    ClipValuesApplier,
    execution_capabilities=(
        ExecutionCapability("pandas", "apply", "python_batch", "preserve", "row"),
        ExecutionCapability("polars", "apply", "local", "preserve", "row"),
    ),
)
@node_meta(
    id="ClipValues",
    name="Clip Values",
    category="Preprocessing",
    description=(
        "Limits numeric columns to fixed lower/upper bounds (e.g. employees at most 500) "
        "without removing rows."
    ),
    params={"bounds": {}},
    learns_from_data=False,
)
class ClipValuesCalculator(BaseCalculator):
    """Validate the configured bounds; nothing is learned from the data.

    ``bounds`` maps a column to ``{"lower": ..., "upper": ...}``; either side
    may be omitted. Use Winsorize for bounds learned from percentiles.
    """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Promote fractionally bounded columns to match both runtime engines."""
        schema = input_schema
        for column, bound in (config.get("bounds") or {}).items():
            if column in schema.columns and _fractional_bound(bound):
                schema = schema.with_dtype(column, "float64")
        return schema

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> ClipValuesArtifact:  # pylint: disable=arguments-differ
        """Return the validated bounds as the artifact."""
        raw = config.get("bounds") or {}
        bounds = {column: _clean_bound(column, bound) for column, bound in raw.items()}
        if bounds and X is not None:
            _check_columns(X.to_native() if hasattr(X, "to_native") else X, bounds)
        return {"type": "clip_values", "bounds": bounds}

fit(X, _y, config)

Return the validated bounds as the artifact.

Source code in skyulf-core/skyulf/preprocessing/outliers/clip_values.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> ClipValuesArtifact:  # pylint: disable=arguments-differ
    """Return the validated bounds as the artifact."""
    raw = config.get("bounds") or {}
    bounds = {column: _clean_bound(column, bound) for column, bound in raw.items()}
    if bounds and X is not None:
        _check_columns(X.to_native() if hasattr(X, "to_native") else X, bounds)
    return {"type": "clip_values", "bounds": bounds}

infer_output_schema(input_schema, config)

Promote fractionally bounded columns to match both runtime engines.

Source code in skyulf-core/skyulf/preprocessing/outliers/clip_values.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Promote fractionally bounded columns to match both runtime engines."""
    schema = input_schema
    for column, bound in (config.get("bounds") or {}).items():
        if column in schema.columns and _fractional_bound(bound):
            schema = schema.with_dtype(column, "float64")
    return schema

EllipticEnvelopeApplier

Bases: BaseApplier

Drop rows that any fitted per-column EllipticEnvelope model flags as an outlier.

Source code in skyulf-core/skyulf/preprocessing/outliers/elliptic.py

class EllipticEnvelopeApplier(BaseApplier):
    """Drop rows that any fitted per-column EllipticEnvelope model flags as an outlier."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved envelopes and skipped-column metadata without estimating covariance."""
        if not local_state_fields(
            raw,
            "elliptic_envelope",
            {"type", "models", "contamination", "warnings"},
            allow_empty=True,
        ):
            return raw
        models = raw["models"]
        if type(models) is not dict:
            raise ValueError("Fitted EllipticEnvelope models must be a dictionary.")
        for model in models.values():
            _validate_elliptic_model(model)
        validate_detector_warnings(raw["warnings"])
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe fitted row filtering with nonfinite rows preserved by the native applier."""
        if engine not in ("pandas", "polars"):
            return None
        EllipticEnvelopeApplier.validate_inference_state(state)
        effect = "filter" if state.get("models") else "preserve"
        return ExecutionCapability(engine, "apply", "local", effect, "row")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Filter rows through the fitted per-column models on the active engine."""
        # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
        # own unpack_pipeline_input doesn't silently drop y (leaving it
        # unfiltered when X rows are removed). Omit the wrap when y is None
        # to avoid apply_dual_engine's tuple-with-no-y warning log.
        input_data = (X, y) if y is not None else X
        return apply_dual_engine(
            input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

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

        mask = pl.Series(_elliptic_mask_numpy(X, models))
        return X.filter(mask), _filter_y_polars(y, mask)

    @staticmethod
    def _apply_pandas(X: Any, y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        models = params.get("models", {})
        if not models:
            return X, y
        mask = _elliptic_filter_pandas(X, models)
        return _apply_pandas_mask(X, y, mask)

apply(X, y, params)

Filter rows through the fitted per-column models on the active engine.

Source code in skyulf-core/skyulf/preprocessing/outliers/elliptic.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Filter rows through the fitted per-column models on the active engine."""
    # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
    # own unpack_pipeline_input doesn't silently drop y (leaving it
    # unfiltered when X rows are removed). Omit the wrap when y is None
    # to avoid apply_dual_engine's tuple-with-no-y warning log.
    input_data = (X, y) if y is not None else X
    return apply_dual_engine(
        input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe fitted row filtering with nonfinite rows preserved by the native applier.

Source code in skyulf-core/skyulf/preprocessing/outliers/elliptic.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe fitted row filtering with nonfinite rows preserved by the native applier."""
    if engine not in ("pandas", "polars"):
        return None
    EllipticEnvelopeApplier.validate_inference_state(state)
    effect = "filter" if state.get("models") else "preserve"
    return ExecutionCapability(engine, "apply", "local", effect, "row")

validate_inference_state(raw) staticmethod

Inspect saved envelopes and skipped-column metadata without estimating covariance.

Source code in skyulf-core/skyulf/preprocessing/outliers/elliptic.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved envelopes and skipped-column metadata without estimating covariance."""
    if not local_state_fields(
        raw,
        "elliptic_envelope",
        {"type", "models", "contamination", "warnings"},
        allow_empty=True,
    ):
        return raw
    models = raw["models"]
    if type(models) is not dict:
        raise ValueError("Fitted EllipticEnvelope models must be a dictionary.")
    for model in models.values():
        _validate_elliptic_model(model)
    validate_detector_warnings(raw["warnings"])
    return raw

EllipticEnvelopeCalculator

Bases: BaseCalculator

Fit one univariate EllipticEnvelope per numeric column at the configured contamination.

Source code in skyulf-core/skyulf/preprocessing/outliers/elliptic.py

@NodeRegistry.register("EllipticEnvelope", EllipticEnvelopeApplier)
@node_meta(
    id="EllipticEnvelope",
    name="Elliptic Envelope",
    category="Preprocessing",
    description="Detect outliers in a Gaussian distributed dataset.",
    params={"contamination": 0.01, "columns": [], "random_state": DEFAULT_RANDOM_STATE},
    learns_from_data=True,
)
class EllipticEnvelopeCalculator(BaseCalculator):
    """Fit one univariate EllipticEnvelope per numeric column at the configured contamination."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return the input schema unchanged: row filtering preserves the column set."""
        # Elliptic envelope filters outlier *rows*; column set is preserved.
        return input_schema

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> EllipticEnvelopeArtifact:  # pylint: disable=arguments-differ
        """Fit the per-column envelopes, skipping columns with fewer than five usable samples."""
        if user_picked_no_columns(config):
            return {}

        contamination = config.get("contamination", 0.01)
        random_state = config.get("random_state", DEFAULT_RANDOM_STATE)
        # TODO(pandas-removal): same per-column coercion/dropna caveat as
        # zscore.py — EllipticEnvelope.fit itself takes numpy fine, but the
        # per-column pd.to_numeric(errors="coerce").dropna() skip-logic for
        # non-numeric/short columns needs a native-Polars equivalent first.
        X_pd, cols = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
        if not cols:
            return {}

        models: dict[str, Any] = {}
        warnings = []
        for col in cols:
            series = pd.to_numeric(X_pd[col], errors="coerce").dropna()
            if series.shape[0] < 5:
                warnings.append(f"Column '{col}': Too few samples ({series.shape[0]})")
                continue
            try:
                model = EllipticEnvelope(contamination=contamination, random_state=random_state)
                model.fit(series.to_numpy().reshape(-1, 1))
                models[col] = model
            except Exception as e:  # noqa: BLE001 - per-column fit failure is logged and recorded in warnings
                logger.warning(f"EllipticEnvelope fit failed for column {col}: {e}")
                warnings.append(f"Column '{col}': {str(e)}")

        return {
            "type": "elliptic_envelope",
            "models": models,
            "contamination": contamination,
            "warnings": warnings,
        }

fit(X, _y, config)

Fit the per-column envelopes, skipping columns with fewer than five usable samples.

Source code in skyulf-core/skyulf/preprocessing/outliers/elliptic.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> EllipticEnvelopeArtifact:  # pylint: disable=arguments-differ
    """Fit the per-column envelopes, skipping columns with fewer than five usable samples."""
    if user_picked_no_columns(config):
        return {}

    contamination = config.get("contamination", 0.01)
    random_state = config.get("random_state", DEFAULT_RANDOM_STATE)
    # TODO(pandas-removal): same per-column coercion/dropna caveat as
    # zscore.py — EllipticEnvelope.fit itself takes numpy fine, but the
    # per-column pd.to_numeric(errors="coerce").dropna() skip-logic for
    # non-numeric/short columns needs a native-Polars equivalent first.
    X_pd, cols = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
    if not cols:
        return {}

    models: dict[str, Any] = {}
    warnings = []
    for col in cols:
        series = pd.to_numeric(X_pd[col], errors="coerce").dropna()
        if series.shape[0] < 5:
            warnings.append(f"Column '{col}': Too few samples ({series.shape[0]})")
            continue
        try:
            model = EllipticEnvelope(contamination=contamination, random_state=random_state)
            model.fit(series.to_numpy().reshape(-1, 1))
            models[col] = model
        except Exception as e:  # noqa: BLE001 - per-column fit failure is logged and recorded in warnings
            logger.warning(f"EllipticEnvelope fit failed for column {col}: {e}")
            warnings.append(f"Column '{col}': {str(e)}")

    return {
        "type": "elliptic_envelope",
        "models": models,
        "contamination": contamination,
        "warnings": warnings,
    }

infer_output_schema(input_schema, config)

Return the input schema unchanged: row filtering preserves the column set.

Source code in skyulf-core/skyulf/preprocessing/outliers/elliptic.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Return the input schema unchanged: row filtering preserves the column set."""
    # Elliptic envelope filters outlier *rows*; column set is preserved.
    return input_schema

IQRApplier

Bases: BaseApplier

Drop rows outside the fitted per-column IQR bounds; null/NaN values are kept.

Source code in skyulf-core/skyulf/preprocessing/outliers/iqr.py

class IQRApplier(BaseApplier):
    """Drop rows outside the fitted per-column IQR bounds; null/NaN values are kept."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect learned quartile limits without recomputing request statistics."""
        if local_state_fields(
            raw, "iqr", {"type", "bounds", "multiplier", "warnings"}, allow_empty=True
        ):
            validate_fitted_bounds(raw["bounds"], partial=False)
            validate_detector_warnings(raw["warnings"])
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe saved row filtering; prediction still rejects any removed rows."""
        if engine not in ("pandas", "polars"):
            return None
        IQRApplier.validate_inference_state(state)
        effect = "filter" if state.get("bounds") else "preserve"
        return ExecutionCapability(engine, "apply", "local", effect, "row")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Filter rows through the fitted per-column bounds on the active engine."""
        # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
        # own unpack_pipeline_input doesn't silently drop y (leaving it
        # unfiltered when X rows are removed). Omit the wrap when y is None
        # to avoid apply_dual_engine's tuple-with-no-y warning log.
        input_data = (X, y) if y is not None else X
        return apply_dual_engine(
            input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

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

        mask = pl.repeat(True, pl.len())
        for col, bound in bounds.items():
            if col not in X.columns:
                continue
            values = pl.col(col).cast(pl.Float64, strict=False)
            col_mask = (values >= bound["lower"]) & (values <= bound["upper"])
            missing = values.is_null() | values.is_nan()
            mask = mask & (col_mask | missing)

        mask_series = X.select(mask.alias("mask")).get_column("mask")
        return X.filter(mask_series), _filter_y_polars(y, mask_series)

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

        mask = pd.Series(True, index=X.index)
        for col, bound in bounds.items():
            if col not in X.columns:
                continue
            series = pd.to_numeric(X[col], errors="coerce")
            col_mask = (series >= bound["lower"]) & (series <= bound["upper"])
            mask = mask & (col_mask | series.isna())

        return _apply_pandas_mask(X, y, mask)

apply(X, y, params)

Filter rows through the fitted per-column bounds on the active engine.

Source code in skyulf-core/skyulf/preprocessing/outliers/iqr.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Filter rows through the fitted per-column bounds on the active engine."""
    # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
    # own unpack_pipeline_input doesn't silently drop y (leaving it
    # unfiltered when X rows are removed). Omit the wrap when y is None
    # to avoid apply_dual_engine's tuple-with-no-y warning log.
    input_data = (X, y) if y is not None else X
    return apply_dual_engine(
        input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe saved row filtering; prediction still rejects any removed rows.

Source code in skyulf-core/skyulf/preprocessing/outliers/iqr.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe saved row filtering; prediction still rejects any removed rows."""
    if engine not in ("pandas", "polars"):
        return None
    IQRApplier.validate_inference_state(state)
    effect = "filter" if state.get("bounds") else "preserve"
    return ExecutionCapability(engine, "apply", "local", effect, "row")

validate_inference_state(raw) staticmethod

Inspect learned quartile limits without recomputing request statistics.

Source code in skyulf-core/skyulf/preprocessing/outliers/iqr.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect learned quartile limits without recomputing request statistics."""
    if local_state_fields(
        raw, "iqr", {"type", "bounds", "multiplier", "warnings"}, allow_empty=True
    ):
        validate_fitted_bounds(raw["bounds"], partial=False)
        validate_detector_warnings(raw["warnings"])
    return raw

IQRCalculator

Bases: BaseCalculator

Fit per-column bounds at the quartiles widened by multiplier times the IQR.

Source code in skyulf-core/skyulf/preprocessing/outliers/iqr.py

@NodeRegistry.register("IQR", IQRApplier)
@node_meta(
    id="IQR",
    name="IQR Outlier Removal",
    category="Preprocessing",
    description="Remove outliers using Interquartile Range.",
    params={"multiplier": 1.5, "columns": []},
    learns_from_data=True,
)
class IQRCalculator(BaseCalculator):
    """Fit per-column bounds at the quartiles widened by ``multiplier`` times the IQR."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return the input schema unchanged: row filtering preserves the column set."""
        # IQR removes outlier *rows*; column set is preserved.
        return input_schema

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> IQRArtifact:  # pylint: disable=arguments-differ
        """Compute quartile bounds per numeric column; warn on empty or non-numeric ones."""
        if user_picked_no_columns(config):
            return {}

        multiplier = config.get("multiplier", 1.5)
        # TODO(pandas-removal): stays on resolve_columns_then_to_pandas (not
        # ..._to_numpy) because `.quantile()` needs Pandas' linear
        # interpolation semantics, which differ from Polars' default
        # "nearest" interpolation (see _helpers.py note). Switching to numpy
        # would require doing the quantile math with
        # np.percentile(..., method="linear") plus matching Pandas'
        # errors="coerce" NaN-drop behavior before benchmarking equivalence.
        X_pd, cols = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
        if not cols:
            return {}

        bounds: dict[str, dict[str, float]] = {}
        warnings = []
        for col in cols:
            series = pd.to_numeric(X_pd[col], errors="coerce").dropna()
            if series.empty:
                warnings.append(f"Column '{col}': Empty or non-numeric")
                continue
            q1 = series.quantile(0.25)
            q3 = series.quantile(0.75)
            iqr = q3 - q1
            bounds[col] = {"lower": q1 - (multiplier * iqr), "upper": q3 + (multiplier * iqr)}

        return {
            "type": "iqr",
            "bounds": bounds,
            "multiplier": multiplier,
            "warnings": warnings,
        }

fit(X, _y, config)

Compute quartile bounds per numeric column; warn on empty or non-numeric ones.

Source code in skyulf-core/skyulf/preprocessing/outliers/iqr.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> IQRArtifact:  # pylint: disable=arguments-differ
    """Compute quartile bounds per numeric column; warn on empty or non-numeric ones."""
    if user_picked_no_columns(config):
        return {}

    multiplier = config.get("multiplier", 1.5)
    # TODO(pandas-removal): stays on resolve_columns_then_to_pandas (not
    # ..._to_numpy) because `.quantile()` needs Pandas' linear
    # interpolation semantics, which differ from Polars' default
    # "nearest" interpolation (see _helpers.py note). Switching to numpy
    # would require doing the quantile math with
    # np.percentile(..., method="linear") plus matching Pandas'
    # errors="coerce" NaN-drop behavior before benchmarking equivalence.
    X_pd, cols = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
    if not cols:
        return {}

    bounds: dict[str, dict[str, float]] = {}
    warnings = []
    for col in cols:
        series = pd.to_numeric(X_pd[col], errors="coerce").dropna()
        if series.empty:
            warnings.append(f"Column '{col}': Empty or non-numeric")
            continue
        q1 = series.quantile(0.25)
        q3 = series.quantile(0.75)
        iqr = q3 - q1
        bounds[col] = {"lower": q1 - (multiplier * iqr), "upper": q3 + (multiplier * iqr)}

    return {
        "type": "iqr",
        "bounds": bounds,
        "multiplier": multiplier,
        "warnings": warnings,
    }

infer_output_schema(input_schema, config)

Return the input schema unchanged: row filtering preserves the column set.

Source code in skyulf-core/skyulf/preprocessing/outliers/iqr.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Return the input schema unchanged: row filtering preserves the column set."""
    # IQR removes outlier *rows*; column set is preserved.
    return input_schema

ManualBoundsApplier

Bases: BaseApplier

Drop rows outside the user-specified per-column lower/upper bounds.

Source code in skyulf-core/skyulf/preprocessing/outliers/manual_bounds.py

class ManualBoundsApplier(BaseApplier):
    """Drop rows outside the user-specified per-column lower/upper bounds."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved optional bounds without applying the row filter."""
        local_state_fields(raw, "manual_bounds", {"type", "bounds"})
        validate_fitted_bounds(raw["bounds"], partial=True)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Declare row-local filtering; prediction rejects requests that lose rows."""
        if engine not in ("pandas", "polars"):
            return None
        ManualBoundsApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "filter", "row")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Filter rows through the configured bounds on the active engine."""
        # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
        # own unpack_pipeline_input doesn't silently drop y (leaving it
        # unfiltered when X rows are removed). Omit the wrap when y is None
        # to avoid apply_dual_engine's tuple-with-no-y warning log.
        input_data = (X, y) if y is not None else X
        return apply_dual_engine(
            input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

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

        # A scalar mask broadcasts for X but yields only index zero when filtering y.
        mask = pl.repeat(True, pl.len())
        for col, bound in bounds.items():
            if col not in X.columns:
                continue
            mask = mask & _manual_bounds_col_mask_polars(col, bound)

        mask_series = X.select(mask.alias("mask")).get_column("mask")
        return X.filter(mask_series), _filter_y_polars(y, mask_series)

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

        mask = pd.Series(True, index=X.index)
        for col, bound in bounds.items():
            if col not in X.columns:
                continue
            series = pd.to_numeric(X[col], errors="coerce")
            mask = mask & _manual_bounds_col_mask_pandas(series, bound)

        return _apply_pandas_mask(X, y, mask)

apply(X, y, params)

Filter rows through the configured bounds on the active engine.

Source code in skyulf-core/skyulf/preprocessing/outliers/manual_bounds.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Filter rows through the configured bounds on the active engine."""
    # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
    # own unpack_pipeline_input doesn't silently drop y (leaving it
    # unfiltered when X rows are removed). Omit the wrap when y is None
    # to avoid apply_dual_engine's tuple-with-no-y warning log.
    input_data = (X, y) if y is not None else X
    return apply_dual_engine(
        input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Declare row-local filtering; prediction rejects requests that lose rows.

Source code in skyulf-core/skyulf/preprocessing/outliers/manual_bounds.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Declare row-local filtering; prediction rejects requests that lose rows."""
    if engine not in ("pandas", "polars"):
        return None
    ManualBoundsApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "filter", "row")

validate_inference_state(raw) staticmethod

Inspect saved optional bounds without applying the row filter.

Source code in skyulf-core/skyulf/preprocessing/outliers/manual_bounds.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved optional bounds without applying the row filter."""
    local_state_fields(raw, "manual_bounds", {"type", "bounds"})
    validate_fitted_bounds(raw["bounds"], partial=True)
    return raw

ManualBoundsCalculator

Bases: BaseCalculator

Package the user-configured bounds into an artifact without reading the data.

Source code in skyulf-core/skyulf/preprocessing/outliers/manual_bounds.py

@NodeRegistry.register("ManualBounds", ManualBoundsApplier)
@node_meta(
    id="ManualBounds",
    name="Manual Bounds",
    category="Preprocessing",
    description="Filter outliers by manually specifying lower and upper bounds for columns.",
    params={"bounds": {}},
    learns_from_data=False,
)
class ManualBoundsCalculator(BaseCalculator):
    """Package the user-configured bounds into an artifact without reading the data."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return the input schema unchanged: row filtering preserves the column set."""
        # Manual bounds filter rows; column set is preserved.
        return input_schema

    @fit_method
    def fit(self, _X: Any, _y: Any, config: dict[str, Any]) -> ManualBoundsArtifact:  # pylint: disable=arguments-differ
        """Return the ``bounds`` config as-is; no statistics are computed."""
        # Config: {'bounds': {'col1': {'lower': 0, 'upper': 100}, ...}}
        return {"type": "manual_bounds", "bounds": config.get("bounds", {})}

fit(_X, _y, config)

Return the bounds config as-is; no statistics are computed.

Source code in skyulf-core/skyulf/preprocessing/outliers/manual_bounds.py

@fit_method
def fit(self, _X: Any, _y: Any, config: dict[str, Any]) -> ManualBoundsArtifact:  # pylint: disable=arguments-differ
    """Return the ``bounds`` config as-is; no statistics are computed."""
    # Config: {'bounds': {'col1': {'lower': 0, 'upper': 100}, ...}}
    return {"type": "manual_bounds", "bounds": config.get("bounds", {})}

infer_output_schema(input_schema, config)

Return the input schema unchanged: row filtering preserves the column set.

Source code in skyulf-core/skyulf/preprocessing/outliers/manual_bounds.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Return the input schema unchanged: row filtering preserves the column set."""
    # Manual bounds filter rows; column set is preserved.
    return input_schema

WinsorizeApplier

Bases: BaseApplier

Clip values to the fitted percentile bounds; rows are never removed.

Source code in skyulf-core/skyulf/preprocessing/outliers/winsorize.py

class WinsorizeApplier(BaseApplier):
    """Clip values to the fitted percentile bounds; rows are never removed."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect learned bounds and metadata without recomputing request quantiles."""
        if not local_state_fields(
            raw,
            "winsorize",
            {"type", "bounds", "lower_percentile", "upper_percentile", "warnings"},
            allow_empty=True,
        ):
            return raw
        validate_fitted_bounds(raw["bounds"], partial=False)
        for field in ("lower_percentile", "upper_percentile"):
            value = raw[field]
            if isinstance(value, bool) or not isinstance(value, Real) or not 0 <= value <= 100:
                raise ValueError("Fitted percentiles must be numbers between zero and 100.")
        if type(raw["warnings"]) is not list or any(
            type(item) is not str for item in raw["warnings"]
        ):
            raise ValueError("Fitted warnings must be a list of strings.")
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe local row clipping with saved limits without admitting Spark workers."""
        if engine not in ("pandas", "polars"):
            return None
        WinsorizeApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Clip ``X`` to the fitted per-column bounds; ``y`` passes through untouched."""
        # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
        # own unpack_pipeline_input doesn't silently drop y. Winsorize never
        # filters rows, but the wrap keeps behavior consistent with the other
        # outlier nodes and avoids losing y from the returned tuple.
        input_data = (X, y) if y is not None else X
        return apply_dual_engine(
            input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

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

        numeric_cols = set(auto_detect_numeric_columns(X))
        exprs = []
        for col, bound in bounds.items():
            if col not in X.columns or col not in numeric_cols:
                continue
            exprs.append(_clip_polars_column(X, col, bound))
        return X.with_columns(exprs), y

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

        df_out = X.copy()
        for col, bound in bounds.items():
            if col not in df_out.columns:
                continue
            if pd.api.types.is_numeric_dtype(df_out[col]) or is_decimal_series(df_out[col]):
                df_out[col] = _clip_pandas_column(df_out[col], bound, col)
        return df_out, y

apply(X, y, params)

Clip X to the fitted per-column bounds; y passes through untouched.

Source code in skyulf-core/skyulf/preprocessing/outliers/winsorize.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Clip ``X`` to the fitted per-column bounds; ``y`` passes through untouched."""
    # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
    # own unpack_pipeline_input doesn't silently drop y. Winsorize never
    # filters rows, but the wrap keeps behavior consistent with the other
    # outlier nodes and avoids losing y from the returned tuple.
    input_data = (X, y) if y is not None else X
    return apply_dual_engine(
        input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe local row clipping with saved limits without admitting Spark workers.

Source code in skyulf-core/skyulf/preprocessing/outliers/winsorize.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe local row clipping with saved limits without admitting Spark workers."""
    if engine not in ("pandas", "polars"):
        return None
    WinsorizeApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect learned bounds and metadata without recomputing request quantiles.

Source code in skyulf-core/skyulf/preprocessing/outliers/winsorize.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect learned bounds and metadata without recomputing request quantiles."""
    if not local_state_fields(
        raw,
        "winsorize",
        {"type", "bounds", "lower_percentile", "upper_percentile", "warnings"},
        allow_empty=True,
    ):
        return raw
    validate_fitted_bounds(raw["bounds"], partial=False)
    for field in ("lower_percentile", "upper_percentile"):
        value = raw[field]
        if isinstance(value, bool) or not isinstance(value, Real) or not 0 <= value <= 100:
            raise ValueError("Fitted percentiles must be numbers between zero and 100.")
    if type(raw["warnings"]) is not list or any(
        type(item) is not str for item in raw["warnings"]
    ):
        raise ValueError("Fitted warnings must be a list of strings.")
    return raw

WinsorizeCalculator

Bases: BaseCalculator

Fit per-column clip bounds at the configured lower/upper percentiles.

Source code in skyulf-core/skyulf/preprocessing/outliers/winsorize.py

@NodeRegistry.register("Winsorize", WinsorizeApplier)
@node_meta(
    id="Winsorize",
    name="Winsorization",
    category="Preprocessing",
    description="Limit extreme values in the data.",
    params={"lower_percentile": 5.0, "upper_percentile": 95.0, "columns": []},
    learns_from_data=True,
)
class WinsorizeCalculator(BaseCalculator):
    """Fit per-column clip bounds at the configured lower/upper percentiles."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Preserve integer columns and retain existing promotion for other selected types."""
        if not isinstance(input_schema, SkyulfSchema):
            return input_schema
        output = promote_configured_columns_to_float64(input_schema, config)
        for column, dtype in input_schema.dtypes.items():
            if dtype.lower().startswith(("int", "uint")):
                output = output.with_dtype(column, dtype)
        return output

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> WinsorizeArtifact:  # pylint: disable=arguments-differ
        """Compute each column's lower/upper quantiles; warn on empty or non-numeric ones."""
        if user_picked_no_columns(config):
            return {}

        lower_p = config.get("lower_percentile", 5.0)
        upper_p = config.get("upper_percentile", 95.0)
        X = X.to_native() if hasattr(X, "to_native") else X
        cols = resolve_columns(X, config, detect_numeric_columns)
        if not cols:
            return {}

        bounds: dict[str, dict[str, float]] = {}
        warnings = []
        for col in cols:
            series, integer = _fit_series(X, col)
            if series.empty:
                warnings.append(f"Column '{col}': Empty or non-numeric")
                continue
            bounds[col] = {
                "lower": _fit_bound(series, lower_p, col, integer=integer),
                "upper": _fit_bound(series, upper_p, col, integer=integer),
            }
            if integer:
                bounds[col] = _pandas_integer_bounds(series, bounds[col], col)

        return {
            "type": "winsorize",
            "bounds": bounds,
            "lower_percentile": lower_p,
            "upper_percentile": upper_p,
            "warnings": warnings,
        }

fit(X, _y, config)

Compute each column's lower/upper quantiles; warn on empty or non-numeric ones.

Source code in skyulf-core/skyulf/preprocessing/outliers/winsorize.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> WinsorizeArtifact:  # pylint: disable=arguments-differ
    """Compute each column's lower/upper quantiles; warn on empty or non-numeric ones."""
    if user_picked_no_columns(config):
        return {}

    lower_p = config.get("lower_percentile", 5.0)
    upper_p = config.get("upper_percentile", 95.0)
    X = X.to_native() if hasattr(X, "to_native") else X
    cols = resolve_columns(X, config, detect_numeric_columns)
    if not cols:
        return {}

    bounds: dict[str, dict[str, float]] = {}
    warnings = []
    for col in cols:
        series, integer = _fit_series(X, col)
        if series.empty:
            warnings.append(f"Column '{col}': Empty or non-numeric")
            continue
        bounds[col] = {
            "lower": _fit_bound(series, lower_p, col, integer=integer),
            "upper": _fit_bound(series, upper_p, col, integer=integer),
        }
        if integer:
            bounds[col] = _pandas_integer_bounds(series, bounds[col], col)

    return {
        "type": "winsorize",
        "bounds": bounds,
        "lower_percentile": lower_p,
        "upper_percentile": upper_p,
        "warnings": warnings,
    }

infer_output_schema(input_schema, config)

Preserve integer columns and retain existing promotion for other selected types.

Source code in skyulf-core/skyulf/preprocessing/outliers/winsorize.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Preserve integer columns and retain existing promotion for other selected types."""
    if not isinstance(input_schema, SkyulfSchema):
        return input_schema
    output = promote_configured_columns_to_float64(input_schema, config)
    for column, dtype in input_schema.dtypes.items():
        if dtype.lower().startswith(("int", "uint")):
            output = output.with_dtype(column, dtype)
    return output

ZScoreApplier

Bases: BaseApplier

Drop rows whose |z| against the fitted mean/std exceeds threshold in any column.

Source code in skyulf-core/skyulf/preprocessing/outliers/zscore.py

class ZScoreApplier(BaseApplier):
    """Drop rows whose |z| against the fitted mean/std exceeds ``threshold`` in any column."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved numeric statistics and native scalar threshold semantics."""
        if local_state_fields(
            raw, "zscore", {"type", "stats", "threshold", "warnings"}, allow_empty=True
        ):
            _validate_zscore_statistics(raw["stats"])
            if not isinstance(raw["threshold"], (Real, Decimal)):
                local_scalar(raw["threshold"], "ZScore threshold")
            validate_detector_warnings(raw["warnings"])
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe saved row filtering without bypassing the prediction row-count guard."""
        if engine not in ("pandas", "polars"):
            return None
        ZScoreApplier.validate_inference_state(state)
        effect = "filter" if state.get("stats") else "preserve"
        return ExecutionCapability(engine, "apply", "local", effect, "row")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Filter rows through the fitted z-score threshold on the active engine."""
        # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
        # own unpack_pipeline_input doesn't silently drop y (leaving it
        # unfiltered when X rows are removed). Omit the wrap when y is None
        # to avoid apply_dual_engine's tuple-with-no-y warning log.
        input_data = (X, y) if y is not None else X
        return apply_dual_engine(
            input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

    @staticmethod
    def _apply_polars(X: Any, y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        stats = params.get("stats", {})
        threshold = params.get("threshold", 3.0)
        if not stats:
            return X, y

        mask = pl.repeat(True, pl.len())
        for col, stat in stats.items():
            if col not in X.columns or stat["std"] == 0:
                continue
            values = pl.col(col).cast(pl.Float64, strict=False)
            z = (values - stat["mean"]) / stat["std"]
            col_mask = z.abs() <= threshold
            missing = values.is_null() | values.is_nan()
            mask = mask & (col_mask | missing)

        mask_series = X.select(mask.alias("mask")).get_column("mask")
        return X.filter(mask_series), _filter_y_polars(y, mask_series)

    @staticmethod
    def _apply_pandas(X: Any, y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        stats = params.get("stats", {})
        threshold = params.get("threshold", 3.0)
        if not stats:
            return X, y

        mask = pd.Series(True, index=X.index)
        for col, stat in stats.items():
            if col not in X.columns or stat["std"] == 0:
                continue
            series = pd.to_numeric(X[col], errors="coerce")
            z = (series - stat["mean"]) / stat["std"]
            col_mask = z.abs() <= threshold
            mask = mask & (col_mask | series.isna())

        return _apply_pandas_mask(X, y, mask)

apply(X, y, params)

Filter rows through the fitted z-score threshold on the active engine.

Source code in skyulf-core/skyulf/preprocessing/outliers/zscore.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Filter rows through the fitted z-score threshold on the active engine."""
    # apply_method already unpacked (X, y); re-wrap so apply_dual_engine's
    # own unpack_pipeline_input doesn't silently drop y (leaving it
    # unfiltered when X rows are removed). Omit the wrap when y is None
    # to avoid apply_dual_engine's tuple-with-no-y warning log.
    input_data = (X, y) if y is not None else X
    return apply_dual_engine(
        input_data, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe saved row filtering without bypassing the prediction row-count guard.

Source code in skyulf-core/skyulf/preprocessing/outliers/zscore.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe saved row filtering without bypassing the prediction row-count guard."""
    if engine not in ("pandas", "polars"):
        return None
    ZScoreApplier.validate_inference_state(state)
    effect = "filter" if state.get("stats") else "preserve"
    return ExecutionCapability(engine, "apply", "local", effect, "row")

validate_inference_state(raw) staticmethod

Inspect saved numeric statistics and native scalar threshold semantics.

Source code in skyulf-core/skyulf/preprocessing/outliers/zscore.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved numeric statistics and native scalar threshold semantics."""
    if local_state_fields(
        raw, "zscore", {"type", "stats", "threshold", "warnings"}, allow_empty=True
    ):
        _validate_zscore_statistics(raw["stats"])
        if not isinstance(raw["threshold"], (Real, Decimal)):
            local_scalar(raw["threshold"], "ZScore threshold")
        validate_detector_warnings(raw["warnings"])
    return raw

ZScoreCalculator

Bases: BaseCalculator

Fit per-column mean and population std (ddof=0) for z-score filtering.

Source code in skyulf-core/skyulf/preprocessing/outliers/zscore.py

@NodeRegistry.register("ZScore", ZScoreApplier)
@node_meta(
    id="ZScore",
    name="Z-Score Outlier Removal",
    category="Preprocessing",
    description="Remove outliers using Z-Score.",
    params={"threshold": 3.0, "columns": []},
    learns_from_data=True,
)
class ZScoreCalculator(BaseCalculator):
    """Fit per-column mean and population std (``ddof=0``) for z-score filtering."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return the input schema unchanged: row filtering preserves the column set."""
        # Z-score removes outlier *rows*; column set is preserved.
        return input_schema

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> ZScoreArtifact:  # pylint: disable=arguments-differ
        """Record each column's mean/std; warn on empty or zero-variance columns."""
        if user_picked_no_columns(config):
            return {}

        threshold = config.get("threshold", 3.0)
        # TODO(pandas-removal): mean/std alone would be numpy-safe (Polars std
        # ddof=0 already matches Pandas — verified), but this loop relies on
        # per-column pd.to_numeric(errors="coerce") to gracefully skip
        # non-numeric/mixed-dtype columns and .dropna() per column before
        # computing stats. Revisit once there's a native Polars equivalent
        # (pl.col(col).cast(pl.Float64, strict=False).drop_nulls()) wired in
        # to replace the coercion step without changing which columns get a
        # "non-numeric" warning.
        X_pd, cols = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
        if not cols:
            return {}

        stats: dict[str, dict[str, float]] = {}
        warnings = []
        for col in cols:
            series = pd.to_numeric(X_pd[col], errors="coerce").dropna()
            series = series[np.isfinite(series)]
            if series.empty:
                warnings.append(f"Column '{col}': Empty, non-numeric or non-finite")
                continue
            std = series.std(ddof=0)
            if std == 0:
                warnings.append(f"Column '{col}': Zero variance (std=0)")
                continue
            stats[col] = {"mean": series.mean(), "std": std}

        return {
            "type": "zscore",
            "stats": stats,
            "threshold": threshold,
            "warnings": warnings,
        }

fit(X, _y, config)

Record each column's mean/std; warn on empty or zero-variance columns.

Source code in skyulf-core/skyulf/preprocessing/outliers/zscore.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> ZScoreArtifact:  # pylint: disable=arguments-differ
    """Record each column's mean/std; warn on empty or zero-variance columns."""
    if user_picked_no_columns(config):
        return {}

    threshold = config.get("threshold", 3.0)
    # TODO(pandas-removal): mean/std alone would be numpy-safe (Polars std
    # ddof=0 already matches Pandas — verified), but this loop relies on
    # per-column pd.to_numeric(errors="coerce") to gracefully skip
    # non-numeric/mixed-dtype columns and .dropna() per column before
    # computing stats. Revisit once there's a native Polars equivalent
    # (pl.col(col).cast(pl.Float64, strict=False).drop_nulls()) wired in
    # to replace the coercion step without changing which columns get a
    # "non-numeric" warning.
    X_pd, cols = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
    if not cols:
        return {}

    stats: dict[str, dict[str, float]] = {}
    warnings = []
    for col in cols:
        series = pd.to_numeric(X_pd[col], errors="coerce").dropna()
        series = series[np.isfinite(series)]
        if series.empty:
            warnings.append(f"Column '{col}': Empty, non-numeric or non-finite")
            continue
        std = series.std(ddof=0)
        if std == 0:
            warnings.append(f"Column '{col}': Zero variance (std=0)")
            continue
        stats[col] = {"mean": series.mean(), "std": std}

    return {
        "type": "zscore",
        "stats": stats,
        "threshold": threshold,
        "warnings": warnings,
    }

infer_output_schema(input_schema, config)

Return the input schema unchanged: row filtering preserves the column set.

Source code in skyulf-core/skyulf/preprocessing/outliers/zscore.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Return the input schema unchanged: row filtering preserves the column set."""
    # Z-score removes outlier *rows*; column set is preserved.
    return input_schema