API: preprocessing.outliers

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("_display_name", 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

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

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)

@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,
        }

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)

@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,
        }

class IsolationForestApplier(BaseApplier):
    """Exclude rows using the saved joint forest, retaining unscored nonfinite rows."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the fitted forest and its feature axes without running tree inference."""
        model = validate_detector(raw, "isolation_forest", IsolationForest)
        if model is None:
            return raw
        trees = getattr(model, "estimators_", None)
        axes = getattr(model, "estimators_features_", None)
        if (
            not isinstance(trees, list)
            or not trees
            or not isinstance(axes, list)
            or len(trees) != len(axes)
        ):
            raise ValueError("Fitted IsolationForest requires matching tree and feature lists.")
        for tree, features in zip(trees, axes, strict=True):
            _validate_tree(tree, features, len(raw["columns"]))
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Declare local row filtering only after validating the fitted detector."""
        if engine not in {"pandas", "polars"}:
            return None
        IsolationForestApplier.validate_inference_state(state)
        return ExecutionCapability(
            engine, "apply", "local", "filter" if state else "preserve", "row"
        )

    @apply_method
    def apply(self, X: Any, y: Any, params: dict) -> Any:
        """Apply one positional mask to features and targets on either native engine."""
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"pandas": filter_detector_rows, "polars": filter_detector_rows},
        )

@NodeRegistry.register("IsolationForest", IsolationForestApplier)
@node_meta(
    id="IsolationForest",
    name="Isolation Forest",
    category="Preprocessing",
    description="Exclude multivariate outliers using a learned joint isolation forest.",
    params={
        "columns": [],
        "n_estimators": 100,
        "max_samples": "auto",
        "contamination": 0.01,
        "random_state": 42,
    },
    learns_from_data=True,
)
class IsolationForestCalculator(BaseCalculator):
    """Fit one forest across selected numeric features, excluding the implicit target."""

    def infer_output_schema(self, input_schema: Any, config: dict) -> Any:
        """Row exclusion preserves input column names and dtypes."""
        return input_schema

    @fit_method
    def fit(self, X: Any, y: Any, config: dict) -> dict:
        """Learn only from finite training rows; malformed options and fit errors propagate."""
        if user_picked_no_columns(config):
            return {}
        return fit_dual_engine(
            (X, y) if y is not None else X, config, {"pandas": _fit, "polars": _fit}
        )

class LocalOutlierFactorApplier(BaseApplier):
    """Exclude unseen rows using the fitted training population as the neighbor reference."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Require novelty mode and internally consistent learned neighbor statistics."""
        model = validate_detector(raw, "local_outlier_factor", LocalOutlierFactor)
        if model is None:
            return raw
        if model.novelty is not True or model.metric != "minkowski" or model.p != 2:
            raise ValueError("Fitted LocalOutlierFactor requires Euclidean novelty scoring.")
        rows = getattr(model, "n_samples_fit_", 0)
        neighbors = getattr(model, "n_neighbors_", 0)
        if not 1 <= neighbors < rows:
            raise ValueError("Fitted LocalOutlierFactor neighbor count is invalid.")
        fitted_array(model._fit_X, (rows, len(raw["columns"])))
        fitted_array(model._lrd, (rows,), positive=True)
        fitted_array(model.negative_outlier_factor_, (rows,))
        fitted_array(model._distances_fit_X_, (rows, neighbors))
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Declare local row filtering against fixed learned neighbors."""
        if engine not in {"pandas", "polars"}:
            return None
        LocalOutlierFactorApplier.validate_inference_state(state)
        return ExecutionCapability(
            engine, "apply", "local", "filter" if state else "preserve", "row"
        )

    @apply_method
    def apply(self, X: Any, y: Any, params: dict) -> Any:
        """Score new rows only, retaining nonfinite rows for explicit downstream handling."""
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"pandas": filter_detector_rows, "polars": filter_detector_rows},
        )

@NodeRegistry.register("LocalOutlierFactor", LocalOutlierFactorApplier)
@node_meta(
    id="LocalOutlierFactor",
    name="Local Outlier Factor",
    category="Preprocessing",
    description="Exclude joint local-density outliers; use fixed neighbors for new rows.",
    params={"columns": [], "n_neighbors": 20, "contamination": 0.01},
    learns_from_data=True,
)
class LocalOutlierFactorCalculator(BaseCalculator):
    """Learn LOF training scores and novelty state with separate train/apply representations."""

    def infer_output_schema(self, input_schema: Any, config: dict) -> Any:
        """Filtering rows retains all input column names and dtypes."""
        return input_schema

    @fit_method
    def fit(self, X: Any, y: Any, config: dict) -> dict:
        """Fit a saved detector; use fit_transform_train to obtain filtered training data."""
        if user_picked_no_columns(config):
            return {}
        return fit_dual_engine(
            (X, y) if y is not None else X, config, {"pandas": _fit, "polars": _fit}
        )

    def fit_transform_train(self, df: Any, config: dict) -> tuple[dict, Any]:
        """Filter training rows with negative_outlier_factor_, never novelty predict."""
        if user_picked_no_columns(config):
            return {}, df
        return fit_transform_train_dual_engine(
            df, config, {"pandas": _fit_train, "polars": _fit_train}
        )

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)

@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", {})}

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."""
        fields = {"type", "bounds", "lower_percentile", "upper_percentile", "warnings"}
        if isinstance(raw, dict):
            fields.update({"percentiles_by_column"}.intersection(raw))
        if not local_state_fields(
            raw,
            "winsorize",
            fields,
            allow_empty=True,
        ):
            return raw
        validate_fitted_bounds(raw["bounds"], partial=False)
        _validate_saved_percentiles(raw)
        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