API: preprocessing.resampling

skyulf.preprocessing.resampling

Resampling nodes (Oversampling, Undersampling).

Both Appliers route through :func:apply_dual_engine. Imblearn is purely pandas/numpy-bound, so the Polars path round-trips through pandas (convert in, run sampler, convert back) while keeping all engine handling out of class bodies. Sampler construction is split per-method and per-family so each helper stays at low CCN.

OversamplingApplier

Bases: BaseApplier

Grow the minority classes with the imblearn over-sampler named in the artifact.

Sees the training split only: :mod:.pipeline excludes the resampling nodes from apply_on_test/apply_on_validation, so synthetic rows can never reach a held-out split and inflate its metrics.

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

class OversamplingApplier(BaseApplier):
    """Grow the minority classes with the imblearn over-sampler named in the artifact.

    Sees the training split only: :mod:`.pipeline` excludes the resampling nodes
    from ``apply_on_test``/``apply_on_validation``, so synthetic rows can never
    reach a held-out split and inflate its metrics.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect the saved sampler configuration without fitting optional estimators."""
        return _validate_resampling_state(raw, "oversampling", "smote")

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Require the full class distribution; mixed SMOTETomek effects stay unknown."""
        if engine not in ("pandas", "polars"):
            return None
        OversamplingApplier.validate_inference_state(state)
        if not _over_context_known(state):
            return None
        return ExecutionCapability(engine, "apply", "local", "expand", "global")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Resample ``(X, y)`` with the configured over-sampler.

        ``y`` is lifted out of ``X`` via ``target_column`` when not supplied
        separately; with no usable target the data passes through untouched, as
        it does for a ``method`` no builder recognizes.

        Raises:
            ValueError: If any feature column is non-numeric, which imblearn
                cannot resample. Encode first.
        """
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {
                "polars": lambda Xi, yi, p: _resample_polars(
                    Xi, yi, p, _build_oversampler, "smote"
                ),
                "pandas": lambda Xi, yi, p: _resample_pandas(
                    Xi, yi, p, _build_oversampler, "smote"
                ),
            },
        )

apply(X, y, params)

Resample (X, y) with the configured over-sampler.

y is lifted out of X via target_column when not supplied separately; with no usable target the data passes through untouched, as it does for a method no builder recognizes.

Raises:

Type	Description
ValueError	If any feature column is non-numeric, which imblearn cannot resample. Encode first.

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

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Resample ``(X, y)`` with the configured over-sampler.

    ``y`` is lifted out of ``X`` via ``target_column`` when not supplied
    separately; with no usable target the data passes through untouched, as
    it does for a ``method`` no builder recognizes.

    Raises:
        ValueError: If any feature column is non-numeric, which imblearn
            cannot resample. Encode first.
    """
    return apply_dual_engine(
        (X, y) if y is not None else X,
        params,
        {
            "polars": lambda Xi, yi, p: _resample_polars(
                Xi, yi, p, _build_oversampler, "smote"
            ),
            "pandas": lambda Xi, yi, p: _resample_pandas(
                Xi, yi, p, _build_oversampler, "smote"
            ),
        },
    )

inference_capability(state, *, engine) staticmethod

Require the full class distribution; mixed SMOTETomek effects stay unknown.

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

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Require the full class distribution; mixed SMOTETomek effects stay unknown."""
    if engine not in ("pandas", "polars"):
        return None
    OversamplingApplier.validate_inference_state(state)
    if not _over_context_known(state):
        return None
    return ExecutionCapability(engine, "apply", "local", "expand", "global")

validate_inference_state(raw) staticmethod

Inspect the saved sampler configuration without fitting optional estimators.

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

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect the saved sampler configuration without fitting optional estimators."""
    return _validate_resampling_state(raw, "oversampling", "smote")

OversamplingCalculator

Bases: BaseCalculator

Configure class balancing by oversampling the minority class.

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

@NodeRegistry.register("Oversampling", OversamplingApplier)
@node_meta(
    id="Oversampling",
    name="Oversampling",
    category="Preprocessing",
    description="Resample dataset to balance classes by oversampling minority class.",
    params={"method": "smote", "target_column": "target", "sampling_strategy": "auto"},
    learns_from_data=True,
)
class OversamplingCalculator(BaseCalculator):
    """Configure class balancing by oversampling the minority class."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return ``input_schema`` untouched."""
        # Resampling changes row counts only; column set is preserved.
        return input_schema

    @fit_method
    def fit(self, _X: Any, _y: Any, config: dict[str, Any]) -> OversamplingArtifact:  # pylint: disable=arguments-differ
        """Snapshot the sampler settings from ``config`` into the artifact.

        Neither ``_X`` nor ``_y`` is read: the artifact is pure configuration,
        and the class imbalance it corrects is only observed when the applier
        runs the sampler on the training split.
        """
        return {
            "type": "oversampling",
            "synthetic_weight": config.get("synthetic_weight"),
            "method": config.get("method", "smote"),
            "target_column": config.get("target_column"),
            "sampling_strategy": config.get("sampling_strategy", "auto"),
            "random_state": config.get("random_state", 42),
            "k_neighbors": config.get("k_neighbors", 5),
            "m_neighbors": config.get("m_neighbors", 10),
            "kind": config.get("kind", "borderline-1"),
            "svm_estimator": config.get("svm_estimator"),
            "out_step": config.get("out_step", 0.5),
            "kmeans_estimator": config.get("kmeans_estimator"),
            "cluster_balance_threshold": config.get("cluster_balance_threshold", 0.1),
            "density_exponent": config.get("density_exponent", "auto"),
            "n_jobs": config.get("n_jobs", -1),
        }

