API: preprocessing.encoding

skyulf.preprocessing.encoding

Encoding nodes package.

Split from a single 832-LOC module into per-encoder files

_common.py — shared helpers (detect_categorical_columns, _exclude_target_column, …) one_hot.py — OneHotEncoder ordinal.py — OrdinalEncoder label.py — LabelEncoder target.py — TargetEncoder hash.py — HashEncoder dummy.py — DummyEncoder woe.py — WOEEncoder (Weight-of-Evidence / Information Value)

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

DummyEncoderApplier

Bases: BaseApplier

Replace each categorical column with one <col>_<category> indicator column.

The originals are always dropped. Parity depends on both engines rendering a value to the same string before comparing it with the learned categories — one rule, implemented per engine by _pandas_col_to_str and _polars_col_to_str_expr, which is a function of the value alone so a category cannot stop matching because of what else shares its batch. A value unseen at fit time yields an all-zero row rather than raising. Generated names that duplicate another indicator or a retained input column raise ValueError before output construction.

Version 1 artifacts render booleans and datetimes consistently across engines, including nanoseconds and timezone-aware instants. Midnight naive datetimes share date keys. Literal strings are unchanged. Unversioned artifacts retain their original rendering; refit the encoder and downstream model to obtain the portable category contract.

Source code in skyulf-core/skyulf/preprocessing/encoding/dummy.py

