API: preprocessing.scaling

class MaxAbsScalerApplier(BaseApplier):
    """Divide the selected columns by their fitted maximum absolute value."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved local statistics without changing them or fitting data."""
        if local_state_fields(
            raw, "maxabs_scaler", {"type", "columns", "scale", "max_abs"}, allow_empty=True
        ):
            columns = _columns(raw["columns"])
            validate_scaler_vector(raw["scale"], len(columns), nonnegative=True)
            validate_scaler_vector(raw["max_abs"], len(columns), nonnegative=True)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe local saved apply context without granting worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        MaxAbsScalerApplier.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
        """Scale ``X`` by the fitted max-abs values; ``y`` passes through."""
        return apply_dual_engine(
            X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

    @staticmethod
    def _apply_polars(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        scale = params.get("scale")
        valid = resolve_valid_columns(X, cols)
        if not valid or scale is None:
            return X, _y

        exprs = []
        for col_name in valid:
            s = scale[cols.index(col_name)]
            # NumPy literals carry dtype metadata that Series scalar arithmetic discards.
            column = (
                pl.col(col_name)
                if isinstance(X, pl.LazyFrame) or isinstance(s, np.generic)
                else X.get_column(col_name)
            )
            exprs.append(divide_polars_column(X, column, s).alias(col_name))
        return X.with_columns(exprs), _y

    @staticmethod
    def _apply_pandas(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        scale = params.get("scale")
        valid = resolve_valid_columns(X, cols)
        if not valid or scale is None:
            return X, _y

        X_out = X.copy()
        col_indices = [cols.index(c) for c in valid]
        subset = decimal_columns_to_float(X_out[valid], valid)
        vals = subset.to_numpy(dtype=np.float64, na_value=np.nan)
        vals = vals / safe_scale(np.array(scale)[col_indices].copy())
        X_out[valid] = vals
        return X_out, _y

@NodeRegistry.register("MaxAbsScaler", MaxAbsScalerApplier)
@node_meta(
    id="MaxAbsScaler",
    name="MaxAbs Scaler",
    category="Preprocessing",
    description="Scale each feature by its maximum absolute value.",
    params={"columns": []},
    learns_from_data=True,
)
class MaxAbsScalerCalculator(BaseCalculator):
    """Fit per-column maximum-absolute-value statistics via sklearn's ``MaxAbsScaler``."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return a schema with transformed columns promoted to ``float64``."""
        return promote_configured_columns_to_float64(input_schema, config)

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> MaxAbsScalerArtifact:  # pylint: disable=arguments-differ
        """Dispatch the fit to the active engine and return the max-abs statistics artifact."""
        if user_picked_no_columns(config):
            return cast(MaxAbsScalerArtifact, {})
        return cast(
            MaxAbsScalerArtifact,
            fit_dual_engine(X, config, {"polars": self._fit_polars, "pandas": self._fit_pandas}),
        )

    @staticmethod
    def _fit_polars(X: Any, _y: Any, config: dict[str, Any]) -> dict[str, Any]:
        cols, X_subset = _select_subset_polars(X, config)
        if not cols:
            return {}
        return _fit_maxabs(X_subset, cols)

    @staticmethod
    def _fit_pandas(X: Any, _y: Any, config: dict[str, Any]) -> dict[str, Any]:
        cols, X_subset = _select_subset_pandas(X, config)
        if not cols:
            return {}
        return _fit_maxabs(X_subset, cols)

class MinMaxScalerApplier(BaseApplier):
    """Rescale selected columns or replay an explicit empty selection unchanged.

    Empty artifacts declare identity only with an explicit empty column recipe.
    Automatic selections and non-finite learned coefficients remain unreviewed.
    """

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

    @staticmethod
    def resolve_fitted_config(raw: dict, state: dict) -> dict:
        """Bind inference configuration to this node's inspected fitted artifact."""
        if not state and (not user_picked_no_columns(raw) or raw.get("_auto_columns")):
            raise ValueError("An empty scaler artifact requires explicit empty columns.")
        return _minmax_config(fitted_columns(_minmax_values(raw), state or {"columns": []}), state)

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Apply the fitted affine scale/shift to ``X``; ``y`` passes through."""
        return apply_dual_engine(
            X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

    @staticmethod
    def _apply_polars(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        min_val = params.get("min")
        scale = params.get("scale")
        valid = resolve_valid_columns(X, cols)
        if not valid or min_val is None or scale is None:
            return X, _y

        exprs = [
            (pl.col(c) * scale[cols.index(c)] + min_val[cols.index(c)]).alias(c) for c in valid
        ]
        if params.get("clip", False):
            lower, upper = params["feature_range"]
            exprs = [expression.clip(lower, upper) for expression in exprs]
        return X.with_columns(exprs), _y

    @staticmethod
    def _apply_pandas(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        min_val = params.get("min")
        scale = params.get("scale")
        valid = resolve_valid_columns(X, cols)
        if not valid or min_val is None or scale is None:
            return X, _y

        X_out = X.copy()
        col_indices = [cols.index(c) for c in valid]
        subset = decimal_columns_to_float(X_out[valid], valid)
        vals = subset.to_numpy(dtype=np.float64, na_value=np.nan)
        vals = vals * np.array(scale)[col_indices] + np.array(min_val)[col_indices]
        if params.get("clip", False):
            vals = np.clip(vals, *params["feature_range"])
        X_out[valid] = vals
        return X_out, _y

@NodeRegistry.register(
    "MinMaxScaler",
    MinMaxScalerApplier,
    execution_capabilities=(
        ExecutionCapability("pandas", "apply", "python_batch", "preserve", "row"),
        ExecutionCapability("polars", "apply", "local", "preserve", "row"),
    ),
)
@node_meta(
    id="MinMaxScaler",
    name="Min-Max Scaler",
    category="Preprocessing",
    description="Transform features by scaling each feature to a given range.",
    params={"feature_range": [0, 1], "clip": False, "columns": []},
    learns_from_data=True,
)
class MinMaxScalerCalculator(BaseCalculator):
    """Fit scale/offset statistics that map the data into ``feature_range``."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return a schema with transformed columns promoted to ``float64``."""
        return promote_configured_columns_to_float64(input_schema, config)

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> MinMaxScalerArtifact:  # pylint: disable=arguments-differ
        """Dispatch the fit to the active engine and return the scaler-statistics artifact."""
        if user_picked_no_columns(config):
            return cast(MinMaxScalerArtifact, {})
        return cast(
            MinMaxScalerArtifact,
            fit_dual_engine(X, config, {"polars": self._fit_polars, "pandas": self._fit_pandas}),
        )

    @staticmethod
    def _fit_polars(X: Any, _y: Any, config: dict[str, Any]) -> dict[str, Any]:
        cols, X_subset = _select_subset_polars(X, config)
        if not cols:
            return {}
        return _fit_minmax(X_subset, cols, config)

    @staticmethod
    def _fit_pandas(X: Any, _y: Any, config: dict[str, Any]) -> dict[str, Any]:
        cols, X_subset = _select_subset_pandas(X, config)
        if not cols:
            return {}
        return _fit_minmax(X_subset, cols, config)

class NormalizerApplier(BaseApplier):
    """Normalize each row with the saved norm and selected feature axes."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the saved native normalizer without deriving request-dependent statistics."""
        transformer = validate_scaler_object(raw, "normalizer", Normalizer)
        if transformer is not None and transformer.norm not in ("l1", "l2", "max"):
            raise ValueError("Fitted Normalizer norm must be l1, l2 or max.")
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe local row normalization without granting worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        NormalizerApplier.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:
        """Normalize selected columns on the active engine while retaining paired labels."""
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"pandas": apply_scaler_pandas, "polars": apply_scaler_polars},
        )

@NodeRegistry.register("Normalizer", NormalizerApplier)
@node_meta(
    id="Normalizer",
    name="Normalizer",
    category="Preprocessing",
    description="Normalize each selected feature vector to unit norm.",
    params={"columns": [], "norm": "l2"},
    learns_from_data=False,
)
class NormalizerCalculator(BaseCalculator):
    """Validate the training matrix and save sklearn's norm and feature width."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema | None:
        """Keep all names and promote selected numeric features to float64."""
        return scaler_output_schema(input_schema, config)

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> NormalizerArtifact:
        """Fit the native normalizer for consistent validation and saved feature axes."""
        if user_picked_no_columns(config):
            return {}
        values, columns = scaler_fit_values(X, config)
        if not columns:
            return {}
        transformer = Normalizer(norm=config.get("norm", "l2")).fit(values)
        return {"type": "normalizer", "columns": columns, "transformer_object": transformer}

class QuantileTransformerApplier(BaseApplier):
    """Replay the learned marginal distributions without fitting on inference requests."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the native saved transformer and its learned numeric axes."""
        transformer = validate_scaler_object(raw, "quantile_transformer", QuantileTransformer)
        if transformer is not None:
            _validate_quantile_matrix(transformer, len(raw["columns"]))
            _validate_quantile_references(transformer)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe local row-preserving replay without granting worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        QuantileTransformerApplier.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:
        """Transform saved columns on the request's engine and preserve paired labels."""
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"pandas": apply_scaler_pandas, "polars": apply_scaler_polars},
        )

