Skip to content

API: preprocessing.transformations

skyulf.preprocessing.transformations

Transformation nodes package.

Importing this package registers PowerTransformer / SimpleTransformation / GeneralTransformation with the node registry and re-exports their public classes.

GeneralTransformationApplier

Bases: BaseApplier

Apply per-column simple ops or fitted Box-Cox/Yeo-Johnson transforms.

Source code in skyulf-core/skyulf/preprocessing/transformations/general.py 
 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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
class GeneralTransformationApplier(BaseApplier):
    """Apply per-column simple ops or fitted Box-Cox/Yeo-Johnson transforms."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect ordered saved rules while retaining the existing simple-rule checks."""
        local_state_fields(raw, "general_transformation", {"type", "transformations"})
        rules = raw["transformations"]
        if type(rules) not in (list, tuple):
            raise ValueError("GeneralTransformation rules must be a list or tuple.")
        for rule in rules:
            _validate_general_rule(rule)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Separate pointwise formulas from batch-dependent power failure fallback."""
        if engine not in ("pandas", "polars"):
            return None
        GeneralTransformationApplier.validate_inference_state(state)
        has_power = any(rule.get("method") in _POWER_METHODS for rule in state["transformations"])
        return ExecutionCapability(
            engine, "apply", "local", "preserve", "global" if has_power else "row"
        )

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Run each configured transformation on the active engine; ``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]:
        transformations = params.get("transformations", [])
        if not transformations:
            return X, _y

        X_out = X
        for item in transformations:
            col = item.get("column")
            method = item.get("method")
            if col not in X_out.columns:
                continue
            if method in _POWER_METHODS:
                X_out = _apply_power_to_polars_col(X_out, item)
                continue
            op = _POLARS_OPS.get(method)
            if op is None:
                continue
            X_out = _apply_polars_op(X_out, item, op)
        return X_out, _y

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

        df_out = X.copy()
        for item in transformations:
            col = item.get("column")
            method = item.get("method")
            if col not in df_out.columns:
                continue
            if method in _POWER_METHODS:
                df_out = _apply_power_to_pandas_col(df_out, item)
                continue
            op = _PANDAS_OPS.get(method)
            if op is None:
                continue
            df_out[col] = op(pd.to_numeric(df_out[col], errors="coerce"), item)
        return df_out, _y

apply(X, _y, params)

Run each configured transformation on the active engine; y passes through.

Source code in skyulf-core/skyulf/preprocessing/transformations/general.py 
114
115
116
117
118
119
@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Run each configured transformation on the active engine; ``y`` passes through."""
    return apply_dual_engine(
        X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Separate pointwise formulas from batch-dependent power failure fallback.

Source code in skyulf-core/skyulf/preprocessing/transformations/general.py 
103
104
105
106
107
108
109
110
111
112
@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Separate pointwise formulas from batch-dependent power failure fallback."""
    if engine not in ("pandas", "polars"):
        return None
    GeneralTransformationApplier.validate_inference_state(state)
    has_power = any(rule.get("method") in _POWER_METHODS for rule in state["transformations"])
    return ExecutionCapability(
        engine, "apply", "local", "preserve", "global" if has_power else "row"
    )

validate_inference_state(raw) staticmethod

Inspect ordered saved rules while retaining the existing simple-rule checks.

Source code in skyulf-core/skyulf/preprocessing/transformations/general.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect ordered saved rules while retaining the existing simple-rule checks."""
    local_state_fields(raw, "general_transformation", {"type", "transformations"})
    rules = raw["transformations"]
    if type(rules) not in (list, tuple):
        raise ValueError("GeneralTransformation rules must be a list or tuple.")
    for rule in rules:
        _validate_general_rule(rule)
    return raw

GeneralTransformationCalculator

Bases: BaseCalculator

Resolve configured transformations, fitting lambdas for the power methods.

Rules run in list order. Each power rule is fitted on the training values produced by earlier rules on the same column, matching artifact replay. Each power rule accepts standardize (default True). Set it to False to apply only the fitted Box-Cox or Yeo-Johnson transform. The artifact retains the choice for subsequent transforms.

Source code in skyulf-core/skyulf/preprocessing/transformations/general.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
@NodeRegistry.register("GeneralTransformation", GeneralTransformationApplier)
@node_meta(
    id="GeneralTransformation",
    name="General Transformation",
    category="Preprocessing",
    description="Apply various function transformations (log, sqrt, square, exp) to columns.",
    params={"transformations": []},
    learns_from_data=True,
)
class GeneralTransformationCalculator(BaseCalculator):
    """Resolve configured transformations, fitting lambdas for the power methods.

    Rules run in list order. Each power rule is fitted on the training values
    produced by earlier rules on the same column, matching artifact replay.
    Each power rule accepts ``standardize`` (default ``True``). Set it to
    ``False`` to apply only the fitted Box-Cox or Yeo-Johnson transform.
    The artifact retains the choice for subsequent transforms.
    """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return updated dtypes for transformed numeric columns."""
        schema = input_schema
        touched: set[str] = set()
        for item in config.get("transformations", []):
            col = item.get("column")
            method = item.get("method")
            if col not in schema.columns or method is None:
                continue
            if method in _FLOAT_METHODS:
                touched.add(col)

        for col in touched:
            schema = schema.with_dtype(col, "float64")
        return schema

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> GeneralTransformationArtifact:  # pylint: disable=arguments-differ
        """Fit power lambdas in rule order; retain simple ops in the artifact.

        Box-Cox is skipped (with a warning) for non-positive values at that
        point in the rule sequence. A fit failure logs and skips that rule.
        """
        # Config: {'transformations': [{'column': 'col1', 'method': 'log'},
        #                              {'column': 'col2', 'method': 'yeo-johnson'}]}
        is_polars = get_engine(X).name == EngineName.POLARS
        fitted_transformations: list[dict[str, Any]] = []
        transformations = config.get("transformations", [])
        last_power_rule = {
            item.get("column"): index
            for index, item in enumerate(transformations)
            if item.get("method") in _POWER_METHODS
        }
        working_X = X
        applier = GeneralTransformationApplier()

        for index, item in enumerate(transformations):
            col = item.get("column")
            if col not in X.columns:
                continue

            fitted_item = _fit_transformation_rule(working_X, col, item, is_polars)
            if fitted_item is None:
                continue

            fitted_transformations.append(fitted_item)
            # Only materialize rules whose output is needed by a later fit.
            # Reuse replay semantics without mutating the caller's training data.
            if index < last_power_rule.get(col, -1):
                # apply_method exposes the public (data, params) call signature.
                working_X = applier.apply(  # pylint: disable=no-value-for-parameter
                    working_X, {"transformations": [fitted_item]}
                )

        return {
            "type": "general_transformation",
            "transformations": fitted_transformations,
        }

fit(X, _y, config)

Fit power lambdas in rule order; retain simple ops in the artifact.

Box-Cox is skipped (with a warning) for non-positive values at that point in the rule sequence. A fit failure logs and skips that rule.

Source code in skyulf-core/skyulf/preprocessing/transformations/general.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> GeneralTransformationArtifact:  # pylint: disable=arguments-differ
    """Fit power lambdas in rule order; retain simple ops in the artifact.

    Box-Cox is skipped (with a warning) for non-positive values at that
    point in the rule sequence. A fit failure logs and skips that rule.
    """
    # Config: {'transformations': [{'column': 'col1', 'method': 'log'},
    #                              {'column': 'col2', 'method': 'yeo-johnson'}]}
    is_polars = get_engine(X).name == EngineName.POLARS
    fitted_transformations: list[dict[str, Any]] = []
    transformations = config.get("transformations", [])
    last_power_rule = {
        item.get("column"): index
        for index, item in enumerate(transformations)
        if item.get("method") in _POWER_METHODS
    }
    working_X = X
    applier = GeneralTransformationApplier()

    for index, item in enumerate(transformations):
        col = item.get("column")
        if col not in X.columns:
            continue

        fitted_item = _fit_transformation_rule(working_X, col, item, is_polars)
        if fitted_item is None:
            continue

        fitted_transformations.append(fitted_item)
        # Only materialize rules whose output is needed by a later fit.
        # Reuse replay semantics without mutating the caller's training data.
        if index < last_power_rule.get(col, -1):
            # apply_method exposes the public (data, params) call signature.
            working_X = applier.apply(  # pylint: disable=no-value-for-parameter
                working_X, {"transformations": [fitted_item]}
            )

    return {
        "type": "general_transformation",
        "transformations": fitted_transformations,
    }

infer_output_schema(input_schema, config)

Return updated dtypes for transformed numeric columns.

Source code in skyulf-core/skyulf/preprocessing/transformations/general.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Return updated dtypes for transformed numeric columns."""
    schema = input_schema
    touched: set[str] = set()
    for item in config.get("transformations", []):
        col = item.get("column")
        method = item.get("method")
        if col not in schema.columns or method is None:
            continue
        if method in _FLOAT_METHODS:
            touched.add(col)

    for col in touched:
        schema = schema.with_dtype(col, "float64")
    return schema

PowerTransformerApplier

Bases: BaseApplier

Apply the fitted Box-Cox / Yeo-Johnson transform to the selected columns.

Source code in skyulf-core/skyulf/preprocessing/transformations/power.py 
 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
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
class PowerTransformerApplier(BaseApplier):
    """Apply the fitted Box-Cox / Yeo-Johnson transform to the selected columns."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the saved lambdas and scaler without normalizing or relearning them."""
        if local_state_fields(
            raw,
            "power_transformer",
            {"type", "columns", "method", "standardize", "lambdas", "scaler_params"},
            allow_empty=True,
        ):
            columns = raw["columns"]
            if type(columns) is not list or any(not isinstance(col, str) for col in columns):
                raise ValueError("Fitted power columns must be a list of strings.")
            validate_power_parameters(raw, len(columns))
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Expose whole-request fallback when any value prevents a power transform."""
        if engine not in ("pandas", "polars"):
            return None
        PowerTransformerApplier.validate_inference_state(state)
        context = "global" if state.get("columns") else "row"
        return ExecutionCapability(engine, "apply", "local", "preserve", context)

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Rebuild the transformer from stored lambdas and transform ``X``; ``y`` untouched.

        Fails open, like every applier in ``preprocessing/transformations/``: a
        failed transform is logged at ERROR with its traceback and leaves the data
        unchanged, so a degraded frame beats a hard failure at inference time.
        Every path that leaves data untransformed logs — including the one that
        used not to, a frame missing columns the transformer was fitted on, which
        is train/serve skew rather than a no-op and is warned about by name.
        """
        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", [])
        if params.get("lambdas") is None:
            return X, _y
        valid_cols = _fitted_columns_present(X, cols)
        if not valid_cols:
            return X, _y

        try:
            X_vals = X.select(valid_cols).to_numpy()
            X_trans = _power_transform_array(X_vals, params, cols, valid_cols)
            series = [pl.Series(name, X_trans[:, i]) for i, name in enumerate(valid_cols)]
            return X.with_columns(series), _y
        except Exception:  # noqa: BLE001 - transform failure is logged; frame left unchanged
            logger.exception("PowerTransformer (Polars) application failed")
            return X, _y

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

        try:
            # Computed before the copy so a failure hands back the caller's frame
            # rather than one whose columns were already cast to float64, matching
            # what the polars path returns.
            subset = decimal_columns_to_float(X[valid_cols], valid_cols)
            X_vals = subset.to_numpy(dtype=np.float64, na_value=np.nan)
            X_trans = np.asarray(_power_transform_array(X_vals, params, cols, valid_cols))
            df_out = X.copy()
            # Replace selected columns so nullable integers and Decimal objects
            # can receive float results without incompatible-dtype assignment.
            df_out[valid_cols] = X_trans
        except Exception:  # noqa: BLE001 - transform failure is logged; frame left unchanged
            logger.exception("PowerTransformer (Pandas) application failed")
            return X, _y
        return df_out, _y

apply(X, _y, params)

Rebuild the transformer from stored lambdas and transform X; y untouched.

Fails open, like every applier in preprocessing/transformations/: a failed transform is logged at ERROR with its traceback and leaves the data unchanged, so a degraded frame beats a hard failure at inference time. Every path that leaves data untransformed logs — including the one that used not to, a frame missing columns the transformer was fitted on, which is train/serve skew rather than a no-op and is warned about by name.

Source code in skyulf-core/skyulf/preprocessing/transformations/power.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Rebuild the transformer from stored lambdas and transform ``X``; ``y`` untouched.

    Fails open, like every applier in ``preprocessing/transformations/``: a
    failed transform is logged at ERROR with its traceback and leaves the data
    unchanged, so a degraded frame beats a hard failure at inference time.
    Every path that leaves data untransformed logs — including the one that
    used not to, a frame missing columns the transformer was fitted on, which
    is train/serve skew rather than a no-op and is warned about by name.
    """
    return apply_dual_engine(
        X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Expose whole-request fallback when any value prevents a power transform.

Source code in skyulf-core/skyulf/preprocessing/transformations/power.py 
112
113
114
115
116
117
118
119
@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Expose whole-request fallback when any value prevents a power transform."""
    if engine not in ("pandas", "polars"):
        return None
    PowerTransformerApplier.validate_inference_state(state)
    context = "global" if state.get("columns") else "row"
    return ExecutionCapability(engine, "apply", "local", "preserve", context)

validate_inference_state(raw) staticmethod

Inspect the saved lambdas and scaler without normalizing or relearning them.

Source code in skyulf-core/skyulf/preprocessing/transformations/power.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect the saved lambdas and scaler without normalizing or relearning them."""
    if local_state_fields(
        raw,
        "power_transformer",
        {"type", "columns", "method", "standardize", "lambdas", "scaler_params"},
        allow_empty=True,
    ):
        columns = raw["columns"]
        if type(columns) is not list or any(not isinstance(col, str) for col in columns):
            raise ValueError("Fitted power columns must be a list of strings.")
        validate_power_parameters(raw, len(columns))
    return raw

PowerTransformerCalculator

Bases: BaseCalculator

Fit one shared PowerTransformer across the selected columns.

Source code in skyulf-core/skyulf/preprocessing/transformations/power.py 
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
@NodeRegistry.register("PowerTransformer", PowerTransformerApplier)
@node_meta(
    id="PowerTransformer",
    name="Power Transformer",
    category="Preprocessing",
    description="Apply a power transform featurewise to make data more Gaussian-like.",
    params={"method": "yeo-johnson", "standardize": True, "columns": []},
    learns_from_data=True,
)
class PowerTransformerCalculator(BaseCalculator):
    """Fit one shared PowerTransformer across the selected columns."""

    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]) -> PowerTransformerArtifact:  # pylint: disable=arguments-differ
        """Fit per-column lambdas, dropping columns Box-Cox cannot accept."""
        if user_picked_no_columns(config):
            return {}

        method = config.get("method", "yeo-johnson")
        standardize = config.get("standardize", True)
        # TODO(pandas-removal): PowerTransformer.fit accepts numpy directly,
        # but _filter_power_columns() uses Pandas column-wise boolean
        # indexing (`X_pd[c] <= 0`) to drop non-positive columns for box-cox.
        # That check is easy to port to numpy (X_np[:, i] <= 0), but do it
        # together with a matching rewrite of _filter_power_columns to take
        # an ndarray, so both stay in sync.
        X_pd, cols = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
        valid_cols = _filter_power_columns(X_pd, cols, method)
        if not valid_cols:
            return {}

        transformer = PowerTransformer(method=method, standardize=standardize)
        transformer.fit(X_pd[valid_cols])

        return {
            "type": "power_transformer",
            "lambdas": transformer.lambdas_.tolist(),
            "method": method,
            "standardize": standardize,
            "columns": valid_cols,
            "scaler_params": _extract_scaler_params(transformer, standardize),
        }

fit(X, _y, config)

Fit per-column lambdas, dropping columns Box-Cox cannot accept.

Source code in skyulf-core/skyulf/preprocessing/transformations/power.py 
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
@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> PowerTransformerArtifact:  # pylint: disable=arguments-differ
    """Fit per-column lambdas, dropping columns Box-Cox cannot accept."""
    if user_picked_no_columns(config):
        return {}

    method = config.get("method", "yeo-johnson")
    standardize = config.get("standardize", True)
    # TODO(pandas-removal): PowerTransformer.fit accepts numpy directly,
    # but _filter_power_columns() uses Pandas column-wise boolean
    # indexing (`X_pd[c] <= 0`) to drop non-positive columns for box-cox.
    # That check is easy to port to numpy (X_np[:, i] <= 0), but do it
    # together with a matching rewrite of _filter_power_columns to take
    # an ndarray, so both stay in sync.
    X_pd, cols = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
    valid_cols = _filter_power_columns(X_pd, cols, method)
    if not valid_cols:
        return {}

    transformer = PowerTransformer(method=method, standardize=standardize)
    transformer.fit(X_pd[valid_cols])

    return {
        "type": "power_transformer",
        "lambdas": transformer.lambdas_.tolist(),
        "method": method,
        "standardize": standardize,
        "columns": valid_cols,
        "scaler_params": _extract_scaler_params(transformer, standardize),
    }

infer_output_schema(input_schema, config)

Return a schema with transformed columns promoted to float64.

Source code in skyulf-core/skyulf/preprocessing/transformations/power.py 
192
193
194
195
196
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)

SimpleTransformationApplier

Bases: BaseApplier

Apply the configured per-column math transformations (log, sqrt, ...).

Source code in skyulf-core/skyulf/preprocessing/transformations/simple.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
class SimpleTransformationApplier(BaseApplier):
    """Apply the configured per-column math transformations (log, sqrt, ...)."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect ordered native rules without consuming iterators or normalizing state."""
        local_state_fields(raw, "simple_transformation", {"type", "transformations"})
        transformations = raw["transformations"]
        if transformations is None:
            return raw
        if type(transformations) not in (list, tuple):
            raise ValueError("SimpleTransformation rules must be a list or tuple.")
        for item in transformations:
            _validate_simple_rule(item)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe native pointwise math as local row-preserving inference."""
        if engine not in ("pandas", "polars"):
            return None
        SimpleTransformationApplier.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
        """Run each configured transformation on the active engine; ``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]:
        transformations = params.get("transformations", [])
        if not transformations:
            return X, _y

        X_out = X
        for item in transformations:
            col = item.get("column")
            method = item.get("method")
            if col not in X_out.columns:
                continue
            op = _POLARS_OPS.get(method)
            if op is None:
                continue
            X_out = _apply_polars_op(X_out, item, op)
        return X_out, _y

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

        df_out = X.copy()
        for item in transformations:
            col = item.get("column")
            method = item.get("method")
            if col not in df_out.columns:
                continue
            op = _PANDAS_OPS.get(method)
            if op is None:
                continue
            df_out[col] = op(pd.to_numeric(df_out[col], errors="coerce"), item)
        return df_out, _y