class DummyEncoderApplier(BaseApplier):
    """Replace each categorical column with one ``<col>_<category>`` indicator column.

    The originals are always dropped. Parity depends on both engines rendering
    a value to the *same* string before comparing it with the learned
    categories — one rule, implemented per engine by ``_pandas_col_to_str``
    and ``_polars_col_to_str_expr``, which is a function of the value alone so
    a category cannot stop matching because of what else shares its batch. A
    value unseen at fit time yields an all-zero row rather than raising.
    Generated names that duplicate another indicator or a retained input
    column raise ``ValueError`` before output construction.

    Version 1 artifacts render booleans and datetimes consistently across engines,
    including nanoseconds and timezone-aware instants. Midnight naive datetimes
    share date keys. Literal strings are unchanged. Unversioned artifacts retain
    their original rendering; refit the encoder and downstream model to obtain
    the portable category contract.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved categories without deriving a vocabulary from inference rows."""
        fields = {"type", "columns", "categories", "drop_first"}
        if isinstance(raw, dict) and "category_key_version" in raw:
            fields.add("category_key_version")
        if not local_state_fields(raw, "dummy_encoder", fields, allow_empty=True):
            return raw
        local_boolean(raw["drop_first"], "drop_first")
        uses_category_keys(raw)
        _validate_dummy_categories(raw)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Keep legacy pandas batch rendering visible in the local context declaration."""
        if engine not in ("pandas", "polars"):
            return None
        DummyEncoderApplier.validate_inference_state(state)
        legacy = state.get("columns") and not uses_category_keys(state)
        context = "global" if engine == "pandas" and legacy 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
        """Dispatch to the engine-specific dummy encoding, forwarding ``(X, y)`` when present."""
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"polars": _dummy_apply_polars, "pandas": _dummy_apply_pandas},
        )

apply(X, y, params)

Dispatch to the engine-specific dummy encoding, forwarding (X, y) when present.

Source code in skyulf-core/skyulf/preprocessing/encoding/dummy.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch to the engine-specific dummy encoding, forwarding ``(X, y)`` when present."""
    return apply_dual_engine(
        (X, y) if y is not None else X,
        params,
        {"polars": _dummy_apply_polars, "pandas": _dummy_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Keep legacy pandas batch rendering visible in the local context declaration.

Source code in skyulf-core/skyulf/preprocessing/encoding/dummy.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Keep legacy pandas batch rendering visible in the local context declaration."""
    if engine not in ("pandas", "polars"):
        return None
    DummyEncoderApplier.validate_inference_state(state)
    legacy = state.get("columns") and not uses_category_keys(state)
    context = "global" if engine == "pandas" and legacy else "row"
    return ExecutionCapability(engine, "apply", "local", "preserve", context)

validate_inference_state(raw) staticmethod

Inspect saved categories without deriving a vocabulary from inference rows.

Source code in skyulf-core/skyulf/preprocessing/encoding/dummy.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved categories without deriving a vocabulary from inference rows."""
    fields = {"type", "columns", "categories", "drop_first"}
    if isinstance(raw, dict) and "category_key_version" in raw:
        fields.add("category_key_version")
    if not local_state_fields(raw, "dummy_encoder", fields, allow_empty=True):
        return raw
    local_boolean(raw["drop_first"], "drop_first")
    uses_category_keys(raw)
    _validate_dummy_categories(raw)
    return raw

DummyEncoderCalculator

Bases: BaseCalculator

Learn the sorted category list for each resolved categorical column.

The target column is excluded first: dummy encoding replaces the column it encodes with several derived ones, which would break a downstream Feature/Target Split. Nulls and float NaN are dropped before the list is built, so they never become a category. An explicit columns: [] yields an empty artifact so the applier no-ops.

Source code in skyulf-core/skyulf/preprocessing/encoding/dummy.py

@NodeRegistry.register("DummyEncoder", DummyEncoderApplier)
@node_meta(
    id="DummyEncoder",
    name="Dummy Encoder",
    category="Preprocessing",
    description="Convert categorical variables into dummy/indicator variables (pandas.get_dummies).",
    params={"columns": [], "drop_first": False},
    learns_from_data=True,
)
class DummyEncoderCalculator(BaseCalculator):
    """Learn the sorted category list for each resolved categorical column.

    The target column is excluded first: dummy encoding replaces the column it
    encodes with several derived ones, which would break a downstream
    Feature/Target Split. Nulls and float NaN are dropped before the list is
    built, so they never become a category. An explicit ``columns: []`` yields
    an empty artifact so the applier no-ops.
    """

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> DummyEncoderArtifact:  # pylint: disable=arguments-differ
        """Short-circuit an explicit empty column selection, else learn the categories."""
        if user_picked_no_columns(config):
            return {}
        return cast(
            DummyEncoderArtifact,
            fit_dual_engine(
                (X, y) if y is not None else X,
                config,
                {"polars": _dummy_fit_polars, "pandas": _dummy_fit_pandas},
            ),
        )

fit(X, y, config)

Short-circuit an explicit empty column selection, else learn the categories.

Source code in skyulf-core/skyulf/preprocessing/encoding/dummy.py

@fit_method
def fit(self, X: Any, y: Any, config: dict[str, Any]) -> DummyEncoderArtifact:  # pylint: disable=arguments-differ
    """Short-circuit an explicit empty column selection, else learn the categories."""
    if user_picked_no_columns(config):
        return {}
    return cast(
        DummyEncoderArtifact,
        fit_dual_engine(
            (X, y) if y is not None else X,
            config,
            {"polars": _dummy_fit_polars, "pandas": _dummy_fit_pandas},
        ),
    )

HashEncoderApplier

Bases: BaseApplier

Replace each categorical column's values in place with a bucket index.

Buckets run over [0, n_features). No vocabulary is learned, so the same artifact handles values never seen at fit time — at the cost of collisions, which the hashing trick accepts by design. Both engines must agree on the bucket, which is why they share _stable_hash (blake2b) instead of using polars' native hash(): deployment always crosses engines, so a divergence would corrupt every production encoding. Version 2 normalizes missing and numeric scalars per value, keeping literal strings distinct. Older normalization versions retain their original bucket assignments.

Source code in skyulf-core/skyulf/preprocessing/encoding/hash.py

class HashEncoderApplier(BaseApplier):
    """Replace each categorical column's values in place with a bucket index.

    Buckets run over ``[0, n_features)``. No vocabulary is learned, so the same
    artifact handles values never seen at fit time — at the cost of collisions,
    which the hashing trick accepts by design. Both engines must agree on the
    bucket, which is why they share ``_stable_hash`` (blake2b) instead of using
    polars' native ``hash()``: deployment always crosses engines, so a
    divergence would corrupt every production encoding. Version 2 normalizes
    missing and numeric scalars per value, keeping literal strings distinct.
    Older normalization versions retain their original bucket assignments.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved hashing choices without computing buckets or coercing scalars."""
        fields = {"type", "columns", "n_features"}
        if isinstance(raw, dict) and "numeric_normalization_version" in raw:
            fields.add("numeric_normalization_version")
        if not local_state_fields(raw, "hash_encoder", fields, allow_empty=True):
            return raw
        columns = raw["columns"]
        if type(columns) not in (list, tuple) or any(not isinstance(col, str) for col in columns):
            raise ValueError("Fitted hash columns must be a list or tuple of strings.")
        count = raw["n_features"]
        if not isinstance(count, (Real, Decimal, np.bool_)) or not count > 0:
            raise ValueError("Fitted hash bucket count must be positive and numeric.")
        _uses_numeric_normalization(raw)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe fixed hashing while preserving legacy pandas rendering dependencies."""
        if engine not in ("pandas", "polars"):
            return None
        HashEncoderApplier.validate_inference_state(state)
        legacy = state.get("columns") and state.get("numeric_normalization_version") != 2
        context = "global" if engine == "pandas" and legacy 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
        """Dispatch to the engine-specific hashing, forwarding ``(X, y)`` only when ``y`` exists."""
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"polars": _hash_apply_polars, "pandas": _hash_apply_pandas},
        )

apply(X, y, params)

Dispatch to the engine-specific hashing, forwarding (X, y) only when y exists.

Source code in skyulf-core/skyulf/preprocessing/encoding/hash.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch to the engine-specific hashing, forwarding ``(X, y)`` only when ``y`` exists."""
    return apply_dual_engine(
        (X, y) if y is not None else X,
        params,
        {"polars": _hash_apply_polars, "pandas": _hash_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Describe fixed hashing while preserving legacy pandas rendering dependencies.

Source code in skyulf-core/skyulf/preprocessing/encoding/hash.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe fixed hashing while preserving legacy pandas rendering dependencies."""
    if engine not in ("pandas", "polars"):
        return None
    HashEncoderApplier.validate_inference_state(state)
    legacy = state.get("columns") and state.get("numeric_normalization_version") != 2
    context = "global" if engine == "pandas" and legacy else "row"
    return ExecutionCapability(engine, "apply", "local", "preserve", context)

validate_inference_state(raw) staticmethod

Inspect saved hashing choices without computing buckets or coercing scalars.

Source code in skyulf-core/skyulf/preprocessing/encoding/hash.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved hashing choices without computing buckets or coercing scalars."""
    fields = {"type", "columns", "n_features"}
    if isinstance(raw, dict) and "numeric_normalization_version" in raw:
        fields.add("numeric_normalization_version")
    if not local_state_fields(raw, "hash_encoder", fields, allow_empty=True):
        return raw
    columns = raw["columns"]
    if type(columns) not in (list, tuple) or any(not isinstance(col, str) for col in columns):
        raise ValueError("Fitted hash columns must be a list or tuple of strings.")
    count = raw["n_features"]
    if not isinstance(count, (Real, Decimal, np.bool_)) or not count > 0:
        raise ValueError("Fitted hash bucket count must be positive and numeric.")
    _uses_numeric_normalization(raw)
    return raw

HashEncoderCalculator

Bases: BaseCalculator

Resolve which columns to hash and how many buckets to use.

No vocabulary is learned — hashing is stateless, so the artifact is just the column set plus n_features. The target column is excluded because hash encoding overwrites the column it encodes. An explicit columns: [], or auto-detection finding nothing, yields an empty artifact so the applier no-ops.

Source code in skyulf-core/skyulf/preprocessing/encoding/hash.py

@NodeRegistry.register("HashEncoder", HashEncoderApplier)
@node_meta(
    id="HashEncoder",
    name="Hash Encoder",
    category="Preprocessing",
    description="Encode categorical features using hashing.",
    params={"n_features": 8, "columns": []},
    learns_from_data=True,
)
class HashEncoderCalculator(BaseCalculator):
    """Resolve which columns to hash and how many buckets to use.

    No vocabulary is learned — hashing is stateless, so the artifact is just
    the column set plus ``n_features``. The target column is excluded because
    hash encoding overwrites the column it encodes. An explicit
    ``columns: []``, or auto-detection finding nothing, yields an empty
    artifact so the applier no-ops.
    """

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> HashEncoderArtifact:  # pylint: disable=arguments-differ
        """Build the artifact, auto-detecting categorical columns when none are named."""
        if user_picked_no_columns(config):
            return {}

        cols = resolve_columns(X, config, detect_categorical_columns)
        cols = _exclude_target_column(cols, config, "HashEncoder", y)
        if not cols:
            return {}

        return {
            "type": "hash_encoder",
            "columns": cols,
            "n_features": config.get("n_features", 10),
            "numeric_normalization_version": 2,
        }

    def infer_output_schema(
        self,
        input_schema: SkyulfSchema,
        config: dict[str, Any],
    ) -> SkyulfSchema | None:
        """Pass the input schema through: bucket indices replace values in place."""
        # Hash encoder replaces values in source columns in place
        # (`pl.col(col)...alias(col)`). Schema is unchanged.
        return input_schema

fit(X, y, config)

Build the artifact, auto-detecting categorical columns when none are named.

Source code in skyulf-core/skyulf/preprocessing/encoding/hash.py

@fit_method
def fit(self, X: Any, y: Any, config: dict[str, Any]) -> HashEncoderArtifact:  # pylint: disable=arguments-differ
    """Build the artifact, auto-detecting categorical columns when none are named."""
    if user_picked_no_columns(config):
        return {}

    cols = resolve_columns(X, config, detect_categorical_columns)
    cols = _exclude_target_column(cols, config, "HashEncoder", y)
    if not cols:
        return {}

    return {
        "type": "hash_encoder",
        "columns": cols,
        "n_features": config.get("n_features", 10),
        "numeric_normalization_version": 2,
    }

infer_output_schema(input_schema, config)

Pass the input schema through: bucket indices replace values in place.

Source code in skyulf-core/skyulf/preprocessing/encoding/hash.py

def infer_output_schema(
    self,
    input_schema: SkyulfSchema,
    config: dict[str, Any],
) -> SkyulfSchema | None:
    """Pass the input schema through: bucket indices replace values in place."""
    # Hash encoder replaces values in source columns in place
    # (`pl.col(col)...alias(col)`). Schema is unchanged.
    return input_schema

LabelEncoderApplier

Bases: BaseApplier

Replace categorical values in place with the integer ids their fitted encoder assigned.

New feature artifacts use typed scalar keys: integral floats match their integer category without matching literal numeric strings. Legacy artifacts retain their original string lookup. Values unseen at fit time map to missing_code. Target encoders retain their existing string-label behavior. Encoded targets are native Series for the feature engine; pandas Series keep their index and all native Series keep their name. Unencoded targets pass through unchanged, including lists and NumPy arrays.

Source code in skyulf-core/skyulf/preprocessing/encoding/label.py

class LabelEncoderApplier(BaseApplier):
    """Replace categorical values in place with the integer ids their fitted encoder assigned.

    New feature artifacts use typed scalar keys: integral floats match their
    integer category without matching literal numeric strings. Legacy artifacts
    retain their original string lookup. Values unseen at fit time map to
    ``missing_code``. Target encoders retain their existing string-label behavior.
    Encoded targets are native Series for the feature engine; pandas Series keep
    their index and all native Series keep their name. Unencoded targets pass
    through unchanged, including lists and NumPy arrays.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect native learned classes without normalizing the fitted artifact."""
        fields = {"type", "encoders", "columns", "classes_count", "missing_code"}
        if isinstance(raw, dict):
            fields.update(set(raw) & {"category_key_version", "target_column"})
        local_state_fields(raw, "label_encoder", fields)
        uses_category_keys(raw)
        if not isinstance(raw["missing_code"], Decimal):
            local_scalar(raw["missing_code"], "LabelEncoder missing code")
        _validate_label_encoders(raw)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Declare fixed lookups while retaining legacy feature rendering dependencies."""
        if engine not in ("pandas", "polars"):
            return None
        LabelEncoderApplier.validate_inference_state(state)
        legacy = state["columns"] and not uses_category_keys(state)
        context = "global" if engine == "pandas" and legacy 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
        """Dispatch to the engine-specific encode, forwarding ``(X, y)`` only when ``y`` exists."""
        return apply_target_encoder(
            X,
            y,
            params,
            {"polars": _label_apply_polars, "pandas": _label_apply_pandas},
        )

apply(X, y, params)

Dispatch to the engine-specific encode, forwarding (X, y) only when y exists.

Source code in skyulf-core/skyulf/preprocessing/encoding/label.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch to the engine-specific encode, forwarding ``(X, y)`` only when ``y`` exists."""
    return apply_target_encoder(
        X,
        y,
        params,
        {"polars": _label_apply_polars, "pandas": _label_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Declare fixed lookups while retaining legacy feature rendering dependencies.

Source code in skyulf-core/skyulf/preprocessing/encoding/label.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Declare fixed lookups while retaining legacy feature rendering dependencies."""
    if engine not in ("pandas", "polars"):
        return None
    LabelEncoderApplier.validate_inference_state(state)
    legacy = state["columns"] and not uses_category_keys(state)
    context = "global" if engine == "pandas" and legacy else "row"
    return ExecutionCapability(engine, "apply", "local", "preserve", context)

validate_inference_state(raw) staticmethod

Inspect native learned classes without normalizing the fitted artifact.

Source code in skyulf-core/skyulf/preprocessing/encoding/label.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect native learned classes without normalizing the fitted artifact."""
    fields = {"type", "encoders", "columns", "classes_count", "missing_code"}
    if isinstance(raw, dict):
        fields.update(set(raw) & {"category_key_version", "target_column"})
    local_state_fields(raw, "label_encoder", fields)
    uses_category_keys(raw)
    if not isinstance(raw["missing_code"], Decimal):
        local_scalar(raw["missing_code"], "LabelEncoder missing code")
    _validate_label_encoders(raw)
    return raw

LabelEncoderCalculator

Bases: BaseCalculator

Fit one LabelEncoder per named column, plus one for the target when it qualifies.

Unlike its sibling encoders this node deliberately does not auto-detect feature columns: an empty columns means "encode the target only", mirroring sklearn's convention that LabelEncoder is a 1-D label transformer and OrdinalEncoder is the tool for feature columns. Fitting zero encoders logs a warning rather than passing a silent no-op downstream.

Source code in skyulf-core/skyulf/preprocessing/encoding/label.py

@NodeRegistry.register("LabelEncoder", LabelEncoderApplier)
@node_meta(
    id="LabelEncoder",
    name="Label Encoder",
    category="Preprocessing",
    description="Encode target labels with value between 0 and n_classes-1.",
    params={"columns": [], "missing_code": -1},
    learns_from_data=True,
)
class LabelEncoderCalculator(BaseCalculator):
    """Fit one LabelEncoder per named column, plus one for the target when it qualifies.

    Unlike its sibling encoders this node deliberately does *not* auto-detect
    feature columns: an empty ``columns`` means "encode the target only",
    mirroring sklearn's convention that ``LabelEncoder`` is a 1-D label
    transformer and ``OrdinalEncoder`` is the tool for feature columns. Fitting
    zero encoders logs a warning rather than passing a silent no-op downstream.
    """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Pass the input schema through: encoded columns keep their name and position."""
        # Label encoding replaces categorical values with ints in place;
        # column set is preserved.
        return input_schema

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> LabelEncoderArtifact:  # pylint: disable=arguments-differ
        """Fit the encoders on the frame's own engine, forwarding ``y`` when present."""
        return cast(
            LabelEncoderArtifact,
            fit_target_encoder(
                X,
                y,
                config,
                {"polars": _label_fit_polars, "pandas": _label_fit_pandas},
            ),
        )

fit(X, y, config)

Fit the encoders on the frame's own engine, forwarding y when present.

Source code in skyulf-core/skyulf/preprocessing/encoding/label.py

@fit_method
def fit(self, X: Any, y: Any, config: dict[str, Any]) -> LabelEncoderArtifact:  # pylint: disable=arguments-differ
    """Fit the encoders on the frame's own engine, forwarding ``y`` when present."""
    return cast(
        LabelEncoderArtifact,
        fit_target_encoder(
            X,
            y,
            config,
            {"polars": _label_fit_polars, "pandas": _label_fit_pandas},
        ),
    )

infer_output_schema(input_schema, config)

Pass the input schema through: encoded columns keep their name and position.

Source code in skyulf-core/skyulf/preprocessing/encoding/label.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Pass the input schema through: encoded columns keep their name and position."""
    # Label encoding replaces categorical values with ints in place;
    # column set is preserved.
    return input_schema

OneHotEncoderApplier

Bases: BaseApplier

Expand each categorical column into its one-hot indicator columns.

Parity here comes from both engines transforming through the same fitted sklearn encoder carried in the artifact, not from two parallel implementations. Nulls are filled with the sentinel token before transforming, which only lines up when include_missing was set at fit time too — the encoder knows that level solely if it saw it. Indicator columns are concatenated on and the originals dropped unless drop_original is false. Generated names must be unique and must not collide with retained input columns; conflicts raise ValueError. New artifacts escape literal missing tokens and escape-prefixed strings; unversioned artifacts retain their original missing-value transformation. An empty artifact declares identity replay only for an explicit empty column selection with a validated recipe.

Source code in skyulf-core/skyulf/preprocessing/encoding/one_hot.py

class OneHotEncoderApplier(BaseApplier):
    """Expand each categorical column into its one-hot indicator columns.

    Parity here comes from both engines transforming through the *same* fitted
    sklearn encoder carried in the artifact, not from two parallel
    implementations. Nulls are filled with the sentinel token before
    transforming, which only lines up when ``include_missing`` was set at fit
    time too — the encoder knows that level solely if it saw it. Indicator
    columns are concatenated on and the originals dropped unless
    ``drop_original`` is false. Generated names must be unique and must not
    collide with retained input columns; conflicts raise ``ValueError``.
    New artifacts escape literal missing tokens and escape-prefixed strings;
    unversioned artifacts retain their original missing-value transformation.
    An empty artifact declares identity replay only for an explicit empty
    column selection with a validated recipe.
    """

    @staticmethod
    def validate_fitted_state(raw: dict) -> dict:
        """Inspect this node's supported saved state without fitting or applying data."""
        return _onehot_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 encoder artifact requires explicit empty columns.")
        return _onehot_config(fitted_columns(raw, state or {"columns": []}), state)

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch to the engine-specific transform, forwarding ``(X, y)`` only when present."""
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"polars": _onehot_apply_polars, "pandas": _onehot_apply_pandas},
        )

apply(X, y, params)

Dispatch to the engine-specific transform, forwarding (X, y) only when present.

Source code in skyulf-core/skyulf/preprocessing/encoding/one_hot.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch to the engine-specific transform, forwarding ``(X, y)`` only when present."""
    return apply_dual_engine(
        (X, y) if y is not None else X,
        params,
        {"polars": _onehot_apply_polars, "pandas": _onehot_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/encoding/one_hot.py

@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 encoder artifact requires explicit empty columns.")
    return _onehot_config(fitted_columns(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/encoding/one_hot.py

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

OneHotEncoderCalculator

Bases: BaseCalculator

Fit a sklearn OneHotEncoder on the resolved categorical columns.

The target column is excluded before fitting: one-hot replaces the column it encodes with several derived ones, which would break a downstream Feature/Target Split. An explicit columns: [], or auto-detection finding no categorical columns, yields an empty artifact so the applier no-ops. Degenerate columns (zero categories, or one category with drop_first) are warned about rather than silently producing nothing.

Source code in skyulf-core/skyulf/preprocessing/encoding/one_hot.py

@NodeRegistry.register(
    "OneHotEncoder",
    OneHotEncoderApplier,
    execution_capabilities=(
        ExecutionCapability(
            "pandas",
            "apply",
            "python_batch",
            "preserve",
            "row",
            config_match=(("max_categories", None), ("include_missing", False)),
        ),
        ExecutionCapability("pandas", "apply", "local", "preserve", "row"),
        ExecutionCapability("polars", "apply", "local", "preserve", "row"),
    ),
)
@node_meta(
    id="OneHotEncoder",
    name="One-Hot Encoder",
    category="Preprocessing",
    description="Encodes categorical features as a one-hot numeric array.",
    params={
        "handle_unknown": "ignore",
        "drop_first": False,
        "max_categories": 20,
        "columns": [],
        "include_missing": False,
    },
    learns_from_data=True,
)
class OneHotEncoderCalculator(BaseCalculator):
    """Fit a sklearn ``OneHotEncoder`` on the resolved categorical columns.

    The target column is excluded before fitting: one-hot replaces the column
    it encodes with several derived ones, which would break a downstream
    Feature/Target Split. An explicit ``columns: []``, or auto-detection
    finding no categorical columns, yields an empty artifact so the applier
    no-ops. Degenerate columns (zero categories, or one category with
    ``drop_first``) are warned about rather than silently producing nothing.
    """

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> OneHotArtifact:  # pylint: disable=arguments-differ
        """Short-circuit an explicit empty column selection, else fit on the frame's own engine."""
        if user_picked_no_columns(config):
            return {}
        return cast(
            OneHotArtifact,
            fit_dual_engine(
                (X, y) if y is not None else X,
                config,
                {"polars": _onehot_fit_polars, "pandas": _onehot_fit_pandas},
            ),
        )

fit(X, y, config)

Short-circuit an explicit empty column selection, else fit on the frame's own engine.

Source code in skyulf-core/skyulf/preprocessing/encoding/one_hot.py

@fit_method
def fit(self, X: Any, y: Any, config: dict[str, Any]) -> OneHotArtifact:  # pylint: disable=arguments-differ
    """Short-circuit an explicit empty column selection, else fit on the frame's own engine."""
    if user_picked_no_columns(config):
        return {}
    return cast(
        OneHotArtifact,
        fit_dual_engine(
            (X, y) if y is not None else X,
            config,
            {"polars": _onehot_fit_polars, "pandas": _onehot_fit_pandas},
        ),
    )

OrdinalEncoderApplier

Bases: BaseApplier

Replace categorical values in place with fitted ordinal indices, target included.

Features use the shared encoder_object and versioned scalar keys; y uses a separate __target__ encoder with its existing string-label rules. Older feature artifacts retain their original string lookup. An unseen value becomes unknown_value unless handle_unknown is "error". Encoded output is float32, not integer.

Source code in skyulf-core/skyulf/preprocessing/encoding/ordinal.py

class OrdinalEncoderApplier(BaseApplier):
    """Replace categorical values in place with fitted ordinal indices, target included.

    Features use the shared ``encoder_object`` and versioned scalar keys; ``y``
    uses a separate ``__target__`` encoder with its existing string-label rules.
    Older feature artifacts retain their original string lookup. An unseen
    value becomes ``unknown_value`` unless ``handle_unknown`` is ``"error"``.
    Encoded output is ``float32``, not integer.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect feature and optional target encoders without changing lookup provenance."""
        fields = {"type", "columns", "encoder_object", "encoders", "categories_count"}
        if isinstance(raw, dict):
            fields.update({"category_key_version", "target_column"}.intersection(raw))
        if not local_state_fields(raw, "ordinal", fields, allow_empty=True):
            return raw
        uses_category_keys(raw)
        _validate_ordinal_features(raw)
        OrdinalEncoderApplier._validate_targets(raw)
        return raw

    @staticmethod
    def _validate_targets(raw: dict) -> None:
        """Keep embedded-target routing separate from selected feature encoders."""
        encoders = raw["encoders"]
        if type(encoders) is not dict or set(encoders) - {"__target__"}:
            raise ValueError("Unexpected fitted ordinal target encoders.")
        if "__target__" in encoders:
            _validate_ordinal_encoder(encoders["__target__"], 1)
        if "target_column" in raw and (
            not isinstance(raw["target_column"], str)
            or raw["target_column"] in raw["columns"]
            or "__target__" not in encoders
        ):
            raise ValueError("Fitted ordinal target routing disagrees with its encoders.")

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe saved ordinal lookups without enabling worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        OrdinalEncoderApplier.validate_inference_state(state)
        # Legacy pandas datetime string formatting can depend on neighboring rows.
        legacy_pandas = (
            engine == "pandas" and state.get("columns") and not uses_category_keys(state)
        )
        context = "global" if legacy_pandas 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
        """Dispatch to the engine-specific encode, forwarding ``(X, y)`` only when ``y`` exists."""
        return apply_target_encoder(
            X,
            y,
            params,
            {"polars": _ordinal_apply_polars, "pandas": _ordinal_apply_pandas},
        )

apply(X, y, params)

Dispatch to the engine-specific encode, forwarding (X, y) only when y exists.

Source code in skyulf-core/skyulf/preprocessing/encoding/ordinal.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch to the engine-specific encode, forwarding ``(X, y)`` only when ``y`` exists."""
    return apply_target_encoder(
        X,
        y,
        params,
        {"polars": _ordinal_apply_polars, "pandas": _ordinal_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Describe saved ordinal lookups without enabling worker execution.

Source code in skyulf-core/skyulf/preprocessing/encoding/ordinal.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe saved ordinal lookups without enabling worker execution."""
    if engine not in ("pandas", "polars"):
        return None
    OrdinalEncoderApplier.validate_inference_state(state)
    # Legacy pandas datetime string formatting can depend on neighboring rows.
    legacy_pandas = (
        engine == "pandas" and state.get("columns") and not uses_category_keys(state)
    )
    context = "global" if legacy_pandas else "row"
    return ExecutionCapability(engine, "apply", "local", "preserve", context)

validate_inference_state(raw) staticmethod

Inspect feature and optional target encoders without changing lookup provenance.

Source code in skyulf-core/skyulf/preprocessing/encoding/ordinal.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect feature and optional target encoders without changing lookup provenance."""
    fields = {"type", "columns", "encoder_object", "encoders", "categories_count"}
    if isinstance(raw, dict):
        fields.update({"category_key_version", "target_column"}.intersection(raw))
    if not local_state_fields(raw, "ordinal", fields, allow_empty=True):
        return raw
    uses_category_keys(raw)
    _validate_ordinal_features(raw)
    OrdinalEncoderApplier._validate_targets(raw)
    return raw

OrdinalEncoderCalculator

Bases: BaseCalculator

Fit an ordinal encoder for the features and, when split out, for the target.

Picking no feature columns is not a no-op here — unlike its sibling encoders, this node still fits a target encoder when y is available. The target is encoded only when y exists and the configured target name is listed in columns but absent from X, i.e. it has already been split away. categories_order supplies an explicit ordering, and the target reads the last row of an n_features + 1 row table.

Source code in skyulf-core/skyulf/preprocessing/encoding/ordinal.py

@NodeRegistry.register("OrdinalEncoder", OrdinalEncoderApplier)
@node_meta(
    id="OrdinalEncoder",
    name="Ordinal Encoder",
    category="Preprocessing",
    description="Encodes categorical features as an integer array.",
    params={
        "columns": [],
        "handle_unknown": "use_encoded_value",
        "unknown_value": -1,
        "categories_order": "",
    },
    learns_from_data=True,
)
class OrdinalEncoderCalculator(BaseCalculator):
    """Fit an ordinal encoder for the features and, when split out, for the target.

    Picking no feature columns is *not* a no-op here — unlike its sibling
    encoders, this node still fits a target encoder when ``y`` is available.
    The target is encoded only when ``y`` exists and the configured target name
    is listed in ``columns`` but absent from ``X``, i.e. it has already been
    split away. ``categories_order`` supplies an explicit ordering, and the
    target reads the last row of an ``n_features + 1`` row table.
    """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Pass the input schema through: encoded columns keep their name and position."""
        # Ordinal encoding replaces categorical values with ints in place;
        # column set is preserved.
        return input_schema

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> OrdinalArtifact:  # pylint: disable=arguments-differ
        """Fit the encoders on the frame's own engine, forwarding ``y`` when present."""
        return cast(
            OrdinalArtifact,
            fit_target_encoder(
                X,
                y,
                config,
                {"polars": _ordinal_fit_polars, "pandas": _ordinal_fit_pandas},
            ),
        )

fit(X, y, config)

Fit the encoders on the frame's own engine, forwarding y when present.

Source code in skyulf-core/skyulf/preprocessing/encoding/ordinal.py

@fit_method
def fit(self, X: Any, y: Any, config: dict[str, Any]) -> OrdinalArtifact:  # pylint: disable=arguments-differ
    """Fit the encoders on the frame's own engine, forwarding ``y`` when present."""
    return cast(
        OrdinalArtifact,
        fit_target_encoder(
            X,
            y,
            config,
            {"polars": _ordinal_fit_polars, "pandas": _ordinal_fit_pandas},
        ),
    )

infer_output_schema(input_schema, config)

Pass the input schema through: encoded columns keep their name and position.

Source code in skyulf-core/skyulf/preprocessing/encoding/ordinal.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Pass the input schema through: encoded columns keep their name and position."""
    # Ordinal encoding replaces categorical values with ints in place;
    # column set is preserved.
    return input_schema

TargetEncoderApplier

Bases: BaseApplier

Replace categorical columns with the target statistics the fitted encoder learned.

Both engines transform through the same sklearn TargetEncoder carried in the artifact, so parity comes from sharing it. The output shape follows the target type: binary and regression targets are encoded in place, while a multiclass target drops the originals and emits one {col}_cls{i} column per class. Multiclass output names must not collide with retained columns; conflicts raise ValueError before any input values are replaced.

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

class TargetEncoderApplier(BaseApplier):
    """Replace categorical columns with the target statistics the fitted encoder learned.

    Both engines transform through the same sklearn ``TargetEncoder`` carried in
    the artifact, so parity comes from sharing it. The output shape follows the
    target type: binary and regression targets are encoded in place, while a
    multiclass target drops the originals and emits one ``{col}_cls{i}`` column
    per class. Multiclass output names must not collide with retained columns;
    conflicts raise ``ValueError`` before any input values are replaced.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the saved full-data encoder while leaving cross-fitted training values alone."""
        if not local_state_fields(
            raw, "target_encoder", {"type", "columns", "encoder_object"}, allow_empty=True
        ):
            return raw
        columns = raw["columns"]
        if not isinstance(columns, (list, tuple)) or any(not isinstance(c, str) for c in columns):
            raise ValueError("Fitted target encoder columns must be ordered strings.")
        encoder = raw["encoder_object"]
        required = {
            "n_features_in_",
            "categories_",
            "target_type_",
            "classes_",
            "target_mean_",
            "encodings_",
        }
        if type(encoder) is not TargetEncoder or not required.issubset(vars(encoder)):
            raise ValueError("Expected a fitted TargetEncoder.")
        if any(callable(value) for value in vars(encoder).values()):
            raise ValueError("Overridden target estimator methods are unsupported.")
        _validate_target_categories(encoder, len(columns))
        _validate_target_statistics(encoder)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe saved target-statistic lookups without refitting or admitting workers."""
        if engine not in ("pandas", "polars"):
            return None
        TargetEncoderApplier.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
        """Dispatch to the engine-specific transform, forwarding ``(X, y)`` only when present."""
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"polars": _target_apply_polars, "pandas": _target_apply_pandas},
        )

apply(X, y, params)

Dispatch to the engine-specific transform, forwarding (X, y) only when present.

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch to the engine-specific transform, forwarding ``(X, y)`` only when present."""
    return apply_dual_engine(
        (X, y) if y is not None else X,
        params,
        {"polars": _target_apply_polars, "pandas": _target_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Describe saved target-statistic lookups without refitting or admitting workers.

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe saved target-statistic lookups without refitting or admitting workers."""
    if engine not in ("pandas", "polars"):
        return None
    TargetEncoderApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect the saved full-data encoder while leaving cross-fitted training values alone.

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect the saved full-data encoder while leaving cross-fitted training values alone."""
    if not local_state_fields(
        raw, "target_encoder", {"type", "columns", "encoder_object"}, allow_empty=True
    ):
        return raw
    columns = raw["columns"]
    if not isinstance(columns, (list, tuple)) or any(not isinstance(c, str) for c in columns):
        raise ValueError("Fitted target encoder columns must be ordered strings.")
    encoder = raw["encoder_object"]
    required = {
        "n_features_in_",
        "categories_",
        "target_type_",
        "classes_",
        "target_mean_",
        "encodings_",
    }
    if type(encoder) is not TargetEncoder or not required.issubset(vars(encoder)):
        raise ValueError("Expected a fitted TargetEncoder.")
    if any(callable(value) for value in vars(encoder).values()):
        raise ValueError("Overridden target estimator methods are unsupported.")
    _validate_target_categories(encoder, len(columns))
    _validate_target_statistics(encoder)
    return raw

TargetEncoderCalculator

Bases: BaseCalculator

Fit a sklearn TargetEncoder against y, cross-fitting the training rows.

A target is mandatory: with no y the node logs a warning and returns an empty artifact rather than failing. The target column is excluded from the encoded set, since encoding it against itself is degenerate. This calculator also satisfies TrainTransformCalculatorProtocol, so the training split receives out-of-fold encodings — a row's own target never leaks into its feature — with the fold count capped by the smallest class count, while held-out splits go through the plain applier.

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

@NodeRegistry.register("TargetEncoder", TargetEncoderApplier)
@node_meta(
    id="TargetEncoder",
    name="Target Encoder",
    category="Preprocessing",
    description="Encode categorical features using target statistics.",
    params={"smooth": "auto", "target_type": "auto", "columns": []},
    learns_from_data=True,
)
class TargetEncoderCalculator(BaseCalculator):
    """Fit a sklearn ``TargetEncoder`` against ``y``, cross-fitting the training rows.

    A target is mandatory: with no ``y`` the node logs a warning and returns an
    empty artifact rather than failing. The target column is excluded from the
    encoded set, since encoding it against itself is degenerate. This calculator
    also satisfies ``TrainTransformCalculatorProtocol``, so the training split
    receives out-of-fold encodings — a row's own target never leaks into its
    feature — with the fold count capped by the smallest class count, while
    held-out splits go through the plain applier.
    """

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> TargetEncoderArtifact:  # pylint: disable=arguments-differ
        """Short-circuit an explicit empty column selection, else fit on the frame's own engine."""
        if user_picked_no_columns(config):
            return {}
        return cast(
            TargetEncoderArtifact,
            fit_dual_engine(
                (X, y) if y is not None else X,
                config,
                {"polars": _target_fit_polars, "pandas": _target_fit_pandas},
            ),
        )

    def fit_transform_train(
        self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
    ) -> tuple[TargetEncoderArtifact, Any]:
        """Fit sklearn TargetEncoder and cross-fit the pipeline training rows."""
        if user_picked_no_columns(config):
            return {}, df

        artifact, transformed = fit_transform_train_dual_engine(
            df,
            config,
            {
                "polars": _target_fit_transform_train_polars,
                "pandas": _target_fit_transform_train_pandas,
            },
        )
        return cast(
            TargetEncoderArtifact,
            artifact,
        ), transformed

    def infer_output_schema(
        self,
        input_schema: SkyulfSchema,
        config: dict[str, Any],
    ) -> SkyulfSchema | None:
        """Predict the schema only when ``target_type`` is explicitly binary or continuous.

        Anything else — ``"auto"`` included — can resolve to multiclass at fit
        time, which changes the column set in a data-dependent way, so ``None``
        is returned and callers fall back to runtime introspection.
        """
        # For binary/continuous targets, the encoder replaces values in
        # source columns in place — same column names, dtype becomes float
        # (per-column dtype is best-effort so we don't bother rewriting it).
        #
        # For multiclass targets, the apply logic (see
        # ``_target_apply_polars``/``_target_apply_pandas``) drops the
        # original columns and creates ``{col}_cls{i}`` columns instead — the
        # number of classes is data-dependent and unknown here, so we can't
        # confidently predict the output columns. The default/"auto"
        # target_type is resolved to multiclass at fit time whenever y has
        # more than two classes, so we must also treat "auto" as unknown
        # rather than assuming binary/continuous. Only the explicit
        # "binary"/"continuous" config values are confidently in-place.
        #
        # "continuous" is sklearn's name for a regression target. The value is
        # forwarded to ``TargetEncoder(target_type=...)`` verbatim and its
        # accepted set is {auto, binary, multiclass, continuous}, so the more
        # natural-sounding "regression" raises InvalidParameterError at fit and
        # could never reach this check through a working config.
        if config.get("target_type", "auto") not in ("binary", "continuous"):
            return None
        return input_schema

fit(X, y, config)

Short-circuit an explicit empty column selection, else fit on the frame's own engine.

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

@fit_method
def fit(self, X: Any, y: Any, config: dict[str, Any]) -> TargetEncoderArtifact:  # pylint: disable=arguments-differ
    """Short-circuit an explicit empty column selection, else fit on the frame's own engine."""
    if user_picked_no_columns(config):
        return {}
    return cast(
        TargetEncoderArtifact,
        fit_dual_engine(
            (X, y) if y is not None else X,
            config,
            {"polars": _target_fit_polars, "pandas": _target_fit_pandas},
        ),
    )

fit_transform_train(df, config)

Fit sklearn TargetEncoder and cross-fit the pipeline training rows.

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

def fit_transform_train(
    self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
) -> tuple[TargetEncoderArtifact, Any]:
    """Fit sklearn TargetEncoder and cross-fit the pipeline training rows."""
    if user_picked_no_columns(config):
        return {}, df

    artifact, transformed = fit_transform_train_dual_engine(
        df,
        config,
        {
            "polars": _target_fit_transform_train_polars,
            "pandas": _target_fit_transform_train_pandas,
        },
    )
    return cast(
        TargetEncoderArtifact,
        artifact,
    ), transformed

infer_output_schema(input_schema, config)

Predict the schema only when target_type is explicitly binary or continuous.

Anything else — "auto" included — can resolve to multiclass at fit time, which changes the column set in a data-dependent way, so None is returned and callers fall back to runtime introspection.

Source code in skyulf-core/skyulf/preprocessing/encoding/target.py

def infer_output_schema(
    self,
    input_schema: SkyulfSchema,
    config: dict[str, Any],
) -> SkyulfSchema | None:
    """Predict the schema only when ``target_type`` is explicitly binary or continuous.

    Anything else — ``"auto"`` included — can resolve to multiclass at fit
    time, which changes the column set in a data-dependent way, so ``None``
    is returned and callers fall back to runtime introspection.
    """
    # For binary/continuous targets, the encoder replaces values in
    # source columns in place — same column names, dtype becomes float
    # (per-column dtype is best-effort so we don't bother rewriting it).
    #
    # For multiclass targets, the apply logic (see
    # ``_target_apply_polars``/``_target_apply_pandas``) drops the
    # original columns and creates ``{col}_cls{i}`` columns instead — the
    # number of classes is data-dependent and unknown here, so we can't
    # confidently predict the output columns. The default/"auto"
    # target_type is resolved to multiclass at fit time whenever y has
    # more than two classes, so we must also treat "auto" as unknown
    # rather than assuming binary/continuous. Only the explicit
    # "binary"/"continuous" config values are confidently in-place.
    #
    # "continuous" is sklearn's name for a regression target. The value is
    # forwarded to ``TargetEncoder(target_type=...)`` verbatim and its
    # accepted set is {auto, binary, multiclass, continuous}, so the more
    # natural-sounding "regression" raises InvalidParameterError at fit and
    # could never reach this check through a working config.
    if config.get("target_type", "auto") not in ("binary", "continuous"):
        return None
    return input_schema

WOEEncoderApplier

Bases: BaseApplier

Replace each category with the Weight-of-Evidence value learned for it.

Both engines look the category up in the same fitted mapping, keyed by the scalar key of the value, and fall back to default for categories never seen at fit time. Nulls are keyed as the literal "nan" on both sides, so a missing category gets its own learned WOE instead of always taking the default (F-28). New artifacts distinguish numeric values from literal strings; older artifacts preserve their original string keys. Encoded columns become float.

Source code in skyulf-core/skyulf/preprocessing/encoding/woe.py

class WOEEncoderApplier(BaseApplier):
    """Replace each category with the Weight-of-Evidence value learned for it.

    Both engines look the category up in the same fitted mapping, keyed by the
    scalar key of the value, and fall back to ``default`` for categories
    never seen at fit time. Nulls are keyed as the literal ``"nan"`` on both
    sides, so a missing category gets its own learned WOE instead of always
    taking the default (F-28). New artifacts distinguish numeric values from
    literal strings; older artifacts preserve their original string keys.
    Encoded columns become float.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect learned mappings and their metadata without relearning target statistics."""
        fields = {"type", "columns", "mappings", "information_value", "default"}
        if isinstance(raw, dict) and "category_key_version" in raw:
            fields.add("category_key_version")
        if not local_state_fields(raw, "woe_encoder", fields, allow_empty=True):
            return raw
        columns = raw["columns"]
        if not isinstance(columns, (list, tuple)) or any(not isinstance(c, str) for c in columns):
            raise ValueError("Fitted WOE columns must be ordered strings.")
        uses_category_keys(raw)
        WOEEncoderApplier._validate_mappings(raw, columns)
        _validate_woe_values([raw["default"], *raw["information_value"].values()])
        return raw

    @staticmethod
    def _validate_mappings(raw: dict, columns: list[str]) -> None:
        """Keep every map and information-value entry attached to its fitted source column."""
        for field in ("mappings", "information_value"):
            if type(raw[field]) is not dict or set(raw[field]) != set(columns):
                raise ValueError("Fitted WOE maps and metadata must match selected columns.")
        for mapping in raw["mappings"].values():
            if type(mapping) is not dict or any(not isinstance(key, str) for key in mapping):
                raise ValueError("Fitted WOE maps must use string category keys.")
            _validate_woe_values(mapping.values())

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe fixed WOE category lookups without granting partition execution."""
        if engine not in ("pandas", "polars"):
            return None
        WOEEncoderApplier.validate_inference_state(state)
        # Legacy pandas datetime string formatting can depend on neighboring rows.
        legacy_pandas = (
            engine == "pandas" and state.get("columns") and not uses_category_keys(state)
        )
        context = "global" if legacy_pandas 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
        """Dispatch to the engine-specific WOE lookup, forwarding ``(X, y)`` only when present."""
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {"polars": _woe_apply_polars, "pandas": _woe_apply_pandas},
        )

apply(X, y, params)

Dispatch to the engine-specific WOE lookup, forwarding (X, y) only when present.

Source code in skyulf-core/skyulf/preprocessing/encoding/woe.py

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch to the engine-specific WOE lookup, forwarding ``(X, y)`` only when present."""
    return apply_dual_engine(
        (X, y) if y is not None else X,
        params,
        {"polars": _woe_apply_polars, "pandas": _woe_apply_pandas},
    )

inference_capability(state, *, engine) staticmethod

Describe fixed WOE category lookups without granting partition execution.

Source code in skyulf-core/skyulf/preprocessing/encoding/woe.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe fixed WOE category lookups without granting partition execution."""
    if engine not in ("pandas", "polars"):
        return None
    WOEEncoderApplier.validate_inference_state(state)
    # Legacy pandas datetime string formatting can depend on neighboring rows.
    legacy_pandas = (
        engine == "pandas" and state.get("columns") and not uses_category_keys(state)
    )
    context = "global" if legacy_pandas else "row"
    return ExecutionCapability(engine, "apply", "local", "preserve", context)

validate_inference_state(raw) staticmethod

Inspect learned mappings and their metadata without relearning target statistics.

Source code in skyulf-core/skyulf/preprocessing/encoding/woe.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect learned mappings and their metadata without relearning target statistics."""
    fields = {"type", "columns", "mappings", "information_value", "default"}
    if isinstance(raw, dict) and "category_key_version" in raw:
        fields.add("category_key_version")
    if not local_state_fields(raw, "woe_encoder", fields, allow_empty=True):
        return raw
    columns = raw["columns"]
    if not isinstance(columns, (list, tuple)) or any(not isinstance(c, str) for c in columns):
        raise ValueError("Fitted WOE columns must be ordered strings.")
    uses_category_keys(raw)
    WOEEncoderApplier._validate_mappings(raw, columns)
    _validate_woe_values([raw["default"], *raw["information_value"].values()])
    return raw

WOEEncoderCalculator

Bases: BaseCalculator

Fit the WOE mapping and per-column Information Value from a binary target.

Supervised and binary-only: a target without exactly two classes logs a warning and yields an empty artifact rather than failing. The target column is excluded from the encoded set, since WOE computed against itself is near-perfect separation. The math is engine-agnostic, so fit funnels both engines through one narrow pandas boundary while apply stays in the caller's engine. fit_transform_train cross-fits the training rows so a row's own target never leaks into its encoding.

regularization adds a pseudocount to each observed category in each target class. Class totals include regularization * n_categories so both smoothed distributions sum to one. Missing feature values count as a category; each cross-fitting complement uses its own observed categories.

Source code in skyulf-core/skyulf/preprocessing/encoding/woe.py

@NodeRegistry.register("WOEEncoder", WOEEncoderApplier)
@node_meta(
    id="WOEEncoder",
    name="WOE / IV Encoder",
    category="Preprocessing",
    description=(
        "Weight-of-Evidence encoder for binary classification. Replaces each "
        "category with its log-odds and records Information Value per column."
    ),
    params={"regularization": 0.5, "columns": []},
    learns_from_data=True,
)
class WOEEncoderCalculator(BaseCalculator):
    """Fit the WOE mapping and per-column Information Value from a binary target.

    Supervised and binary-only: a target without exactly two classes logs a
    warning and yields an empty artifact rather than failing. The target column
    is excluded from the encoded set, since WOE computed against itself is
    near-perfect separation. The math is engine-agnostic, so fit funnels both
    engines through one narrow pandas boundary while ``apply`` stays in the
    caller's engine. ``fit_transform_train`` cross-fits the training rows so a
    row's own target never leaks into its encoding.

    ``regularization`` adds a pseudocount to each observed category in each
    target class. Class totals include ``regularization * n_categories`` so
    both smoothed distributions sum to one. Missing feature values count as
    a category; each cross-fitting complement uses its own observed categories.
    """

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> Mapping[str, Any]:  # pylint: disable=arguments-differ
        """Short-circuit an explicit empty column selection, else fit via the shared boundary.

        Both engine keys dispatch to the same ``_woe_fit``: it converts to
        pandas internally, so there is no engine-specific fit to choose between.
        """
        if user_picked_no_columns(config):
            return {}
        return cast(
            Mapping[str, Any],
            fit_dual_engine(
                (X, y) if y is not None else X, config, {"polars": _woe_fit, "pandas": _woe_fit}
            ),
        )

    def fit_transform_train(
        self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
    ) -> tuple[Mapping[str, Any], Any]:
        """Fit the full-data WOE artifact and cross-fit the training rows."""
        if user_picked_no_columns(config):
            return {}, df

        artifact, transformed = fit_transform_train_dual_engine(
            df,
            config,
            {"polars": _woe_fit_transform_train_polars, "pandas": _woe_fit_transform_train_pandas},
        )
        return artifact, transformed

    def infer_output_schema(
        self,
        input_schema: SkyulfSchema,
        config: dict[str, Any],
    ) -> SkyulfSchema | None:
        """Pass the input schema through: WOE values replace the source columns in place."""
        # WOE replaces values in source columns in place (now float-valued);
        # column names are unchanged.
        return input_schema

fit(X, y, config)

Short-circuit an explicit empty column selection, else fit via the shared boundary.

Both engine keys dispatch to the same _woe_fit: it converts to pandas internally, so there is no engine-specific fit to choose between.

Source code in skyulf-core/skyulf/preprocessing/encoding/woe.py

@fit_method
def fit(self, X: Any, y: Any, config: dict[str, Any]) -> Mapping[str, Any]:  # pylint: disable=arguments-differ
    """Short-circuit an explicit empty column selection, else fit via the shared boundary.

    Both engine keys dispatch to the same ``_woe_fit``: it converts to
    pandas internally, so there is no engine-specific fit to choose between.
    """
    if user_picked_no_columns(config):
        return {}
    return cast(
        Mapping[str, Any],
        fit_dual_engine(
            (X, y) if y is not None else X, config, {"polars": _woe_fit, "pandas": _woe_fit}
        ),
    )

fit_transform_train(df, config)

Fit the full-data WOE artifact and cross-fit the training rows.

Source code in skyulf-core/skyulf/preprocessing/encoding/woe.py

def fit_transform_train(
    self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
) -> tuple[Mapping[str, Any], Any]:
    """Fit the full-data WOE artifact and cross-fit the training rows."""
    if user_picked_no_columns(config):
        return {}, df

    artifact, transformed = fit_transform_train_dual_engine(
        df,
        config,
        {"polars": _woe_fit_transform_train_polars, "pandas": _woe_fit_transform_train_pandas},
    )
    return artifact, transformed

infer_output_schema(input_schema, config)

Pass the input schema through: WOE values replace the source columns in place.

Source code in skyulf-core/skyulf/preprocessing/encoding/woe.py

def infer_output_schema(
    self,
    input_schema: SkyulfSchema,
    config: dict[str, Any],
) -> SkyulfSchema | None:
    """Pass the input schema through: WOE values replace the source columns in place."""
    # WOE replaces values in source columns in place (now float-valued);
    # column names are unchanged.
    return input_schema

detect_categorical_columns(df)

Auto-detect the string/categorical columns an encoder should target.

The fallback every encoder passes to resolve_columns when the user configured no explicit column list: polars frames are matched on Utf8/Categorical/Enum/Object dtypes, pandas frames via select_dtypes on object/category/string — so "encode categoricals" means the same thing on either engine.

Source code in skyulf-core/skyulf/preprocessing/encoding/_common.py

def detect_categorical_columns(df: Any) -> list[str]:
    """Auto-detect the string/categorical columns an encoder should target.

    The fallback every encoder passes to ``resolve_columns`` when the user
    configured no explicit column list: polars frames are matched on
    ``Utf8``/``Categorical``/``Enum``/``Object`` dtypes, pandas frames via
    ``select_dtypes`` on ``object``/``category``/``string`` — so "encode categoricals"
    means the same thing on either engine.
    """
    engine = get_engine(df)
    if engine.name == EngineName.POLARS:
        df_pl: Any = df
        return [
            c
            for c, t in zip(df_pl.columns, df_pl.dtypes, strict=True)
            if t in [pl.Utf8, pl.Categorical, pl.Object] or isinstance(t, pl.Enum)
        ]
    return df.select_dtypes(include=["object", "category", "string"]).columns.tolist()