@NodeRegistry.register("QuantileTransformer", QuantileTransformerApplier)
@node_meta(
    id="QuantileTransformer",
    name="Quantile Transformer",
    category="Preprocessing",
    description="Map numeric features to a uniform or normal marginal distribution.",
    params={
        "columns": [],
        "n_quantiles": 1000,
        "output_distribution": "uniform",
        "subsample": 10000,
        "random_state": DEFAULT_RANDOM_STATE,
    },
    learns_from_data=True,
)
class QuantileTransformerCalculator(BaseCalculator):
    """Fit sklearn quantiles on the selected numeric training features only."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema | None:
        """Keep the column set and promote selected features to float64."""
        return scaler_output_schema(input_schema, config)

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> QuantileTransformerArtifact:
        """Learn deterministic quantiles; sklearn caps their count at the training row count."""
        if user_picked_no_columns(config):
            return {}
        values, columns = scaler_fit_values(X, config)
        if not columns:
            return {}
        for name in ("n_quantiles", "subsample"):
            if isinstance(config.get(name), (bool, np.bool_)):
                raise ValueError(
                    f"QuantileTransformer {name} must be a positive integer, not bool."
                )
        transformer = QuantileTransformer(
            n_quantiles=config.get("n_quantiles", 1000),
            output_distribution=config.get("output_distribution", "uniform"),
            subsample=config.get("subsample", 10000),
            random_state=config.get("random_state", DEFAULT_RANDOM_STATE),
        ).fit(values)
        return {
            "type": "quantile_transformer",
            "columns": columns,
            "transformer_object": transformer,
        }

class RobustScalerApplier(BaseApplier):
    """Center on the fitted median and scale by the fitted quantile spread.

    Honors ``with_centering``/``with_scaling``; ``y`` is never modified.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved local flags and statistics without recalculating quantiles."""
        fields = {
            "type",
            "columns",
            "center",
            "scale",
            "quantile_range",
            "with_centering",
            "with_scaling",
        }
        if isinstance(raw, dict) and "unit_variance" in raw:
            fields.add("unit_variance")
        if not local_state_fields(
            raw,
            "robust_scaler",
            fields,
            allow_empty=True,
        ):
            return raw
        local_boolean(raw.get("unit_variance", False), "unit_variance")
        columns = _columns(raw["columns"])
        if type(raw["quantile_range"]) not in (list, tuple):
            raise ValueError("Fitted robust scaler quantiles must be a pair of bounds.")
        validate_scaling_range(raw["quantile_range"], "quantile_range")
        _validate_robust_statistic(raw, "with_centering", "center", len(columns))
        _validate_robust_statistic(raw, "with_scaling", "scale", len(columns))
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe local saved apply context without granting worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        RobustScalerApplier.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
        """Apply the fitted robust center/scale to ``X`` on the active engine."""
        return apply_dual_engine(
            X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

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

        with_centering = params.get("with_centering", True)
        with_scaling = params.get("with_scaling", True)
        if not any((with_centering, with_scaling)):
            return X, _y

        exprs = []
        for col_name in valid:
            i = cols.index(col_name)
            column = pl.col(col_name) if isinstance(X, pl.LazyFrame) else X.get_column(col_name)
            e = column.cast(pl.Float64)
            if with_centering and center is not None:
                e = e - center[i]
            if with_scaling and scale is not None:
                s = scale[i]
                e = divide_polars_column(X, e, s)
            exprs.append(e.alias(col_name))
        return X.with_columns(exprs), _y

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

        if not (params.get("with_centering", True) or params.get("with_scaling", True)):
            return X, _y

        X_out = X.copy()
        col_indices = [cols.index(c) for c in valid]
        subset = decimal_columns_to_float(X_out[valid], valid)
        vals = subset.to_numpy(dtype=np.float64, na_value=np.nan)

        if params.get("with_centering", True) and center is not None:
            vals = vals - np.array(center)[col_indices]
        if params.get("with_scaling", True) and scale is not None:
            vals = vals / safe_scale(np.array(scale)[col_indices].copy())

        X_out[valid] = vals
        return X_out, _y

@NodeRegistry.register("RobustScaler", RobustScalerApplier)
@node_meta(
    id="RobustScaler",
    name="Robust Scaler",
    category="Preprocessing",
    description="Scale features using statistics that are robust to outliers.",
    params={
        "quantile_range": [25.0, 75.0],
        "with_centering": True,
        "with_scaling": True,
        "unit_variance": False,
        "columns": [],
    },
    learns_from_data=True,
)
class RobustScalerCalculator(BaseCalculator):
    """Fit outlier-robust median/quantile-range statistics via sklearn's ``RobustScaler``."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return a schema with transformed columns promoted to ``float64``."""
        if not (config.get("with_centering", True) or config.get("with_scaling", True)):
            return input_schema
        return promote_configured_columns_to_float64(input_schema, config)

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> RobustScalerArtifact:  # pylint: disable=arguments-differ
        """Dispatch the fit to the active engine and return the robust-statistics artifact."""
        if user_picked_no_columns(config):
            return cast(RobustScalerArtifact, {})
        return cast(
            RobustScalerArtifact,
            fit_dual_engine(X, config, {"polars": self._fit_polars, "pandas": self._fit_pandas}),
        )

    @staticmethod
    def _fit_polars(X: Any, _y: Any, config: dict[str, Any]) -> dict[str, Any]:
        cols, X_subset = _select_subset_polars(X, config)
        if not cols:
            return {}
        return _fit_robust(X_subset, cols, config)

    @staticmethod
    def _fit_pandas(X: Any, _y: Any, config: dict[str, Any]) -> dict[str, Any]:
        cols, X_subset = _select_subset_pandas(X, config)
        if not cols:
            return {}
        return _fit_robust(X_subset, cols, config)

