API: preprocessing.encoding

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

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

class FrequencyEncoderApplier(BaseApplier):
    """Replay fixed category statistics without recomputing inference-batch frequencies."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Check exact learned counts and dimensions before granting local replay."""
        fields = {"type", "columns", "mode", "category_key_version", "n_samples", "mappings"}
        if not local_state_fields(raw, "frequency_encoder", fields, allow_empty=True):
            return raw
        uses_category_keys(raw)
        _mode(raw["mode"])
        columns = raw["columns"]
        if type(columns) is not list or not columns or any(type(c) is not str for c in columns):
            raise ValueError("Fitted frequency columns must be a nonempty ordered string list.")
        if len(columns) != len(set(columns)):
            raise ValueError("Fitted frequency columns must be unique.")
        FrequencyEncoderApplier._validate_counts(raw)
        return raw

    @staticmethod
    def _validate_counts(raw: dict) -> None:
        """Each column counts the same positive training population exactly once."""
        n_samples = raw["n_samples"]
        if type(n_samples) is not int or n_samples < 1:
            raise ValueError("Fitted frequency n_samples must be a positive integer.")
        mappings = raw["mappings"]
        if type(mappings) is not dict or set(mappings) != set(raw["columns"]):
            raise ValueError("Fitted frequency mappings must match selected columns.")
        for mapping in mappings.values():
            FrequencyEncoderApplier._validate_column_counts(mapping, n_samples)

    @staticmethod
    def _validate_column_counts(mapping: Any, n_samples: int) -> None:
        """Inspect one categorical histogram without casting or correcting its entries."""
        if type(mapping) is not dict or any(type(key) is not str for key in mapping):
            raise ValueError("Fitted frequency categories must be string keys.")
        if any(type(count) is not int or count < 1 for count in mapping.values()):
            raise ValueError("Fitted frequency counts must be positive integers.")
        if sum(mapping.values()) != n_samples:
            raise ValueError("Fitted frequency counts disagree with n_samples.")

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe validated row-local lookups without granting worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        FrequencyEncoderApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict) -> Any:
        """Replace selected values while retaining row order, index, and target."""
        return apply_dual_engine(
            (X, y) if y is not None else X, params, {"pandas": _apply, "polars": _apply}
        )

@NodeRegistry.register("FrequencyEncoder", FrequencyEncoderApplier)
@node_meta(
    id="FrequencyEncoder",
    name="Frequency Encoder",
    category="Preprocessing",
    description="Replace categories with training frequencies or counts; unseen values become zero.",
    params={"columns": [], "mode": "frequency"},
    learns_from_data=True,
)
class FrequencyEncoderCalculator(BaseCalculator):
    """Learn counts including missing values; frequency divides by all training rows.

    ``mode='frequency'`` emits float64 and ``mode='count'`` emits int64.
    Omitted columns select categorical features, excluding the named target.
    An explicit empty selection is a no-op. Selected empty training data fails.
    """

    @fit_method
    def fit(self, X: Any, y: Any, config: dict) -> dict:
        """Persist category counts for the selected native dataframe columns."""
        if user_picked_no_columns(config):
            return {}
        return fit_dual_engine(
            (X, y) if y is not None else X, config, {"pandas": _fit, "polars": _fit}
        )

class 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},
        )

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

        n_features = config.get("n_features", 8)
        if isinstance(n_features, bool) or not isinstance(n_features, Integral) or n_features <= 0:
            raise ValueError("HashEncoder n_features must be a positive integer.")
        return {
            "type": "hash_encoder",
            "columns": cols,
            "n_features": int(n_features),
            "numeric_normalization_version": 2,
        }

    def infer_output_schema(
        self,
        input_schema: SkyulfSchema,
        config: dict[str, Any],
    ) -> SkyulfSchema | None:
        """Describe explicit signed integer buckets; abstain for automatic or unbounded output."""
        columns = config.get("columns")
        if columns is None:
            return None
        columns = [col for col in columns if col != config.get("target_column")]
        width = config.get("n_features", 8)
        if columns and not (
            isinstance(width, Integral) and not isinstance(width, bool) and 0 < width <= 2**63
        ):
            return None
        return encoded_columns_schema(input_schema, columns, "int64")

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

