API: preprocessing.bucketing

skyulf.preprocessing.bucketing

Binning / discretization nodes (GeneralBinning, CustomBinning, KBinsDiscretizer).

All Appliers share a single :class:BaseBinningApplier which routes through :func:apply_dual_engine. Per-engine helpers live at module level so each one stays at low CCN. Fits are sklearn / pandas-bound (pd.cut, pd.qcut, KBinsDiscretizer), so they convert via to_pandas once at the top and do not use :func:fit_dual_engine.

BaseBinningApplier

Bases: BaseApplier

Shared Applier for all binning Calculators.

Expects bin_edges in params: Dict[str, List[float]] mapping column names to bin edges.

missing_strategy="label" fills null/NaN and out-of-range bins with missing_label (default "Missing"). Polars emits String columns for this mode, including complete and empty batches; ordinal/bin-index values become "0", "1", etc. Pandas retains numeric bin values alongside the string sentinel in an object column. Range and custom labels keep their text. The default "keep" strategy preserves missing values.

Newly fitted range artifacts store canonical range_labels so their category keys remain identical across engines. Exact edge text replaces rounded labels if rounding would merge distinct bins. Older artifacts retain their original rendering; refit binning and downstream encoders together before moving a legacy range pipeline to another engine.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

class BaseBinningApplier(BaseApplier):
    """Shared Applier for all binning Calculators.

    Expects ``bin_edges`` in params: ``Dict[str, List[float]]`` mapping column
    names to bin edges.

    ``missing_strategy="label"`` fills null/NaN and out-of-range bins with
    ``missing_label`` (default ``"Missing"``). Polars emits String columns for
    this mode, including complete and empty batches; ordinal/bin-index values
    become ``"0"``, ``"1"``, etc. Pandas retains numeric bin values alongside
    the string sentinel in an object column. Range and custom labels keep
    their text. The default ``"keep"`` strategy preserves missing values.

    Newly fitted range artifacts store canonical ``range_labels`` so their
    category keys remain identical across engines. Exact edge text replaces
    rounded labels if rounding would merge distinct bins. Older artifacts retain
    their original rendering; refit binning and downstream encoders together
    before moving a legacy range pipeline to another engine.
    """

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch through the mapping-keyed dual-engine path; ``y`` passes through.

        Each entry in ``bin_edges`` gains a ``{col}{output_suffix}`` column, and
        ``drop_original`` removes only the columns that were actually binned. A
        column that cannot be binned (fewer than two unique edges, bad dtype) is
        left unbinned rather than failing the whole frame — warned about on the
        pandas path, silently skipped on the polars one.

        Conflicting retained output names and labels that cannot describe
        deduplicated edges raise ``ValueError`` before producing output. A
        dropped source name can be reused safely, including an empty suffix.
        """
        return apply_dual_engine(
            X, params, {"polars": _bucketing_apply_polars, "pandas": _bucketing_apply_pandas}
        )

apply(X, _y, params)

Dispatch through the mapping-keyed dual-engine path; y passes through.

Each entry in bin_edges gains a {col}{output_suffix} column, and drop_original removes only the columns that were actually binned. A column that cannot be binned (fewer than two unique edges, bad dtype) is left unbinned rather than failing the whole frame — warned about on the pandas path, silently skipped on the polars one.

Conflicting retained output names and labels that cannot describe deduplicated edges raise ValueError before producing output. A dropped source name can be reused safely, including an empty suffix.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch through the mapping-keyed dual-engine path; ``y`` passes through.

    Each entry in ``bin_edges`` gains a ``{col}{output_suffix}`` column, and
    ``drop_original`` removes only the columns that were actually binned. A
    column that cannot be binned (fewer than two unique edges, bad dtype) is
    left unbinned rather than failing the whole frame — warned about on the
    pandas path, silently skipped on the polars one.

    Conflicting retained output names and labels that cannot describe
    deduplicated edges raise ``ValueError`` before producing output. A
    dropped source name can be reused safely, including an empty suffix.
    """
    return apply_dual_engine(
        X, params, {"polars": _bucketing_apply_polars, "pandas": _bucketing_apply_pandas}
    )

CustomBinningApplier

Bases: GeneralBinningApplier

Describe fixed-edge inference while reusing the shared binning apply logic.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

