API: preprocessing.feature_selection

skyulf.preprocessing.feature_selection

Feature-selection nodes package.

Importing this package registers all feature-selection nodes (variance, correlation, univariate, model-based, and the unified facade) and re-exports their public classes.

CorrelationThresholdApplier

Bases: BaseApplier

Remove the columns a fitted correlation threshold marked redundant.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/correlation.py

class CorrelationThresholdApplier(BaseApplier):
    """Remove the columns a fitted correlation threshold marked redundant."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved drop decisions without evaluating fit-only metadata.

        Fewer than two fitted candidates produce an empty no-op artifact.
        Threshold and method are unused at apply, including callable methods.
        """
        fields = {"type", "columns_to_drop", "threshold", "method", "drop_columns"}
        if not local_state_fields(raw, "correlation_threshold", fields, allow_empty=True):
            return raw
        _columns(raw["columns_to_drop"])
        _local_selection_drop_flag(raw["drop_columns"])
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe existing local saved apply without admitting partition workers."""
        if engine not in {"pandas", "polars"}:
            return None
        CorrelationThresholdApplier.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
        """Drop the artifact's precomputed ``columns_to_drop`` from ``X``.

        The list is resolved at fit time, so apply never recomputes a correlation
        matrix; ``drop_columns=False`` leaves the frame untouched.
        """
        return apply_dual_engine(
            X, params, {"polars": _corr_drop_polars, "pandas": _corr_drop_pandas}
        )

apply(X, _y, params)

Drop the artifact's precomputed columns_to_drop from X.

The list is resolved at fit time, so apply never recomputes a correlation matrix; drop_columns=False leaves the frame untouched.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/correlation.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Drop the artifact's precomputed ``columns_to_drop`` from ``X``.

    The list is resolved at fit time, so apply never recomputes a correlation
    matrix; ``drop_columns=False`` leaves the frame untouched.
    """
    return apply_dual_engine(
        X, params, {"polars": _corr_drop_polars, "pandas": _corr_drop_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe existing local saved apply without admitting partition workers.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/correlation.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe existing local saved apply without admitting partition workers."""
    if engine not in {"pandas", "polars"}:
        return None
    CorrelationThresholdApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect saved drop decisions without evaluating fit-only metadata.

Fewer than two fitted candidates produce an empty no-op artifact. Threshold and method are unused at apply, including callable methods.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/correlation.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved drop decisions without evaluating fit-only metadata.

    Fewer than two fitted candidates produce an empty no-op artifact.
    Threshold and method are unused at apply, including callable methods.
    """
    fields = {"type", "columns_to_drop", "threshold", "method", "drop_columns"}
    if not local_state_fields(raw, "correlation_threshold", fields, allow_empty=True):
        return raw
    _columns(raw["columns_to_drop"])
    _local_selection_drop_flag(raw["drop_columns"])
    return raw

CorrelationThresholdCalculator

Bases: BaseCalculator

Fit a pairwise-correlation threshold over numeric columns.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/correlation.py

@NodeRegistry.register("CorrelationThreshold", CorrelationThresholdApplier)
@node_meta(
    id="CorrelationThreshold",
    name="Correlation Threshold",
    category="Feature Selection",
    description="Remove features highly correlated with others.",
    params={"threshold": 0.95, "method": "pearson"},
    learns_from_data=True,
)
class CorrelationThresholdCalculator(BaseCalculator):
    """Fit a pairwise-correlation threshold over numeric columns."""

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> CorrelationThresholdArtifact:  # pylint: disable=arguments-differ
        """Return columns whose correlation with an earlier one exceeds the threshold.

        Within each over-threshold pair the later column is the one dropped. The
        native Polars path is used only where it can honour the same contract;
        Kendall and callable methods fall back to pandas. Returns an empty
        artifact — a no-op passthrough — when fewer than two columns resolve.
        """
        threshold = config.get("threshold", 0.95)
        drop_columns = config.get("drop_columns", True)
        # Prefer "correlation_method" — falling back to "method" can collide with the
        # facade's own "method" key (e.g. "correlation_threshold").
        method = config.get("correlation_method", "pearson")
        frame = _as_polars_frame(X)

        if frame is not None:
            columns = resolve_columns(SkyulfPolarsWrapper(frame), config, detect_numeric_columns)
            if len(columns) < 2:
                return cast(CorrelationThresholdArtifact, {})
            if _native_polars_correlation_eligible(frame, columns, method, threshold):
                native_method = cast(NativePolarsCorrelationMethod, method)
                return cast(
                    CorrelationThresholdArtifact,
                    {
                        "type": "correlation_threshold",
                        "columns_to_drop": _polars_correlation_columns_to_drop(
                            frame,
                            columns,
                            native_method,
                            threshold,
                        ),
                        "threshold": threshold,
                        "method": method,
                        "drop_columns": drop_columns,
                    },
                )

        # Retain this compatibility route until Polars supports Kendall and callable correlations.
        return _fit_correlation_threshold_pandas(X, config)

fit(X, _y, config)

Return columns whose correlation with an earlier one exceeds the threshold.

Within each over-threshold pair the later column is the one dropped. The native Polars path is used only where it can honour the same contract; Kendall and callable methods fall back to pandas. Returns an empty artifact — a no-op passthrough — when fewer than two columns resolve.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/correlation.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> CorrelationThresholdArtifact:  # pylint: disable=arguments-differ
    """Return columns whose correlation with an earlier one exceeds the threshold.

    Within each over-threshold pair the later column is the one dropped. The
    native Polars path is used only where it can honour the same contract;
    Kendall and callable methods fall back to pandas. Returns an empty
    artifact — a no-op passthrough — when fewer than two columns resolve.
    """
    threshold = config.get("threshold", 0.95)
    drop_columns = config.get("drop_columns", True)
    # Prefer "correlation_method" — falling back to "method" can collide with the
    # facade's own "method" key (e.g. "correlation_threshold").
    method = config.get("correlation_method", "pearson")
    frame = _as_polars_frame(X)

    if frame is not None:
        columns = resolve_columns(SkyulfPolarsWrapper(frame), config, detect_numeric_columns)
        if len(columns) < 2:
            return cast(CorrelationThresholdArtifact, {})
        if _native_polars_correlation_eligible(frame, columns, method, threshold):
            native_method = cast(NativePolarsCorrelationMethod, method)
            return cast(
                CorrelationThresholdArtifact,
                {
                    "type": "correlation_threshold",
                    "columns_to_drop": _polars_correlation_columns_to_drop(
                        frame,
                        columns,
                        native_method,
                        threshold,
                    ),
                    "threshold": threshold,
                    "method": method,
                    "drop_columns": drop_columns,
                },
            )

    # Retain this compatibility route until Polars supports Kendall and callable correlations.
    return _fit_correlation_threshold_pandas(X, config)

FeatureSelectionApplier

Bases: BaseApplier

Route a fitted selection artifact to the applier that produced it.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/facade.py

class FeatureSelectionApplier(BaseApplier):
    """Route a fitted selection artifact to the applier that produced it."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Delegate inspection to the same saved discriminator used by local apply."""
        if type(raw) is not dict:
            raise ValueError("Local fitted selection state must be a dictionary.")
        kind = raw.get("type")
        if kind is not None and not isinstance(kind, str):
            raise ValueError("Fitted selection type must be a string.")
        owner = _FS_APPLIERS.get(kind)
        if owner is not None:
            owner.validate_inference_state(raw)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Reuse the selected owner's context; unmatched saved types remain passthroughs."""
        if engine not in {"pandas", "polars"}:
            return None
        FeatureSelectionApplier.validate_inference_state(state)
        owner = _FS_APPLIERS.get(state.get("type"))
        if owner is not None:
            return owner.inference_capability(state, engine=engine)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    def apply(
        self,
        df: Any,
        params: dict[str, Any],
    ) -> Any:
        """Apply whichever concrete selection applier the artifact's ``type`` names.

        An unrecognized ``type`` — including the empty artifact an unknown method
        produces at fit time — is an identity passthrough, so this facade can
        never drop data on its own.
        """
        # The params returned by the specific calculator carry a "type" tag
        # that selects the right concrete applier.
        type_name = params.get("type")

        applier: BaseApplier | None = None
        if type_name == "variance_threshold":
            applier = VarianceThresholdApplier()
        elif type_name == "correlation_threshold":
            applier = CorrelationThresholdApplier()
        elif type_name == "univariate_selection":
            applier = UnivariateSelectionApplier()
        elif type_name == "model_based_selection":
            applier = ModelBasedSelectionApplier()

        if applier:
            return applier.apply(df, params)  # pylint: disable=no-value-for-parameter
        # Identity passthrough when no concrete applier matches.
        return df

apply(df, params)

Apply whichever concrete selection applier the artifact's type names.

An unrecognized type — including the empty artifact an unknown method produces at fit time — is an identity passthrough, so this facade can never drop data on its own.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/facade.py

def apply(
    self,
    df: Any,
    params: dict[str, Any],
) -> Any:
    """Apply whichever concrete selection applier the artifact's ``type`` names.

    An unrecognized ``type`` — including the empty artifact an unknown method
    produces at fit time — is an identity passthrough, so this facade can
    never drop data on its own.
    """
    # The params returned by the specific calculator carry a "type" tag
    # that selects the right concrete applier.
    type_name = params.get("type")

    applier: BaseApplier | None = None
    if type_name == "variance_threshold":
        applier = VarianceThresholdApplier()
    elif type_name == "correlation_threshold":
        applier = CorrelationThresholdApplier()
    elif type_name == "univariate_selection":
        applier = UnivariateSelectionApplier()
    elif type_name == "model_based_selection":
        applier = ModelBasedSelectionApplier()

    if applier:
        return applier.apply(df, params)  # pylint: disable=no-value-for-parameter
    # Identity passthrough when no concrete applier matches.
    return df

inference_capability(state, *, engine) staticmethod

Reuse the selected owner's context; unmatched saved types remain passthroughs.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/facade.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Reuse the selected owner's context; unmatched saved types remain passthroughs."""
    if engine not in {"pandas", "polars"}:
        return None
    FeatureSelectionApplier.validate_inference_state(state)
    owner = _FS_APPLIERS.get(state.get("type"))
    if owner is not None:
        return owner.inference_capability(state, engine=engine)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Delegate inspection to the same saved discriminator used by local apply.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/facade.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Delegate inspection to the same saved discriminator used by local apply."""
    if type(raw) is not dict:
        raise ValueError("Local fitted selection state must be a dictionary.")
    kind = raw.get("type")
    if kind is not None and not isinstance(kind, str):
        raise ValueError("Fitted selection type must be a string.")
    owner = _FS_APPLIERS.get(kind)
    if owner is not None:
        owner.validate_inference_state(raw)
    return raw

FeatureSelectionCalculator

Bases: BaseCalculator

Resolve config["method"] to one of the four concrete selection nodes.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/facade.py

@NodeRegistry.register("feature_selection", FeatureSelectionApplier)
@node_meta(
    id="feature_selection",
    name="Feature Selection (Wrapper)",
    category="Feature Selection",
    description="General wrapper for feature selection strategies.",
    params={"method": "variance", "threshold": 0.0},
    learns_from_data=True,
)
class FeatureSelectionCalculator(BaseCalculator):
    """Resolve ``config["method"]`` to one of the four concrete selection nodes."""

    def fit(
        self,
        df: Any,
        config: dict[str, Any],
    ) -> Mapping[str, Any]:
        """Fit the concrete calculator the method alias maps to and return its artifact.

        ``_FS_CALCULATORS`` collapses eleven aliases — node ids such as
        ``variance`` alongside sklearn-style names such as ``select_k_best`` and
        ``rfe`` — onto four calculators, so several spellings reach the same fit.
        An unknown method logs a warning and returns an empty artifact rather
        than raising, which degrades the node to a passthrough.
        """
        method = config.get("method", "select_k_best")
        ctor = _FS_CALCULATORS.get(method)
        if ctor is None:
            logger.warning(f"Unknown feature selection method: {method}")
            return {}
        return ctor().fit(df, config)

fit(df, config)

Fit the concrete calculator the method alias maps to and return its artifact.

_FS_CALCULATORS collapses eleven aliases — node ids such as variance alongside sklearn-style names such as select_k_best and rfe — onto four calculators, so several spellings reach the same fit. An unknown method logs a warning and returns an empty artifact rather than raising, which degrades the node to a passthrough.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/facade.py

def fit(
    self,
    df: Any,
    config: dict[str, Any],
) -> Mapping[str, Any]:
    """Fit the concrete calculator the method alias maps to and return its artifact.

    ``_FS_CALCULATORS`` collapses eleven aliases — node ids such as
    ``variance`` alongside sklearn-style names such as ``select_k_best`` and
    ``rfe`` — onto four calculators, so several spellings reach the same fit.
    An unknown method logs a warning and returns an empty artifact rather
    than raising, which degrades the node to a passthrough.
    """
    method = config.get("method", "select_k_best")
    ctor = _FS_CALCULATORS.get(method)
    if ctor is None:
        logger.warning(f"Unknown feature selection method: {method}")
        return {}
    return ctor().fit(df, config)

ModelBasedSelectionApplier

Bases: BaseApplier

Remove the columns an estimator's importances rejected.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/model_based.py

class ModelBasedSelectionApplier(BaseApplier):
    """Remove the columns an estimator's importances rejected."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved membership while leaving unused fit reports unchanged."""
        fields = {
            "type",
            "selected_columns",
            "candidate_columns",
            "method",
            "drop_columns",
            "feature_importances",
        }
        if local_state_fields(raw, "model_based_selection", fields, allow_empty=True):
            selected = _local_selection_columns(raw["selected_columns"])
            candidates = _local_selection_columns(raw["candidate_columns"])
            if not set(selected).issubset(candidates):
                raise ValueError("Fitted selected columns must belong to candidate columns.")
            _local_selection_drop_flag(raw["drop_columns"])
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe saved column selection without rerunning the training estimator."""
        if engine not in {"pandas", "polars"}:
            return None
        ModelBasedSelectionApplier.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
        """Drop ``candidate_columns - selected_columns`` from ``X``.

        Honours the artifact's ``drop_columns`` flag; the recorded
        ``feature_importances`` are reporting metadata and take no part in the
        transform.
        """
        return apply_dual_engine(
            X, params, {"polars": _drop_selected_polars, "pandas": _drop_selected_pandas}
        )

apply(X, _y, params)

Drop candidate_columns - selected_columns from X.

Honours the artifact's drop_columns flag; the recorded feature_importances are reporting metadata and take no part in the transform.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/model_based.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Drop ``candidate_columns - selected_columns`` from ``X``.

    Honours the artifact's ``drop_columns`` flag; the recorded
    ``feature_importances`` are reporting metadata and take no part in the
    transform.
    """
    return apply_dual_engine(
        X, params, {"polars": _drop_selected_polars, "pandas": _drop_selected_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe saved column selection without rerunning the training estimator.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/model_based.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe saved column selection without rerunning the training estimator."""
    if engine not in {"pandas", "polars"}:
        return None
    ModelBasedSelectionApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect saved membership while leaving unused fit reports unchanged.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/model_based.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved membership while leaving unused fit reports unchanged."""
    fields = {
        "type",
        "selected_columns",
        "candidate_columns",
        "method",
        "drop_columns",
        "feature_importances",
    }
    if local_state_fields(raw, "model_based_selection", fields, allow_empty=True):
        selected = _local_selection_columns(raw["selected_columns"])
        candidates = _local_selection_columns(raw["candidate_columns"])
        if not set(selected).issubset(candidates):
            raise ValueError("Fitted selected columns must belong to candidate columns.")
        _local_selection_drop_flag(raw["drop_columns"])
    return raw

ModelBasedSelectionCalculator

Bases: BaseCalculator

Fit a SelectFromModel or RFE selector around a resolved estimator.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/model_based.py

@NodeRegistry.register("ModelBasedSelection", ModelBasedSelectionApplier)
@node_meta(
    id="ModelBasedSelection",
    name="Model-Based Selection",
    category="Feature Selection",
    description="Select features based on importance weights.",
    params={"estimator": "RandomForest", "threshold": "mean", "max_features": None},
    learns_from_data=True,
)
class ModelBasedSelectionCalculator(BaseCalculator):
    """Fit a ``SelectFromModel`` or ``RFE`` selector around a resolved estimator."""

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> ModelBasedSelectionArtifact:  # pylint: disable=arguments-differ
        """Return the features the fitted estimator's support keeps.

        This selector is supervised: with no target there is nothing to score
        against, so a missing target is treated as an error. Every failure mode
        returns an empty artifact instead of raising, which degrades the node to
        a passthrough — though an unsupported ``method`` does so silently,
        without the log line the other branches emit.
        """
        target_col = config.get("target_column")
        # Resolve candidate columns natively on the raw frame first (works on
        # Polars without conversion), then convert only the columns actually
        # needed -- candidates plus the target column when it must be pulled
        # out of X itself (y is None) -- instead of the whole input frame.
        cols = _resolve_candidate_columns(X, config, target_col)
        if not cols:
            return cast(ModelBasedSelectionArtifact, {})

        needed_cols = cols if y is not None or not target_col else [*cols, target_col]
        X_pd = select_then_to_pandas(X, needed_cols)

        y = _extract_target(X_pd, y, target_col)
        if y is None:
            logger.error(
                f"ModelBasedSelection requires target column '{target_col}' "
                "to be present in training data."
            )
            return cast(ModelBasedSelectionArtifact, {})

        method = config.get("method", "select_from_model")
        estimator_name = config.get("estimator", "auto")
        problem_type = _resolve_problem_type(config.get("problem_type", "auto"), y)

        estimator = _resolve_estimator(estimator_name, problem_type)
        if estimator is None:
            logger.error(
                f"Could not resolve estimator '{estimator_name}' for problem type '{problem_type}'"
            )
            return cast(ModelBasedSelectionArtifact, {})

        selector = _build_model_selector(method, estimator, config)
        if selector is None:
            return cast(ModelBasedSelectionArtifact, {})

        X_np, _ = SklearnBridge.to_sklearn(_fillna_zero_with_warning(X_pd, cols))
        selector.fit(X_np, _prepare_sklearn_y(y, problem_type))
        support = selector.get_support()
        selected_cols = [c for c, s in zip(cols, support, strict=True) if s]

        return cast(
            ModelBasedSelectionArtifact,
            {
                "type": "model_based_selection",
                "selected_columns": selected_cols,
                "candidate_columns": cols,
                "method": method,
                "drop_columns": config.get("drop_columns", True),
                "feature_importances": _model_feature_importances(selector, cols),
            },
        )

fit(X, y, config)

Return the features the fitted estimator's support keeps.

This selector is supervised: with no target there is nothing to score against, so a missing target is treated as an error. Every failure mode returns an empty artifact instead of raising, which degrades the node to a passthrough — though an unsupported method does so silently, without the log line the other branches emit.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/model_based.py

@fit_method
def fit(self, X: Any, y: Any, config: dict[str, Any]) -> ModelBasedSelectionArtifact:  # pylint: disable=arguments-differ
    """Return the features the fitted estimator's support keeps.

    This selector is supervised: with no target there is nothing to score
    against, so a missing target is treated as an error. Every failure mode
    returns an empty artifact instead of raising, which degrades the node to
    a passthrough — though an unsupported ``method`` does so silently,
    without the log line the other branches emit.
    """
    target_col = config.get("target_column")
    # Resolve candidate columns natively on the raw frame first (works on
    # Polars without conversion), then convert only the columns actually
    # needed -- candidates plus the target column when it must be pulled
    # out of X itself (y is None) -- instead of the whole input frame.
    cols = _resolve_candidate_columns(X, config, target_col)
    if not cols:
        return cast(ModelBasedSelectionArtifact, {})

    needed_cols = cols if y is not None or not target_col else [*cols, target_col]
    X_pd = select_then_to_pandas(X, needed_cols)

    y = _extract_target(X_pd, y, target_col)
    if y is None:
        logger.error(
            f"ModelBasedSelection requires target column '{target_col}' "
            "to be present in training data."
        )
        return cast(ModelBasedSelectionArtifact, {})

    method = config.get("method", "select_from_model")
    estimator_name = config.get("estimator", "auto")
    problem_type = _resolve_problem_type(config.get("problem_type", "auto"), y)

    estimator = _resolve_estimator(estimator_name, problem_type)
    if estimator is None:
        logger.error(
            f"Could not resolve estimator '{estimator_name}' for problem type '{problem_type}'"
        )
        return cast(ModelBasedSelectionArtifact, {})

    selector = _build_model_selector(method, estimator, config)
    if selector is None:
        return cast(ModelBasedSelectionArtifact, {})

    X_np, _ = SklearnBridge.to_sklearn(_fillna_zero_with_warning(X_pd, cols))
    selector.fit(X_np, _prepare_sklearn_y(y, problem_type))
    support = selector.get_support()
    selected_cols = [c for c, s in zip(cols, support, strict=True) if s]

    return cast(
        ModelBasedSelectionArtifact,
        {
            "type": "model_based_selection",
            "selected_columns": selected_cols,
            "candidate_columns": cols,
            "method": method,
            "drop_columns": config.get("drop_columns", True),
            "feature_importances": _model_feature_importances(selector, cols),
        },
    )

UnivariateSelectionApplier

Bases: BaseApplier

Remove the columns a univariate statistical test rejected.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/univariate.py

class UnivariateSelectionApplier(BaseApplier):
    """Remove the columns a univariate statistical test rejected."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved selection membership while preserving both fitted schemas.

        A no-target fit uses ``scores``/``pvalues`` reporting fields; a scored
        fit uses ``feature_scores``/``p_values``. Reports and the fitted method
        are unused by apply and remain untouched, including nonfinite scores.
        """
        reports = (
            {"scores", "pvalues"}
            if type(raw) is dict and "scores" in raw
            else {"feature_scores", "p_values"}
        )
        fields = {"type", "selected_columns", "candidate_columns", "method", "drop_columns"}
        if not local_state_fields(raw, "univariate_selection", fields | reports, allow_empty=True):
            return raw
        selected = _local_selection_columns(raw["selected_columns"])
        candidates = _local_selection_columns(raw["candidate_columns"])
        if not set(selected).issubset(candidates):
            raise ValueError("Fitted selected columns must belong to candidate columns.")
        _local_selection_drop_flag(raw["drop_columns"])
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe existing local saved apply without admitting partition workers."""
        if engine not in {"pandas", "polars"}:
            return None
        UnivariateSelectionApplier.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
        """Drop ``candidate_columns - selected_columns`` from ``X``.

        Honours the artifact's ``drop_columns`` flag; the recorded
        ``feature_scores`` and ``p_values`` are reporting metadata only.
        """
        return apply_dual_engine(
            X, params, {"polars": _drop_selected_polars, "pandas": _drop_selected_pandas}
        )

apply(X, _y, params)

Drop candidate_columns - selected_columns from X.

Honours the artifact's drop_columns flag; the recorded feature_scores and p_values are reporting metadata only.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/univariate.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Drop ``candidate_columns - selected_columns`` from ``X``.

    Honours the artifact's ``drop_columns`` flag; the recorded
    ``feature_scores`` and ``p_values`` are reporting metadata only.
    """
    return apply_dual_engine(
        X, params, {"polars": _drop_selected_polars, "pandas": _drop_selected_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe existing local saved apply without admitting partition workers.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/univariate.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe existing local saved apply without admitting partition workers."""
    if engine not in {"pandas", "polars"}:
        return None
    UnivariateSelectionApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect saved selection membership while preserving both fitted schemas.

A no-target fit uses scores/pvalues reporting fields; a scored fit uses feature_scores/p_values. Reports and the fitted method are unused by apply and remain untouched, including nonfinite scores.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/univariate.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved selection membership while preserving both fitted schemas.

    A no-target fit uses ``scores``/``pvalues`` reporting fields; a scored
    fit uses ``feature_scores``/``p_values``. Reports and the fitted method
    are unused by apply and remain untouched, including nonfinite scores.
    """
    reports = (
        {"scores", "pvalues"}
        if type(raw) is dict and "scores" in raw
        else {"feature_scores", "p_values"}
    )
    fields = {"type", "selected_columns", "candidate_columns", "method", "drop_columns"}
    if not local_state_fields(raw, "univariate_selection", fields | reports, allow_empty=True):
        return raw
    selected = _local_selection_columns(raw["selected_columns"])
    candidates = _local_selection_columns(raw["candidate_columns"])
    if not set(selected).issubset(candidates):
        raise ValueError("Fitted selected columns must belong to candidate columns.")
    _local_selection_drop_flag(raw["drop_columns"])
    return raw

UnivariateSelectionCalculator

Bases: BaseCalculator

Fit a univariate statistical selector over numeric candidate columns.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/univariate.py

@NodeRegistry.register("UnivariateSelection", UnivariateSelectionApplier)
@node_meta(
    id="UnivariateSelection",
    name="Univariate Selection",
    category="Feature Selection",
    description="Select best features based on univariate statistical tests.",
    params={"method": "SelectKBest", "score_func": "Auto", "k": 10},
    learns_from_data=True,
)
class UnivariateSelectionCalculator(BaseCalculator):
    """Fit a univariate statistical selector over numeric candidate columns."""

    @fit_method
    def fit(self, X: Any, y: Any, config: dict[str, Any]) -> UnivariateSelectionArtifact:  # pylint: disable=arguments-differ
        """Return the features the score function's support keeps.

        A missing target is normally an error, but ``allow_missing_target``
        downgrades it to a keep-everything artifact so unsupervised pipelines
        can still run this node. The scoring inputs are adjusted first — missing
        values become 0, and chi2 features are MinMax-rescaled when any value is
        negative — because both shift the resulting scores.
        """
        target_col = config.get("target_column")
        # Resolve candidates natively on the raw frame first (Polars-safe),
        # then convert only the columns actually needed.
        cols = _resolve_candidate_columns(X, config, target_col)
        if not cols:
            return cast(UnivariateSelectionArtifact, {})

        needed_cols = cols if y is not None or not target_col else [*cols, target_col]
        X_pd = select_then_to_pandas(X, needed_cols)

        y = _extract_target(X_pd, y, target_col)
        if y is None and not config.get("allow_missing_target", False):
            logger.error(
                f"UnivariateSelection requires target column '{target_col}' "
                "to be present in training data."
            )
            return cast(UnivariateSelectionArtifact, {})

        method = config.get("method", "select_k_best")
        score_func_name = config.get("score_func")
        problem_type = _resolve_problem_type(config.get("problem_type", "auto"), y)
        score_func = _resolve_score_function(score_func_name, problem_type)

        selector = _build_univariate_selector(method, score_func, config)
        if selector is None:
            return cast(UnivariateSelectionArtifact, {})

        X_np, _ = SklearnBridge.to_sklearn(_fillna_zero_with_warning(X_pd, cols))
        X_np = _maybe_chi2_rescale(X_np, score_func_name)

        if y is None:
            return _univariate_no_target_artifact(cols, method, config)

        selector.fit(X_np, _univariate_target(y, problem_type))
        support = selector.get_support()
        selected_cols = [c for c, s in zip(cols, support, strict=True) if s]
        scores, pvalues = _univariate_score_dicts(selector, cols)

        return cast(
            UnivariateSelectionArtifact,
            {
                "type": "univariate_selection",
                "selected_columns": selected_cols,
                "candidate_columns": cols,
                "method": method,
                "drop_columns": config.get("drop_columns", True),
                "feature_scores": scores,
                "p_values": pvalues,
            },
        )

fit(X, y, config)

Return the features the score function's support keeps.

A missing target is normally an error, but allow_missing_target downgrades it to a keep-everything artifact so unsupervised pipelines can still run this node. The scoring inputs are adjusted first — missing values become 0, and chi2 features are MinMax-rescaled when any value is negative — because both shift the resulting scores.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/univariate.py

@fit_method
def fit(self, X: Any, y: Any, config: dict[str, Any]) -> UnivariateSelectionArtifact:  # pylint: disable=arguments-differ
    """Return the features the score function's support keeps.

    A missing target is normally an error, but ``allow_missing_target``
    downgrades it to a keep-everything artifact so unsupervised pipelines
    can still run this node. The scoring inputs are adjusted first — missing
    values become 0, and chi2 features are MinMax-rescaled when any value is
    negative — because both shift the resulting scores.
    """
    target_col = config.get("target_column")
    # Resolve candidates natively on the raw frame first (Polars-safe),
    # then convert only the columns actually needed.
    cols = _resolve_candidate_columns(X, config, target_col)
    if not cols:
        return cast(UnivariateSelectionArtifact, {})

    needed_cols = cols if y is not None or not target_col else [*cols, target_col]
    X_pd = select_then_to_pandas(X, needed_cols)

    y = _extract_target(X_pd, y, target_col)
    if y is None and not config.get("allow_missing_target", False):
        logger.error(
            f"UnivariateSelection requires target column '{target_col}' "
            "to be present in training data."
        )
        return cast(UnivariateSelectionArtifact, {})

    method = config.get("method", "select_k_best")
    score_func_name = config.get("score_func")
    problem_type = _resolve_problem_type(config.get("problem_type", "auto"), y)
    score_func = _resolve_score_function(score_func_name, problem_type)

    selector = _build_univariate_selector(method, score_func, config)
    if selector is None:
        return cast(UnivariateSelectionArtifact, {})

    X_np, _ = SklearnBridge.to_sklearn(_fillna_zero_with_warning(X_pd, cols))
    X_np = _maybe_chi2_rescale(X_np, score_func_name)

    if y is None:
        return _univariate_no_target_artifact(cols, method, config)

    selector.fit(X_np, _univariate_target(y, problem_type))
    support = selector.get_support()
    selected_cols = [c for c, s in zip(cols, support, strict=True) if s]
    scores, pvalues = _univariate_score_dicts(selector, cols)

    return cast(
        UnivariateSelectionArtifact,
        {
            "type": "univariate_selection",
            "selected_columns": selected_cols,
            "candidate_columns": cols,
            "method": method,
            "drop_columns": config.get("drop_columns", True),
            "feature_scores": scores,
            "p_values": pvalues,
        },
    )

VarianceThresholdApplier

Bases: BaseApplier

Remove the columns a fitted variance threshold rejected.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/variance.py

class VarianceThresholdApplier(BaseApplier):
    """Remove the columns a fitted variance threshold rejected."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved selection membership without recomputing variances.

        Empty fitted artifacts and selections are valid. The fit-only threshold
        and variances remain untouched, including undefined all-missing variance.
        """
        fields = {
            "type",
            "selected_columns",
            "candidate_columns",
            "threshold",
            "drop_columns",
            "variances",
        }
        if not local_state_fields(raw, "variance_threshold", fields, allow_empty=True):
            return raw
        selected = _local_selection_columns(raw["selected_columns"])
        candidates = _local_selection_columns(raw["candidate_columns"])
        if not set(selected).issubset(candidates):
            raise ValueError("Fitted selected columns must belong to candidate columns.")
        _local_selection_drop_flag(raw["drop_columns"])
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe existing local saved apply without admitting partition workers."""
        if engine not in {"pandas", "polars"}:
            return None
        VarianceThresholdApplier.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
        """Drop ``candidate_columns - selected_columns`` from ``X``.

        Honours the artifact's ``drop_columns`` flag; when it is false the frame
        passes through untouched.
        """
        return apply_dual_engine(
            X, params, {"polars": _drop_selected_polars, "pandas": _drop_selected_pandas}
        )

apply(X, _y, params)

Drop candidate_columns - selected_columns from X.

Honours the artifact's drop_columns flag; when it is false the frame passes through untouched.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/variance.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Drop ``candidate_columns - selected_columns`` from ``X``.

    Honours the artifact's ``drop_columns`` flag; when it is false the frame
    passes through untouched.
    """
    return apply_dual_engine(
        X, params, {"polars": _drop_selected_polars, "pandas": _drop_selected_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe existing local saved apply without admitting partition workers.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/variance.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe existing local saved apply without admitting partition workers."""
    if engine not in {"pandas", "polars"}:
        return None
    VarianceThresholdApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect saved selection membership without recomputing variances.

Empty fitted artifacts and selections are valid. The fit-only threshold and variances remain untouched, including undefined all-missing variance.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/variance.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved selection membership without recomputing variances.

    Empty fitted artifacts and selections are valid. The fit-only threshold
    and variances remain untouched, including undefined all-missing variance.
    """
    fields = {
        "type",
        "selected_columns",
        "candidate_columns",
        "threshold",
        "drop_columns",
        "variances",
    }
    if not local_state_fields(raw, "variance_threshold", fields, allow_empty=True):
        return raw
    selected = _local_selection_columns(raw["selected_columns"])
    candidates = _local_selection_columns(raw["candidate_columns"])
    if not set(selected).issubset(candidates):
        raise ValueError("Fitted selected columns must belong to candidate columns.")
    _local_selection_drop_flag(raw["drop_columns"])
    return raw

VarianceThresholdCalculator

Bases: BaseCalculator

Fit sklearn.feature_selection.VarianceThreshold on numeric columns.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/variance.py

@NodeRegistry.register("VarianceThreshold", VarianceThresholdApplier)
@node_meta(
    id="VarianceThreshold",
    name="Variance Threshold",
    category="Feature Selection",
    description="Remove features with low variance.",
    params={"threshold": 0.0},
    learns_from_data=True,
)
class VarianceThresholdCalculator(BaseCalculator):
    """Fit ``sklearn.feature_selection.VarianceThreshold`` on numeric columns."""

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> VarianceThresholdArtifact:  # pylint: disable=arguments-differ
        """Record which numeric columns clear ``config["threshold"]`` variance.

        Binary and constant columns are kept as candidates rather than excluded
        up front, so the threshold itself decides their fate. Returns an empty
        artifact — a no-op passthrough — when no numeric column is resolved.
        If no candidate clears the threshold, records an empty selection so
        apply can remove all candidates while preserving other columns.
        """
        threshold = config.get("threshold", 0.0)
        drop_columns = config.get("drop_columns", True)

        X_np, cols = resolve_columns_then_to_numpy(
            X,
            config,
            lambda d: detect_numeric_columns(d, exclude_binary=False, exclude_constant=False),
        )
        if not cols:
            return cast(VarianceThresholdArtifact, {})

        selector = VarianceThreshold(threshold=threshold)
        try:
            selector.fit(X_np)
        except ValueError as exc:
            # sklearn has computed variances but refuses an empty selection.
            if not str(exc).startswith("No feature in X meets the variance threshold"):
                raise
            all_missing = np.isnan(np.asarray(X_np, dtype=np.float64)).all(axis=0)
            if np.any(~np.isfinite(selector.variances_) & ~all_missing):
                # Only all-missing columns may have an undefined variance.
                raise
            selected_cols = []
        else:
            support = selector.get_support()
            selected_cols = [c for c, s in zip(cols, support, strict=True) if s]
        variances = (
            dict(zip(cols, selector.variances_.tolist(), strict=True))
            if hasattr(selector, "variances_")
            else {}
        )
        return cast(
            VarianceThresholdArtifact,
            {
                "type": "variance_threshold",
                "selected_columns": selected_cols,
                "candidate_columns": cols,
                "threshold": threshold,
                "drop_columns": drop_columns,
                "variances": variances,
            },
        )

fit(X, _y, config)

Record which numeric columns clear config["threshold"] variance.

Binary and constant columns are kept as candidates rather than excluded up front, so the threshold itself decides their fate. Returns an empty artifact — a no-op passthrough — when no numeric column is resolved. If no candidate clears the threshold, records an empty selection so apply can remove all candidates while preserving other columns.

Source code in skyulf-core/skyulf/preprocessing/feature_selection/variance.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> VarianceThresholdArtifact:  # pylint: disable=arguments-differ
    """Record which numeric columns clear ``config["threshold"]`` variance.

    Binary and constant columns are kept as candidates rather than excluded
    up front, so the threshold itself decides their fate. Returns an empty
    artifact — a no-op passthrough — when no numeric column is resolved.
    If no candidate clears the threshold, records an empty selection so
    apply can remove all candidates while preserving other columns.
    """
    threshold = config.get("threshold", 0.0)
    drop_columns = config.get("drop_columns", True)

    X_np, cols = resolve_columns_then_to_numpy(
        X,
        config,
        lambda d: detect_numeric_columns(d, exclude_binary=False, exclude_constant=False),
    )
    if not cols:
        return cast(VarianceThresholdArtifact, {})

    selector = VarianceThreshold(threshold=threshold)
    try:
        selector.fit(X_np)
    except ValueError as exc:
        # sklearn has computed variances but refuses an empty selection.
        if not str(exc).startswith("No feature in X meets the variance threshold"):
            raise
        all_missing = np.isnan(np.asarray(X_np, dtype=np.float64)).all(axis=0)
        if np.any(~np.isfinite(selector.variances_) & ~all_missing):
            # Only all-missing columns may have an undefined variance.
            raise
        selected_cols = []
    else:
        support = selector.get_support()
        selected_cols = [c for c, s in zip(cols, support, strict=True) if s]
    variances = (
        dict(zip(cols, selector.variances_.tolist(), strict=True))
        if hasattr(selector, "variances_")
        else {}
    )
    return cast(
        VarianceThresholdArtifact,
        {
            "type": "variance_threshold",
            "selected_columns": selected_cols,
            "candidate_columns": cols,
            "threshold": threshold,
            "drop_columns": drop_columns,
            "variances": variances,
        },
    )