fit(_X, _y, config)

Snapshot the sampler settings from config into the artifact.

Neither _X nor _y is read: the artifact is pure configuration, and the class imbalance it corrects is only observed when the applier runs the sampler on the training split.

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

@fit_method
def fit(self, _X: Any, _y: Any, config: dict[str, Any]) -> OversamplingArtifact:  # pylint: disable=arguments-differ
    """Snapshot the sampler settings from ``config`` into the artifact.

    Neither ``_X`` nor ``_y`` is read: the artifact is pure configuration,
    and the class imbalance it corrects is only observed when the applier
    runs the sampler on the training split.
    """
    return {
        "type": "oversampling",
        "synthetic_weight": config.get("synthetic_weight"),
        "method": config.get("method", "smote"),
        "target_column": config.get("target_column"),
        "sampling_strategy": config.get("sampling_strategy", "auto"),
        "random_state": config.get("random_state", 42),
        "k_neighbors": config.get("k_neighbors", 5),
        "m_neighbors": config.get("m_neighbors", 10),
        "kind": config.get("kind", "borderline-1"),
        "svm_estimator": config.get("svm_estimator"),
        "out_step": config.get("out_step", 0.5),
        "kmeans_estimator": config.get("kmeans_estimator"),
        "cluster_balance_threshold": config.get("cluster_balance_threshold", 0.1),
        "density_exponent": config.get("density_exponent", "auto"),
        "n_jobs": config.get("n_jobs", -1),
    }

infer_output_schema(input_schema, config)

Return input_schema untouched.

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

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Return ``input_schema`` untouched."""
    # Resampling changes row counts only; column set is preserved.
    return input_schema

UndersamplingApplier

Bases: BaseApplier

Shrink the majority classes with the imblearn under-sampler named in the artifact.

Sees the training split only: :mod:.pipeline excludes the resampling nodes from apply_on_test/apply_on_validation. That guard matters more here than for oversampling, because these samplers delete real rows rather than synthesise new ones.

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

class UndersamplingApplier(BaseApplier):
    """Shrink the majority classes with the imblearn under-sampler named in the artifact.

    Sees the training split only: :mod:`.pipeline` excludes the resampling nodes
    from ``apply_on_test``/``apply_on_validation``. That guard matters more here
    than for oversampling, because these samplers delete real rows rather than
    synthesise new ones.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved selection settings without loading the optional sampler package."""
        return _validate_resampling_state(raw, "undersampling", "random_under_sampling")

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Require full class context; prediction still skips every sampling method."""
        if engine not in ("pandas", "polars"):
            return None
        UndersamplingApplier.validate_inference_state(state)
        if not _under_context_known(state):
            return None
        return ExecutionCapability(engine, "apply", "local", "filter", "global")

    @apply_method
    def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:
        """Resample ``(X, y)`` with the configured under-sampler.

        ``y`` is lifted out of ``X`` via ``target_column`` when not supplied
        separately; with no usable target the data passes through untouched, as
        it does for a ``method`` no builder recognizes.

        Raises:
            ValueError: If any feature column is non-numeric, which imblearn
                cannot resample. Encode first.
        """
        return apply_dual_engine(
            (X, y) if y is not None else X,
            params,
            {
                "polars": lambda Xi, yi, p: _resample_polars(
                    Xi, yi, p, _build_undersampler, "random_under_sampling"
                ),
                "pandas": lambda Xi, yi, p: _resample_pandas(
                    Xi, yi, p, _build_undersampler, "random_under_sampling"
                ),
            },
        )

apply(X, y, params)

Resample (X, y) with the configured under-sampler.

y is lifted out of X via target_column when not supplied separately; with no usable target the data passes through untouched, as it does for a method no builder recognizes.

