Skip to content

API: preprocessing.scaling

skyulf.preprocessing.scaling

Scaling nodes package.

Split from a single 496-LOC module into per-scaler files

_common.py — shared subset-selection helpers standard.py — StandardScaler minmax.py — MinMaxScaler robust.py — RobustScaler maxabs.py — MaxAbsScaler

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

MaxAbsScalerApplier

Bases: BaseApplier

Divide the selected columns by their fitted maximum absolute value.

Source code in skyulf-core/skyulf/preprocessing/scaling/maxabs.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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

apply(X, _y, params)

Scale X by the fitted max-abs values; y passes through.

Source code in skyulf-core/skyulf/preprocessing/scaling/maxabs.py 
55
56
57
58
59
60
@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}
    )

inference_capability(state, *, engine) staticmethod

Describe local saved apply context without granting worker execution.

Source code in skyulf-core/skyulf/preprocessing/scaling/maxabs.py 
47
48
49
50
51
52
53
@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")

validate_inference_state(raw) staticmethod

Inspect saved local statistics without changing them or fitting data.

Source code in skyulf-core/skyulf/preprocessing/scaling/maxabs.py 
36
37
38
39
40
41
42
43
44
45
@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

MaxAbsScalerCalculator

Bases: BaseCalculator

Fit per-column maximum-absolute-value statistics via sklearn's MaxAbsScaler.

Source code in skyulf-core/skyulf/preprocessing/scaling/maxabs.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@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)

fit(X, _y, config)

Dispatch the fit to the active engine and return the max-abs statistics artifact.

Source code in skyulf-core/skyulf/preprocessing/scaling/maxabs.py 
117
118
119
120
121
122
123
124
125
@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}),
    )

infer_output_schema(input_schema, config)

Return a schema with transformed columns promoted to float64.

Source code in skyulf-core/skyulf/preprocessing/scaling/maxabs.py 
111
112
113
114
115
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)

MinMaxScalerApplier

Bases: 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.

Source code in skyulf-core/skyulf/preprocessing/scaling/minmax.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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
        ]
        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]
        X_out[valid] = vals
        return X_out, _y

apply(X, _y, params)

Apply the fitted affine scale/shift to X; y passes through.

Source code in skyulf-core/skyulf/preprocessing/scaling/minmax.py 
48
49
50
51
52
53
@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}
    )

resolve_fitted_config(raw, state) staticmethod

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

Source code in skyulf-core/skyulf/preprocessing/scaling/minmax.py 
41
42
43
44
45
46
@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)

validate_fitted_state(raw) staticmethod

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

Source code in skyulf-core/skyulf/preprocessing/scaling/minmax.py 
36
37
38
39
@staticmethod
def validate_fitted_state(raw: dict) -> dict:
    """Inspect this node's supported saved state without fitting or applying data."""
    return _minmax_state(raw)

MinMaxScalerCalculator

Bases: BaseCalculator

Fit scale/offset statistics that map the data into feature_range.

Source code in skyulf-core/skyulf/preprocessing/scaling/minmax.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@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], "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)

fit(X, _y, config)

Dispatch the fit to the active engine and return the scaler-statistics artifact.

Source code in skyulf-core/skyulf/preprocessing/scaling/minmax.py 
112
113
114
115
116
117
118
119
120
@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}),
    )

infer_output_schema(input_schema, config)

Return a schema with transformed columns promoted to float64.

Source code in skyulf-core/skyulf/preprocessing/scaling/minmax.py 
106
107
108
109
110
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)

RobustScalerApplier

Bases: BaseApplier

Center on the fitted median and scale by the fitted quantile spread.

Honors with_centering/with_scaling; y is never modified.

Source code in skyulf-core/skyulf/preprocessing/scaling/robust.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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."""
        if not local_state_fields(
            raw,
            "robust_scaler",
            {
                "type",
                "columns",
                "center",
                "scale",
                "quantile_range",
                "with_centering",
                "with_scaling",
            },
            allow_empty=True,
        ):
            return raw
        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

apply(X, _y, params)

Apply the fitted robust center/scale to X on the active engine.

Source code in skyulf-core/skyulf/preprocessing/scaling/robust.py 
74
75
76
77
78
79
@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}
    )

inference_capability(state, *, engine) staticmethod

Describe local saved apply context without granting worker execution.

Source code in skyulf-core/skyulf/preprocessing/scaling/robust.py 
66
67
68
69
70
71
72
@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")

validate_inference_state(raw) staticmethod

Inspect saved local flags and statistics without recalculating quantiles.

Source code in skyulf-core/skyulf/preprocessing/scaling/robust.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved local flags and statistics without recalculating quantiles."""
    if not local_state_fields(
        raw,
        "robust_scaler",
        {
            "type",
            "columns",
            "center",
            "scale",
            "quantile_range",
            "with_centering",
            "with_scaling",
        },
        allow_empty=True,
    ):
        return raw
    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

RobustScalerCalculator

Bases: BaseCalculator

Fit outlier-robust median/quantile-range statistics via sklearn's RobustScaler.

Source code in skyulf-core/skyulf/preprocessing/scaling/robust.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@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,
        "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)

fit(X, _y, config)

Dispatch the fit to the active engine and return the robust-statistics artifact.

Source code in skyulf-core/skyulf/preprocessing/scaling/robust.py 
169
170
171
172
173
174
175
176
177
@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}),
    )

infer_output_schema(input_schema, config)

Return a schema with transformed columns promoted to float64.

Source code in skyulf-core/skyulf/preprocessing/scaling/robust.py 
161
162
163
164
165
166
167
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)

StandardScalerApplier

Bases: 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.

Source code in skyulf-core/skyulf/preprocessing/scaling/standard.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
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.
                        divided = cast(
                            pl.Expr,
                            np.divide(e, 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

apply(X, _y, params)

Center and scale X with the fitted statistics on the active engine.

Source code in skyulf-core/skyulf/preprocessing/scaling/standard.py 
70
71
72
73
74
75
76
77
78
79
80
81
@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,
        },
    )

resolve_fitted_config(raw, state) staticmethod

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

Source code in skyulf-core/skyulf/preprocessing/scaling/standard.py 
65
66
67
68
@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)

validate_fitted_state(raw) staticmethod

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

Source code in skyulf-core/skyulf/preprocessing/scaling/standard.py 
60
61
62
63
@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)

StandardScalerCalculator

Bases: 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.

Source code in skyulf-core/skyulf/preprocessing/scaling/standard.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
@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)

fit(X, _y, config)

Dispatch the fit to the active engine and return the scaler-statistics artifact.

Source code in skyulf-core/skyulf/preprocessing/scaling/standard.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
@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,
            },
        ),
    )

infer_output_schema(input_schema, config)

Return a schema with touched numeric columns promoted to float64.

Source code in skyulf-core/skyulf/preprocessing/scaling/standard.py 
198
199
200
201
202
203
204
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)