@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 | None:
        """Describe selected integer labels, retaining the target-only empty-selection default."""
        target = config.get("target_column")
        columns = config.get("columns") or ([target] if isinstance(target, str) else [])
        return encoded_columns_schema(input_schema, columns, "int64")

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

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

@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,
        "categories": None,
        "column_names": "sklearn",
    },
    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.

    Optional ``categories`` maps every explicitly selected column to an ordered
    list of unique strings. Unobserved levels retain outputs. Fixed categories
    require ``max_categories=None`` and ``include_missing=False``; unknowns and
    nulls produce zeros with ``handle_unknown='ignore'`` or raise with ``'error'``.
    ``drop_first`` removes the first configured level. ``column_names='slug'``
    folds category labels to ASCII; the default ``'sklearn'`` preserves names.
    """

    @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."""
        _fixed_category_options(config)
        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},
            ),
        )

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

@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 | None:
        """Describe explicit float32 codes, including the embedded target-only empty selection."""
        columns = config.get("columns")
        if columns == []:
            target = config.get("target_column")
            columns = [target] if isinstance(target, str) else []
        return encoded_columns_schema(input_schema, columns, "float32")

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

class RareCategoryGroupingApplier(BaseApplier):
    """Replay learned text-category membership without discovering new frequent values."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Reject malformed or ambiguous saved membership and fallback values."""
        fields = {"type", "columns", "category_key_version", "kept_categories", "replacement"}
        if not local_state_fields(raw, "rare_category_grouping", fields, allow_empty=True):
            return raw
        uses_category_keys(raw)
        _replacement(raw["replacement"])
        columns = raw["columns"]
        if type(columns) is not list or not columns or any(type(c) is not str for c in columns):
            raise ValueError("Fitted rare-grouping columns must be a nonempty ordered string list.")
        if len(columns) != len(set(columns)):
            raise ValueError("Fitted rare-grouping columns must be unique.")
        RareCategoryGroupingApplier._validate_membership(raw)
        return raw

    @staticmethod
    def _validate_membership(raw: dict) -> None:
        """Keep each learned category set attached to its exact source column."""
        kept = raw["kept_categories"]
        if type(kept) is not dict or set(kept) != set(raw["columns"]):
            raise ValueError("Fitted rare-grouping membership must match selected columns.")
        for keys in kept.values():
            if type(keys) is not list or any(type(key) is not str for key in keys):
                raise ValueError("Fitted rare-grouping keys must be ordered strings.")
            if len(keys) != len(set(keys)):
                raise ValueError("Fitted rare-grouping keys must be unique.")

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe validated row-local membership without granting worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        RareCategoryGroupingApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict) -> Any:
        """Replace selected columns with nullable strings, retaining index and target."""
        return apply_dual_engine(
            (X, y) if y is not None else X, params, {"pandas": _apply, "polars": _apply}
        )

@NodeRegistry.register("RareCategoryGrouping", RareCategoryGroupingApplier)
@node_meta(
    id="RareCategoryGrouping",
    name="Rare Category Grouping",
    category="Preprocessing",
    description="Keep frequent text categories and group rare or unseen values under a label.",
    params={"columns": [], "min_count": 2, "min_share": 0.0, "replacement": "Other"},
    learns_from_data=True,
)
class RareCategoryGroupingCalculator(BaseCalculator):
    """Learn text categories meeting both count and share thresholds, including null.

    Frequent missing values stay null; rare and unseen categories become the
    replacement string. A replacement already present in the data merges with
    grouped values. Numeric categories require explicit Casting before this node.
    """

    @fit_method
    def fit(self, X: Any, y: Any, config: dict) -> dict:
        """Persist fixed membership using training rows only."""
        if user_picked_no_columns(config):
            return {}
        return fit_dual_engine(
            (X, y) if y is not None else X, config, {"pandas": _fit, "polars": _fit}
        )

class 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.
    New artifacts use canonical missing and numeric feature keys on both
    engines; artifacts without a key version retain raw sklearn replay.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the saved full-data encoder while leaving cross-fitted training values alone."""
        fields = {"type", "columns", "encoder_object"}
        if isinstance(raw, dict):
            fields.update({"category_key_version"}.intersection(raw))
        if not local_state_fields(raw, "target_encoder", fields, allow_empty=True):
            return raw
        uses_category_keys(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},
        )