class CustomBinningApplier(GeneralBinningApplier):
    """Describe fixed-edge inference while reusing the shared binning apply logic."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect actual local fitted state unchanged, including the explicit empty no-op."""
        return _validate_binning_inference_state(raw, allow_custom_labels=False)

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe row dependencies; exact batch schema equivalence needs a real probe."""
        if engine not in ("pandas", "polars"):
            return None
        CustomBinningApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

inference_capability(state, *, engine) staticmethod

Describe row dependencies; exact batch schema equivalence needs a real probe.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe row dependencies; exact batch schema equivalence needs a real probe."""
    if engine not in ("pandas", "polars"):
        return None
    CustomBinningApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect actual local fitted state unchanged, including the explicit empty no-op.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect actual local fitted state unchanged, including the explicit empty no-op."""
    return _validate_binning_inference_state(raw, allow_custom_labels=False)

CustomBinningCalculator

Bases: BaseCalculator

Apply user-supplied bin edges to selected columns.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@NodeRegistry.register("CustomBinning", CustomBinningApplier)
@node_meta(
    id="CustomBinning",
    name="Custom Binning",
    category="Preprocessing",
    description="Bin data using custom edges.",
    params={"bins": [], "columns": []},
    learns_from_data=True,
)
class CustomBinningCalculator(BaseCalculator):
    """Apply user-supplied bin edges to selected columns."""

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> GeneralBinningArtifact:
        """Attach one shared, sorted edge list from ``config["bins"]`` to every selected column.

        Edges are fixed, but automatic column selection examines observed
        values to exclude binary, constant, and all-missing columns.
        Unlike :class:`GeneralBinningCalculator`
        the same edges go to every column, so a multi-column selection spanning
        different scales will push the out-of-scale columns' values outside the
        outer bins, where both engines null them. An empty or absent ``bins``
        leaves ``bin_edges`` empty, which makes apply a no-op.
        """
        if user_picked_no_columns(config):
            return cast(GeneralBinningArtifact, {})

        input_columns = list(X.columns)
        X, columns = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
        bins = config.get("bins")

        bin_edges_map: dict[str, list[float]] = {}
        if bins:
            sorted_bins = sorted(bins)
            for col in columns:
                if col in X.columns:
                    bin_edges_map[col] = sorted_bins

        artifact: dict[str, Any] = {
            "type": "general_binning",  # Reuses GeneralBinningApplier.
            "bin_edges": bin_edges_map,
        }
        artifact.update(_passthrough_artifact_options(config))
        _validate_binning_artifact(input_columns, artifact)
        _store_range_labels(artifact)
        return cast(GeneralBinningArtifact, artifact)

fit(X, _y, config)

Attach one shared, sorted edge list from config["bins"] to every selected column.

Edges are fixed, but automatic column selection examines observed values to exclude binary, constant, and all-missing columns. Unlike :class:GeneralBinningCalculator the same edges go to every column, so a multi-column selection spanning different scales will push the out-of-scale columns' values outside the outer bins, where both engines null them. An empty or absent bins leaves bin_edges empty, which makes apply a no-op.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> GeneralBinningArtifact:
    """Attach one shared, sorted edge list from ``config["bins"]`` to every selected column.

    Edges are fixed, but automatic column selection examines observed
    values to exclude binary, constant, and all-missing columns.
    Unlike :class:`GeneralBinningCalculator`
    the same edges go to every column, so a multi-column selection spanning
    different scales will push the out-of-scale columns' values outside the
    outer bins, where both engines null them. An empty or absent ``bins``
    leaves ``bin_edges`` empty, which makes apply a no-op.
    """
    if user_picked_no_columns(config):
        return cast(GeneralBinningArtifact, {})

    input_columns = list(X.columns)
    X, columns = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)
    bins = config.get("bins")

    bin_edges_map: dict[str, list[float]] = {}
    if bins:
        sorted_bins = sorted(bins)
        for col in columns:
            if col in X.columns:
                bin_edges_map[col] = sorted_bins

    artifact: dict[str, Any] = {
        "type": "general_binning",  # Reuses GeneralBinningApplier.
        "bin_edges": bin_edges_map,
    }
    artifact.update(_passthrough_artifact_options(config))
    _validate_binning_artifact(input_columns, artifact)
    _store_range_labels(artifact)
    return cast(GeneralBinningArtifact, artifact)

GeneralBinningApplier

Bases: BaseBinningApplier

Inspect fitted bin decisions while inheriting the existing shared apply.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

class GeneralBinningApplier(BaseBinningApplier):
    """Inspect fitted bin decisions while inheriting the existing shared apply."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved edges, optional labels and display choices without fitting."""
        return _validate_binning_inference_state(raw, allow_custom_labels=True)

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe saved bin application as local row-preserving inference."""
        if engine not in ("pandas", "polars"):
            return None
        GeneralBinningApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

inference_capability(state, *, engine) staticmethod

Describe saved bin application as local row-preserving inference.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe saved bin application as local row-preserving inference."""
    if engine not in ("pandas", "polars"):
        return None
    GeneralBinningApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect saved edges, optional labels and display choices without fitting.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved edges, optional labels and display choices without fitting."""
    return _validate_binning_inference_state(raw, allow_custom_labels=True)

GeneralBinningCalculator

Bases: BaseCalculator

Master calculator that handles mixed strategies and per-column overrides.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@NodeRegistry.register("GeneralBinning", GeneralBinningApplier)
@node_meta(
    id="GeneralBinning",
    name="General Binning",
    category="Preprocessing",
    description="Bin continuous data into intervals.",
    params={"n_bins": 5, "strategy": "uniform", "columns": []},
    learns_from_data=True,
)
class GeneralBinningCalculator(BaseCalculator):
    """Master calculator that handles mixed strategies and per-column overrides."""

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> GeneralBinningArtifact:  # pylint: disable=arguments-differ
        """Fit bin edges per column under each column's own strategy.

        An explicitly empty ``columns`` list short-circuits to an empty artifact
        (the user unchecked everything), whereas an absent one falls through to
        :func:`detect_numeric_columns`. Only the selected columns are converted
        to pandas, since the fits are ``pd.cut`` / ``pd.qcut`` / sklearn-bound.
        Columns whose strategy cannot be fit are logged and skipped rather than
        failing the node, so the artifact may cover fewer columns than requested.
        """
        if user_picked_no_columns(config):
            return cast(GeneralBinningArtifact, {})

        input_columns = list(X.columns)
        X, columns = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)

        defaults = {
            "default_n_bins": config.get("n_bins", 5),
            "n_bins": config.get("equal_width_bins", config.get("n_bins", 5)),
            "q_bins": config.get("equal_frequency_bins", config.get("n_bins", 5)),
            "duplicates": config.get("duplicates", "drop"),
        }

        valid_cols = [c for c in columns if c in X.columns]
        bin_edges_map: dict[str, list[float]] = {}
        custom_labels_map: dict[str, Any] = {}

        for col in valid_cols:
            _fit_one_column_into_maps(X, col, config, defaults, bin_edges_map, custom_labels_map)

        artifact: dict[str, Any] = {
            "type": "general_binning",
            "bin_edges": bin_edges_map,
            "custom_labels": custom_labels_map,
        }
        artifact.update(_passthrough_artifact_options(config))
        _validate_binning_artifact(input_columns, artifact)
        _store_range_labels(artifact)
        return cast(GeneralBinningArtifact, artifact)

fit(X, _y, config)

Fit bin edges per column under each column's own strategy.

An explicitly empty columns list short-circuits to an empty artifact (the user unchecked everything), whereas an absent one falls through to :func:detect_numeric_columns. Only the selected columns are converted to pandas, since the fits are pd.cut / pd.qcut / sklearn-bound. Columns whose strategy cannot be fit are logged and skipped rather than failing the node, so the artifact may cover fewer columns than requested.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> GeneralBinningArtifact:  # pylint: disable=arguments-differ
    """Fit bin edges per column under each column's own strategy.

    An explicitly empty ``columns`` list short-circuits to an empty artifact
    (the user unchecked everything), whereas an absent one falls through to
    :func:`detect_numeric_columns`. Only the selected columns are converted
    to pandas, since the fits are ``pd.cut`` / ``pd.qcut`` / sklearn-bound.
    Columns whose strategy cannot be fit are logged and skipped rather than
    failing the node, so the artifact may cover fewer columns than requested.
    """
    if user_picked_no_columns(config):
        return cast(GeneralBinningArtifact, {})

    input_columns = list(X.columns)
    X, columns = resolve_columns_then_to_pandas(X, config, detect_numeric_columns)

    defaults = {
        "default_n_bins": config.get("n_bins", 5),
        "n_bins": config.get("equal_width_bins", config.get("n_bins", 5)),
        "q_bins": config.get("equal_frequency_bins", config.get("n_bins", 5)),
        "duplicates": config.get("duplicates", "drop"),
    }

    valid_cols = [c for c in columns if c in X.columns]
    bin_edges_map: dict[str, list[float]] = {}
    custom_labels_map: dict[str, Any] = {}

    for col in valid_cols:
        _fit_one_column_into_maps(X, col, config, defaults, bin_edges_map, custom_labels_map)

    artifact: dict[str, Any] = {
        "type": "general_binning",
        "bin_edges": bin_edges_map,
        "custom_labels": custom_labels_map,
    }
    artifact.update(_passthrough_artifact_options(config))
    _validate_binning_artifact(input_columns, artifact)
    _store_range_labels(artifact)
    return cast(GeneralBinningArtifact, artifact)

KBinsDiscretizerApplier

Bases: GeneralBinningApplier

Inspect sklearn-selected edges while reusing the shared cut implementation.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

class KBinsDiscretizerApplier(GeneralBinningApplier):
    """Inspect sklearn-selected edges while reusing the shared cut implementation."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the same saved artifact produced by GeneralBinning's sklearn path."""
        return GeneralBinningApplier.validate_inference_state(raw)

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe fixed learned bins without granting distributed execution."""
        if engine not in ("pandas", "polars"):
            return None
        KBinsDiscretizerApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

inference_capability(state, *, engine) staticmethod

Describe fixed learned bins without granting distributed execution.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe fixed learned bins without granting distributed execution."""
    if engine not in ("pandas", "polars"):
        return None
    KBinsDiscretizerApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect the same saved artifact produced by GeneralBinning's sklearn path.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect the same saved artifact produced by GeneralBinning's sklearn path."""
    return GeneralBinningApplier.validate_inference_state(raw)

KBinsDiscretizerCalculator

Bases: GeneralBinningCalculator

Thin wrapper around :class:GeneralBinningCalculator with kbins strategy.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

@NodeRegistry.register("KBinsDiscretizer", KBinsDiscretizerApplier)
@node_meta(
    id="KBinsDiscretizer",
    name="K-Bins Discretizer",
    category="Preprocessing",
    description="Bin continuous data into intervals using sklearn KBinsDiscretizer.",
    params={"n_bins": 5, "encode": "ordinal", "strategy": "quantile", "columns": []},
    learns_from_data=True,
)
class KBinsDiscretizerCalculator(GeneralBinningCalculator):
    """Thin wrapper around :class:`GeneralBinningCalculator` with ``kbins`` strategy."""

    def fit(  # pylint: disable=arguments-differ
        self,
        df: pd.DataFrame | SkyulfDataFrame | tuple[Any, ...] | Any,
        config: dict[str, Any],
    ) -> GeneralBinningArtifact:
        """Translate this node's public params onto GeneralBinning's internal names, then delegate.

        On a copy of *config* (the caller's dict is never mutated) it forces
        ``strategy="kbins"``, moves ``n_bins`` to ``kbins_n_bins``, and preserves
        a user-chosen non-``kbins`` ``strategy`` as ``kbins_strategy`` so
        ``quantile`` / ``uniform`` / ``kmeans`` still reach sklearn. Written with
        the raw ``(df, config)`` signature rather than ``@fit_method`` because it
        only rewrites config and forwards — the parent's decorated ``fit`` does
        the ``(X, y)`` unpack.
        """
        new_config = config.copy()
        new_config["strategy"] = "kbins"
        if "n_bins" in config:
            new_config["kbins_n_bins"] = config["n_bins"]
        if "strategy" in config and config["strategy"] != "kbins":
            new_config["kbins_strategy"] = config["strategy"]
        return super().fit(df, new_config)  # pylint: disable=no-value-for-parameter

fit(df, config)

Translate this node's public params onto GeneralBinning's internal names, then delegate.

On a copy of config (the caller's dict is never mutated) it forces strategy="kbins", moves n_bins to kbins_n_bins, and preserves a user-chosen non-kbins strategy as kbins_strategy so quantile / uniform / kmeans still reach sklearn. Written with the raw (df, config) signature rather than @fit_method because it only rewrites config and forwards — the parent's decorated fit does the (X, y) unpack.

Source code in skyulf-core/skyulf/preprocessing/bucketing.py

def fit(  # pylint: disable=arguments-differ
    self,
    df: pd.DataFrame | SkyulfDataFrame | tuple[Any, ...] | Any,
    config: dict[str, Any],
) -> GeneralBinningArtifact:
    """Translate this node's public params onto GeneralBinning's internal names, then delegate.

    On a copy of *config* (the caller's dict is never mutated) it forces
    ``strategy="kbins"``, moves ``n_bins`` to ``kbins_n_bins``, and preserves
    a user-chosen non-``kbins`` ``strategy`` as ``kbins_strategy`` so
    ``quantile`` / ``uniform`` / ``kmeans`` still reach sklearn. Written with
    the raw ``(df, config)`` signature rather than ``@fit_method`` because it
    only rewrites config and forwards — the parent's decorated ``fit`` does
    the ``(X, y)`` unpack.
    """
    new_config = config.copy()
    new_config["strategy"] = "kbins"
    if "n_bins" in config:
        new_config["kbins_n_bins"] = config["n_bins"]
    if "strategy" in config and config["strategy"] != "kbins":
        new_config["kbins_strategy"] = config["strategy"]
    return super().fit(df, new_config)  # pylint: disable=no-value-for-parameter