Raises:

Type	Description
ValueError	If any feature column is non-numeric, which imblearn cannot resample. Encode first.

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

@apply_method
def apply(self, X: Any, y: Any, params: dict[str, Any]) -> Any:
    """Resample ``(X, y)`` with the configured under-sampler.

    ``y`` is lifted out of ``X`` via ``target_column`` when not supplied
    separately; with no usable target the data passes through untouched, as
    it does for a ``method`` no builder recognizes.

    Raises:
        ValueError: If any feature column is non-numeric, which imblearn
            cannot resample. Encode first.
    """
    return apply_dual_engine(
        (X, y) if y is not None else X,
        params,
        {
            "polars": lambda Xi, yi, p: _resample_polars(
                Xi, yi, p, _build_undersampler, "random_under_sampling"
            ),
            "pandas": lambda Xi, yi, p: _resample_pandas(
                Xi, yi, p, _build_undersampler, "random_under_sampling"
            ),
        },
    )

inference_capability(state, *, engine) staticmethod

Require full class context; prediction still skips every sampling method.

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

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Require full class context; prediction still skips every sampling method."""
    if engine not in ("pandas", "polars"):
        return None
    UndersamplingApplier.validate_inference_state(state)
    if not _under_context_known(state):
        return None
    return ExecutionCapability(engine, "apply", "local", "filter", "global")

validate_inference_state(raw) staticmethod

Inspect saved selection settings without loading the optional sampler package.

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

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved selection settings without loading the optional sampler package."""
    return _validate_resampling_state(raw, "undersampling", "random_under_sampling")

UndersamplingCalculator

Bases: BaseCalculator

Configure class balancing by undersampling the majority class.

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

@NodeRegistry.register("Undersampling", UndersamplingApplier)
@node_meta(
    id="Undersampling",
    name="Undersampling",
    category="Preprocessing",
    description="Resample dataset to balance classes by undersampling majority class.",
    params={
        "method": "random_under_sampling",
        "target_column": "target",
        "sampling_strategy": "auto",
    },
    learns_from_data=True,
)
class UndersamplingCalculator(BaseCalculator):
    """Configure class balancing by undersampling the majority class."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Return ``input_schema`` untouched."""
        # Resampling changes row counts only; column set is preserved.
        return input_schema

    @fit_method
    def fit(self, _X: Any, _y: Any, config: dict[str, Any]) -> UndersamplingArtifact:  # pylint: disable=arguments-differ
        """Snapshot the sampler settings from ``config`` into the artifact.

        Neither ``_X`` nor ``_y`` is read: the artifact is pure configuration.
        Several of its keys are sampler-specific, so each method consumes only
        the subset its imblearn class accepts.
        """
        return {
            "type": "undersampling",
            "method": config.get("method", "random_under_sampling"),
            "target_column": config.get("target_column"),
            "sampling_strategy": config.get("sampling_strategy", "auto"),
            "random_state": config.get("random_state", 42),
            "replacement": config.get("replacement", False),
            "version": config.get("version", 1),  # For NearMiss
            "n_neighbors": config.get("n_neighbors", 3),  # For EditedNearestNeighbours
            "kind_sel": config.get("kind_sel", "all"),  # For EditedNearestNeighbours
            "n_jobs": config.get("n_jobs", -1),
        }

fit(_X, _y, config)

Snapshot the sampler settings from config into the artifact.

Neither _X nor _y is read: the artifact is pure configuration. Several of its keys are sampler-specific, so each method consumes only the subset its imblearn class accepts.

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

@fit_method
def fit(self, _X: Any, _y: Any, config: dict[str, Any]) -> UndersamplingArtifact:  # pylint: disable=arguments-differ
    """Snapshot the sampler settings from ``config`` into the artifact.

    Neither ``_X`` nor ``_y`` is read: the artifact is pure configuration.
    Several of its keys are sampler-specific, so each method consumes only
    the subset its imblearn class accepts.
    """
    return {
        "type": "undersampling",
        "method": config.get("method", "random_under_sampling"),
        "target_column": config.get("target_column"),
        "sampling_strategy": config.get("sampling_strategy", "auto"),
        "random_state": config.get("random_state", 42),
        "replacement": config.get("replacement", False),
        "version": config.get("version", 1),  # For NearMiss
        "n_neighbors": config.get("n_neighbors", 3),  # For EditedNearestNeighbours
        "kind_sel": config.get("kind_sel", "all"),  # For EditedNearestNeighbours
        "n_jobs": config.get("n_jobs", -1),
    }

infer_output_schema(input_schema, config)

Return input_schema untouched.

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

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Return ``input_schema`` untouched."""
    # Resampling changes row counts only; column set is preserved.
    return input_schema