class StandardScalerApplier(BaseApplier):
    """Standardize the selected columns using the fitted mean and scale.

    Honors ``with_mean``/``with_std``; ``X`` passes through unchanged when the
    enabled flags have no fitted statistics, and ``y`` is never modified.
    Spark validates portable state and applies native expressions without a
    data action. Its selected numeric columns become double when scaling is enabled.
    """

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

    @staticmethod
    def resolve_fitted_config(raw: dict, state: dict) -> dict:
        """Bind inference configuration to this node's inspected fitted artifact."""
        return portable_config("StandardScaler", raw, state)

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Center and scale ``X`` with the fitted statistics on the active engine."""
        return apply_dual_engine(
            X,
            params,
            {
                "polars": self._apply_polars,
                "pandas": self._apply_pandas,
                "spark": apply_spark_standard,
            },
        )

    @staticmethod
    def _apply_polars(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        mean = params.get("mean")
        scale = params.get("scale")
        valid = resolve_valid_columns(X, cols)
        with_mean = params.get("with_mean", True)
        with_std = params.get("with_std", True)
        if _needs_fitted_artifact(valid, with_mean, mean, with_std, scale):
            return X, _y

        mean_arr = np.array(mean) if mean is not None else np.zeros(len(cols))
        scale_arr = np.array(scale) if scale is not None else np.ones(len(cols))
        schema = X.collect_schema()

        def _standardized_expr(col_name: str) -> Any:
            idx = cols.index(col_name)
            e: Any = pl.col(col_name)
            if with_mean:
                e = e - mean_arr[idx]
            if with_std:
                s = scale_arr[idx]
                divided = e / (s if s != 0 else 1.0)
                dtype = schema[col_name]
                if s != 0 and (
                    dtype in (pl.Float32, pl.Float64) or (dtype.is_integer() and dtype != pl.Int128)
                ):
                    output_dtype = X.lazy().select(divided).collect_schema()[col_name]
                    if output_dtype in (pl.Float32, pl.Float64):
                        # Use Polars' native promotion with batch-independent NumPy division.
                        # Polars 2 centers unsigned minus signed as Int128, which NumPy cannot
                        # view; casting first is the same promotion NumPy applies itself.
                        divided = cast(
                            pl.Expr,
                            np.divide(
                                e.cast(output_dtype),
                                scale_arr[idx : idx + 1],
                                dtype=str(output_dtype).lower(),
                            ),
                        )
                e = divided
            return e.alias(col_name)

        exprs = [_standardized_expr(col_name) for col_name in valid]
        return X.with_columns(exprs), _y

    @staticmethod
    def _apply_pandas(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        mean = params.get("mean")
        scale = params.get("scale")
        valid = resolve_valid_columns(X, cols)
        with_mean = params.get("with_mean", True)
        with_std = params.get("with_std", True)
        if not (with_mean or with_std):
            return X, _y
        if _needs_fitted_artifact(valid, with_mean, mean, with_std, scale):
            return X, _y

        X_out = X.copy()
        col_indices = [cols.index(c) for c in valid]
        subset = decimal_columns_to_float(X_out[valid], valid)
        vals = subset.to_numpy(dtype=np.float64, na_value=np.nan)
        if with_mean:
            vals = vals - np.array(mean)[col_indices]
        if with_std:
            vals = vals / safe_scale(np.array(scale)[col_indices].copy())

        X_out[valid] = vals
        return X_out, _y

@NodeRegistry.register(
    "StandardScaler",
    StandardScalerApplier,
    execution_capabilities=tuple(
        ExecutionCapability(
            "spark",
            operation,
            "native",
            "preserve",
            "global" if operation == "fit" else "row",
            codec_version=1,
            config_match=(("with_mean", with_mean), ("with_std", with_std)),
        )
        for operation in ("fit", "apply")
        for with_mean in (True, False)
        for with_std in (True, False)
    )
    + tuple(
        ExecutionCapability(
            engine,
            "apply",
            execution_kind,
            "preserve",
            "row",
            codec_version=1,
            config_match=(("with_mean", with_mean), ("with_std", with_std)),
        )
        for engine, execution_kind in (("pandas", "python_batch"), ("polars", "local"))
        for with_mean in (True, False)
        for with_std in (True, False)
    ),
)
@node_meta(
    id="StandardScaler",
    name="Standard Scaler",
    category="Preprocessing",
    description="Standardize features by removing the mean and scaling to unit variance.",
    params={"columns": [], "with_mean": True, "with_std": True},
    learns_from_data=True,
)
class StandardScalerCalculator(BaseCalculator):
    """Fit per-column statistics using sklearn locally or native Spark aggregates.

    Spark uses population variance (``ddof=0``), ignores null/NaN observations,
    and returns at most two statistics rows to the driver. Both options default
    to True. Empty explicit training selections produce a no-op artifact;
    selected features with zero training rows cannot be fitted.
    """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return a schema with touched numeric columns promoted to ``float64``."""
        if not (config.get("with_mean", True) or config.get("with_std", True)):
            return input_schema
        return promote_configured_columns_to_float64(input_schema, config)

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> StandardScalerArtifact:  # pylint: disable=arguments-differ
        """Dispatch the fit to the active engine and return the scaler-statistics artifact."""
        if user_picked_no_columns(config):
            return cast(StandardScalerArtifact, {})
        return cast(
            StandardScalerArtifact,
            fit_dual_engine(
                X,
                config,
                {
                    "polars": self._fit_polars,
                    "pandas": self._fit_pandas,
                    "spark": fit_spark_standard,
                },
            ),
        )

    @staticmethod
    def _fit_polars(X: Any, _y: Any, config: dict[str, Any]) -> dict[str, Any]:
        cols, X_subset = _select_subset_polars(X, config)
        if not cols:
            return {}
        return _fit_standard(X_subset, cols, config)

    @staticmethod
    def _fit_pandas(X: Any, _y: Any, config: dict[str, Any]) -> dict[str, Any]:
        cols, X_subset = _select_subset_pandas(X, config)
        if not cols:
            return {}
        return _fit_standard(X_subset, cols, config)