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

Applier registered for the CustomBinning node id.

Empty subclass — custom edges are applied by exactly the same cut logic as fitted ones, so nothing here needs to differ.

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

class CustomBinningApplier(GeneralBinningApplier):
    """Applier registered for the ``CustomBinning`` node id.

    Empty subclass — custom edges are applied by exactly the same cut logic as
    fitted ones, so nothing here needs to differ.
    """

    pass

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

Applier registered for the GeneralBinning node id.

Empty subclass: every binning behaviour is inherited from :class:BaseBinningApplier. It exists as its own type so this node id registers against a distinct Applier class, and as the base for the CustomBinning and KBinsDiscretizer appliers.

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

class GeneralBinningApplier(BaseBinningApplier):
    """Applier registered for the ``GeneralBinning`` node id.

    Empty subclass: every binning behaviour is inherited from
    :class:`BaseBinningApplier`. It exists as its own type so this node id
    registers against a distinct Applier class, and as the base for the
    ``CustomBinning`` and ``KBinsDiscretizer`` appliers.
    """

    pass

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

Applier registered for the KBinsDiscretizer node id.

Empty subclass — sklearn only picks the edges at fit time; applying them is the same cut as for any other binning node.

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

class KBinsDiscretizerApplier(GeneralBinningApplier):
    """Applier registered for the ``KBinsDiscretizer`` node id.

    Empty subclass — sklearn only picks the edges at fit time; applying them is
    the same cut as for any other binning node.
    """

    pass

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