apply(X, _y, params)

Run each configured transformation on the active engine; y passes through.

Source code in skyulf-core/skyulf/preprocessing/transformations/simple.py 
58
59
60
61
62
63
@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Run each configured transformation on the active engine; ``y`` passes through."""
    return apply_dual_engine(
        X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe native pointwise math as local row-preserving inference.

Source code in skyulf-core/skyulf/preprocessing/transformations/simple.py 
50
51
52
53
54
55
56
@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe native pointwise math as local row-preserving inference."""
    if engine not in ("pandas", "polars"):
        return None
    SimpleTransformationApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect ordered native rules without consuming iterators or normalizing state.

Source code in skyulf-core/skyulf/preprocessing/transformations/simple.py 
37
38
39
40
41
42
43
44
45
46
47
48
@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect ordered native rules without consuming iterators or normalizing state."""
    local_state_fields(raw, "simple_transformation", {"type", "transformations"})
    transformations = raw["transformations"]
    if transformations is None:
        return raw
    if type(transformations) not in (list, tuple):
        raise ValueError("SimpleTransformation rules must be a list or tuple.")
    for item in transformations:
        _validate_simple_rule(item)
    return raw

SimpleTransformationCalculator

Bases: BaseCalculator

Package the configured transformations into an artifact without reading the data.

Source code in skyulf-core/skyulf/preprocessing/transformations/simple.py 
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
147
148
149
150
151
152
153
@NodeRegistry.register("SimpleTransformation", SimpleTransformationApplier)
@node_meta(
    id="SimpleTransformation",
    name="Simple Transformation",
    category="Preprocessing",
    description="Apply simple mathematical transformations (log, sqrt, etc.).",
    params={"func": "log", "columns": []},
    learns_from_data=False,
)
class SimpleTransformationCalculator(BaseCalculator):
    """Package the configured transformations into an artifact without reading the data."""

    _FLOAT_METHODS = {
        "square",
        "log",
        "sqrt",
        "square_root",
        "cube_root",
        "reciprocal",
        "exp",
        "exponential",
    }

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return updated dtypes for transformed numeric columns."""
        schema = input_schema
        touched: set[str] = set()
        for item in config.get("transformations", []):
            col = item.get("column")
            method = item.get("method")
            if col not in schema.columns or method is None:
                continue
            if method in self._FLOAT_METHODS:
                touched.add(col)

        for col in touched:
            schema = schema.with_dtype(col, "float64")
        return schema

    def fit(
        self,
        df: pd.DataFrame | SkyulfDataFrame | tuple[Any, ...] | Any,
        config: dict[str, Any],
    ) -> SimpleTransformationArtifact:
        """Return the ``transformations`` config as-is; nothing is fitted."""
        # Config: {'transformations': [{'column': 'col1', 'method': 'log'}, ...]}
        return {
            "type": "simple_transformation",
            "transformations": config.get("transformations", []),
        }

fit(df, config)

Return the transformations config as-is; nothing is fitted.

Source code in skyulf-core/skyulf/preprocessing/transformations/simple.py 
143
144
145
146
147
148
149
150
151
152
153
def fit(
    self,
    df: pd.DataFrame | SkyulfDataFrame | tuple[Any, ...] | Any,
    config: dict[str, Any],
) -> SimpleTransformationArtifact:
    """Return the ``transformations`` config as-is; nothing is fitted."""
    # Config: {'transformations': [{'column': 'col1', 'method': 'log'}, ...]}
    return {
        "type": "simple_transformation",
        "transformations": config.get("transformations", []),
    }

infer_output_schema(input_schema, config)

Return updated dtypes for transformed numeric columns.

Source code in skyulf-core/skyulf/preprocessing/transformations/simple.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Return updated dtypes for transformed numeric columns."""
    schema = input_schema
    touched: set[str] = set()
    for item in config.get("transformations", []):
        col = item.get("column")
        method = item.get("method")
        if col not in schema.columns or method is None:
            continue
        if method in self._FLOAT_METHODS:
            touched.add(col)

    for col in touched:
        schema = schema.with_dtype(col, "float64")
    return schema