API: preprocessing.feature_generation

skyulf.preprocessing.feature_generation

Feature-generation nodes package.

Importing this package registers the PolynomialFeatures, FeatureInteraction and FeatureGeneration nodes and re-exports their public classes plus shared constants for backward compatibility.

FeatureGenerationApplier

Bases: BaseApplier

Append the columns described by a feature-generation artifact.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/generation.py

class FeatureGenerationApplier(BaseApplier):
    """Append the columns described by a feature-generation artifact."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved operations, pinned similarity and learned group lookup shapes."""
        local_state_fields(
            raw, "feature_generation", {"type", "operations", "epsilon", "allow_overwrite"}
        )
        operations = raw["operations"]
        if type(operations) not in (list, tuple) or any(type(op) is not dict for op in operations):
            raise ValueError(
                "Fitted feature operations must be an ordered sequence of dictionaries."
            )
        if not isinstance(raw["allow_overwrite"], Decimal):
            local_scalar(raw["allow_overwrite"], "allow_overwrite")
        _validate_operation_types(operations)
        for op in operations:
            if op.get("operation_type") == "group_agg":
                _validate_group_mapping(op)
            if op.get("operation_type") == "similarity":
                _validate_similarity_backend(op.get("similarity_backend"))
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe saved feature replay without granting arbitrary partition execution."""
        if engine not in {"pandas", "polars"}:
            return None
        FeatureGenerationApplier.validate_inference_state(state)
        contexts = {_operation_context(op, state["epsilon"], engine) for op in state["operations"]}
        if None in contexts:
            return None
        context = "global" if "global" in contexts else "row"
        return ExecutionCapability(engine, "apply", "local", "preserve", context)

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Evaluate each configured operation and append its output column to ``X``.

        Operations run in order, so a later one can read a column an earlier one
        created. Unsupported operation types raise ``ValueError`` before any
        operation runs. A failure inside a supported operation is logged and
        skipped, and an output name that already exists is suffixed until
        unique unless ``allow_overwrite`` says otherwise.

        Datetime extraction uses ``output_column`` exactly for one configured
        source and feature. With multiple features it produces ``name_feature``;
        with multiple sources it produces ``name_source_feature``. Without an
        explicit name, outputs remain ``source_feature``, optionally preceded
        by ``output_prefix``. Every datetime output follows the collision policy.
        Saved pipelines affected by formerly ignored names or overwritten
        columns require refitting after upgrading.

        Group aggregates require training-fitted mappings. Legacy artifacts
        lacking those mappings must be refitted before inference.
        """
        _validate_operation_types(params.get("operations", []))
        _check_similarity_runtime(params.get("operations", []))
        if any(
            op.get("operation_type") == "group_agg" and "group_agg_mapping" not in op
            for op in params.get("operations", [])
        ):
            raise ValueError(
                "FeatureGeneration group_agg requires fitted statistics; refit the node."
            )
        return apply_dual_engine(
            X, params, {"polars": _featgen_apply_polars, "pandas": _featgen_apply_pandas}
        )

apply(X, _y, params)

Evaluate each configured operation and append its output column to X.

Operations run in order, so a later one can read a column an earlier one created. Unsupported operation types raise ValueError before any operation runs. A failure inside a supported operation is logged and skipped, and an output name that already exists is suffixed until unique unless allow_overwrite says otherwise.

Datetime extraction uses output_column exactly for one configured source and feature. With multiple features it produces name_feature; with multiple sources it produces name_source_feature. Without an explicit name, outputs remain source_feature, optionally preceded by output_prefix. Every datetime output follows the collision policy. Saved pipelines affected by formerly ignored names or overwritten columns require refitting after upgrading.

Group aggregates require training-fitted mappings. Legacy artifacts lacking those mappings must be refitted before inference.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/generation.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Evaluate each configured operation and append its output column to ``X``.

    Operations run in order, so a later one can read a column an earlier one
    created. Unsupported operation types raise ``ValueError`` before any
    operation runs. A failure inside a supported operation is logged and
    skipped, and an output name that already exists is suffixed until
    unique unless ``allow_overwrite`` says otherwise.

    Datetime extraction uses ``output_column`` exactly for one configured
    source and feature. With multiple features it produces ``name_feature``;
    with multiple sources it produces ``name_source_feature``. Without an
    explicit name, outputs remain ``source_feature``, optionally preceded
    by ``output_prefix``. Every datetime output follows the collision policy.
    Saved pipelines affected by formerly ignored names or overwritten
    columns require refitting after upgrading.

    Group aggregates require training-fitted mappings. Legacy artifacts
    lacking those mappings must be refitted before inference.
    """
    _validate_operation_types(params.get("operations", []))
    _check_similarity_runtime(params.get("operations", []))
    if any(
        op.get("operation_type") == "group_agg" and "group_agg_mapping" not in op
        for op in params.get("operations", [])
    ):
        raise ValueError(
            "FeatureGeneration group_agg requires fitted statistics; refit the node."
        )
    return apply_dual_engine(
        X, params, {"polars": _featgen_apply_polars, "pandas": _featgen_apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe saved feature replay without granting arbitrary partition execution.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/generation.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe saved feature replay without granting arbitrary partition execution."""
    if engine not in {"pandas", "polars"}:
        return None
    FeatureGenerationApplier.validate_inference_state(state)
    contexts = {_operation_context(op, state["epsilon"], engine) for op in state["operations"]}
    if None in contexts:
        return None
    context = "global" if "global" in contexts else "row"
    return ExecutionCapability(engine, "apply", "local", "preserve", context)

validate_inference_state(raw) staticmethod

Inspect saved operations, pinned similarity and learned group lookup shapes.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/generation.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved operations, pinned similarity and learned group lookup shapes."""
    local_state_fields(
        raw, "feature_generation", {"type", "operations", "epsilon", "allow_overwrite"}
    )
    operations = raw["operations"]
    if type(operations) not in (list, tuple) or any(type(op) is not dict for op in operations):
        raise ValueError(
            "Fitted feature operations must be an ordered sequence of dictionaries."
        )
    if not isinstance(raw["allow_overwrite"], Decimal):
        local_scalar(raw["allow_overwrite"], "allow_overwrite")
    _validate_operation_types(operations)
    for op in operations:
        if op.get("operation_type") == "group_agg":
            _validate_group_mapping(op)
        if op.get("operation_type") == "similarity":
            _validate_similarity_backend(op.get("similarity_backend"))
    return raw

FeatureGenerationCalculator

Bases: BaseCalculator

Record row-local operations and fit group aggregates from training rows.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/generation.py

@NodeRegistry.register("FeatureGeneration", FeatureGenerationApplier)
@NodeRegistry.register("FeatureMath", FeatureGenerationApplier)
@NodeRegistry.register("FeatureGenerationNode", FeatureGenerationApplier)
@node_meta(
    id="FeatureGenerationNode",
    name="Feature Generation (Math)",
    category="Feature Engineering",
    description="Generate new features using mathematical operations.",
    params={"operations": []},
    learns_from_data=True,
)
class FeatureGenerationCalculator(BaseCalculator):
    """Record row-local operations and fit group aggregates from training rows."""

    @fit_method
    def fit(
        self,
        X: Any,
        _y: Any,
        config: dict[str, Any],
    ) -> FeatureGenerationArtifact:  # pylint: disable=arguments-differ
        """Fit aggregate mappings against the intermediate training feature frame.

        Operations execute in order while fitting so an aggregate can consume
        an earlier generated column. Its saved mapping supplies held-out rows;
        unknown keys stay missing, and no inference values are aggregated.
        Configurations containing only row-local operations need no data fit.
        Unsupported operation types raise ``ValueError``; polynomial expansion
        belongs in the separate ``PolynomialFeatures`` node.
        """
        _validate_operation_types(config.get("operations", []))
        params: FeatureGenerationArtifact = {
            "type": "feature_generation",
            "operations": deepcopy(config.get("operations", [])),
            "epsilon": config.get("epsilon", DEFAULT_EPSILON),
            "allow_overwrite": config.get("allow_overwrite", False),
        }
        for op in params["operations"]:
            if op.get("operation_type") == "similarity":
                op["similarity_backend"] = _similarity_backend()
        if not any(op.get("operation_type") == "group_agg" for op in params["operations"]):
            return params

        working = X
        for index, op in enumerate(params["operations"]):
            if op.get("operation_type") == "group_agg":
                op["group_agg_mapping"] = _fit_group_aggregation(working, op)
            # apply_method exposes the public (data, params) call signature.
            working = FeatureGenerationApplier().apply(  # pylint: disable=no-value-for-parameter
                working, {**params, "operations": [op], "_operation_offset": index}
            )
        return params

fit(X, _y, config)

Fit aggregate mappings against the intermediate training feature frame.

Operations execute in order while fitting so an aggregate can consume an earlier generated column. Its saved mapping supplies held-out rows; unknown keys stay missing, and no inference values are aggregated. Configurations containing only row-local operations need no data fit. Unsupported operation types raise ValueError; polynomial expansion belongs in the separate PolynomialFeatures node.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/generation.py

@fit_method
def fit(
    self,
    X: Any,
    _y: Any,
    config: dict[str, Any],
) -> FeatureGenerationArtifact:  # pylint: disable=arguments-differ
    """Fit aggregate mappings against the intermediate training feature frame.

    Operations execute in order while fitting so an aggregate can consume
    an earlier generated column. Its saved mapping supplies held-out rows;
    unknown keys stay missing, and no inference values are aggregated.
    Configurations containing only row-local operations need no data fit.
    Unsupported operation types raise ``ValueError``; polynomial expansion
    belongs in the separate ``PolynomialFeatures`` node.
    """
    _validate_operation_types(config.get("operations", []))
    params: FeatureGenerationArtifact = {
        "type": "feature_generation",
        "operations": deepcopy(config.get("operations", [])),
        "epsilon": config.get("epsilon", DEFAULT_EPSILON),
        "allow_overwrite": config.get("allow_overwrite", False),
    }
    for op in params["operations"]:
        if op.get("operation_type") == "similarity":
            op["similarity_backend"] = _similarity_backend()
    if not any(op.get("operation_type") == "group_agg" for op in params["operations"]):
        return params

    working = X
    for index, op in enumerate(params["operations"]):
        if op.get("operation_type") == "group_agg":
            op["group_agg_mapping"] = _fit_group_aggregation(working, op)
        # apply_method exposes the public (data, params) call signature.
        working = FeatureGenerationApplier().apply(  # pylint: disable=no-value-for-parameter
            working, {**params, "operations": [op], "_operation_offset": index}
        )
    return params

FeatureInteractionApplier

Bases: BaseApplier

Append the multiplicative interaction columns named by the artifact.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/interaction.py

class FeatureInteractionApplier(BaseApplier):
    """Append the multiplicative interaction columns named by the artifact."""

    @staticmethod
    def validate_fitted_state(raw: dict) -> dict:
        """Inspect this node's supported saved state without fitting or applying data."""
        return interaction_state(raw)

    @staticmethod
    def resolve_fitted_config(raw: dict, state: dict) -> dict:
        """Bind inference configuration to this node's inspected fitted artifact."""
        return interaction_config(raw, state)

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Compute each artifact combination as a column product and append it.

        A combination whose columns are absent from ``X`` is skipped rather than
        raising, so the node survives an upstream column drop; ``include_bias``
        adds a constant 1.0 ``interaction_bias`` column per existing row,
        including zero rows when no input columns remain. When no column is
        generated the frame is returned unchanged. Generated product names must
        be unique and must not collide with existing columns. An existing bias
        column retains its established passthrough behavior.
        """
        combos = [
            tuple(combo)
            for combo in params.get("combinations", [])
            if all(col in X.columns for col in combo)
        ]
        names = _build_interaction_feature_names(
            combos, params.get("include_bias", False) and _BIAS_COLUMN not in X.columns
        )
        _validate_generated_names(names, list(X.columns), "FeatureInteraction")
        return apply_dual_engine(
            X, params, {"polars": _interaction_apply_polars, "pandas": _interaction_apply_pandas}
        )

apply(X, _y, params)

Compute each artifact combination as a column product and append it.

A combination whose columns are absent from X is skipped rather than raising, so the node survives an upstream column drop; include_bias adds a constant 1.0 interaction_bias column per existing row, including zero rows when no input columns remain. When no column is generated the frame is returned unchanged. Generated product names must be unique and must not collide with existing columns. An existing bias column retains its established passthrough behavior.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/interaction.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Compute each artifact combination as a column product and append it.

    A combination whose columns are absent from ``X`` is skipped rather than
    raising, so the node survives an upstream column drop; ``include_bias``
    adds a constant 1.0 ``interaction_bias`` column per existing row,
    including zero rows when no input columns remain. When no column is
    generated the frame is returned unchanged. Generated product names must
    be unique and must not collide with existing columns. An existing bias
    column retains its established passthrough behavior.
    """
    combos = [
        tuple(combo)
        for combo in params.get("combinations", [])
        if all(col in X.columns for col in combo)
    ]
    names = _build_interaction_feature_names(
        combos, params.get("include_bias", False) and _BIAS_COLUMN not in X.columns
    )
    _validate_generated_names(names, list(X.columns), "FeatureInteraction")
    return apply_dual_engine(
        X, params, {"polars": _interaction_apply_polars, "pandas": _interaction_apply_pandas}
    )

resolve_fitted_config(raw, state) staticmethod

Bind inference configuration to this node's inspected fitted artifact.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/interaction.py

@staticmethod
def resolve_fitted_config(raw: dict, state: dict) -> dict:
    """Bind inference configuration to this node's inspected fitted artifact."""
    return interaction_config(raw, state)

validate_fitted_state(raw) staticmethod

Inspect this node's supported saved state without fitting or applying data.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/interaction.py

@staticmethod
def validate_fitted_state(raw: dict) -> dict:
    """Inspect this node's supported saved state without fitting or applying data."""
    return interaction_state(raw)

FeatureInteractionCalculator

Bases: BaseCalculator

Resolve which cross-products to generate, without computing them.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/interaction.py

@NodeRegistry.register(
    "FeatureInteraction",
    FeatureInteractionApplier,
    execution_capabilities=(
        ExecutionCapability("pandas", "apply", "python_batch", "preserve", "row"),
        ExecutionCapability("polars", "apply", "local", "preserve", "row"),
    ),
)
@node_meta(
    id="FeatureInteraction",
    name="Feature Interaction",
    category="Feature Engineering",
    description=(
        "Generate 2-way/3-way/4-way multiplicative interaction features between "
        "numeric columns, using deterministic regularization-friendly names."
    ),
    params={"columns": [], **_DEFAULT_OPTIONS},
    learns_from_data=False,
)
class FeatureInteractionCalculator(BaseCalculator):
    """Resolve which cross-products to generate, without computing them."""

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> FeatureInteractionArtifact:  # pylint: disable=arguments-differ
        """Validate the requested columns and degree, then enumerate the combinations.

        Columns are sorted into the artifact so the generated names depend only
        on the set of inputs, never on the order they were configured in.
        With ``interaction_only=True``, fewer columns than ``degree`` yields
        no products. With ``False``, one column is sufficient for self-products.
        An empty selection yields no products; bias generation is independent.

        Raises:
            ValueError: If a configured column is missing or non-numeric, or if
                ``degree`` falls outside the supported ``(2, 3, 4)``. Generated
                product names must also be unique and absent from the input.
        """
        cols = list(config.get("columns", []))
        X_pd = select_then_to_pandas(X, cols)

        _validate_interaction_columns(X_pd, cols)

        degree = config.get("degree", _DEFAULT_OPTIONS["degree"])
        _validate_interaction_degree(degree)
        interaction_only = config.get("interaction_only", _DEFAULT_OPTIONS["interaction_only"])
        include_bias = config.get("include_bias", _DEFAULT_OPTIONS["include_bias"])

        combos = _resolve_combinations(cols, degree, interaction_only)
        feature_names = _build_interaction_feature_names(combos, include_bias)
        generated = [
            name for name in feature_names if name != _BIAS_COLUMN or name not in X.columns
        ]
        _validate_generated_names(generated, list(X.columns), "FeatureInteraction")

        return cast(
            FeatureInteractionArtifact,
            {
                "type": "feature_interaction",
                "columns": sorted(cols),
                "degree": degree,
                "interaction_only": interaction_only,
                "include_bias": include_bias,
                "combinations": [list(c) for c in combos],
                "feature_names": feature_names,
            },
        )

fit(X, _y, config)

Validate the requested columns and degree, then enumerate the combinations.

Columns are sorted into the artifact so the generated names depend only on the set of inputs, never on the order they were configured in. With interaction_only=True, fewer columns than degree yields no products. With False, one column is sufficient for self-products. An empty selection yields no products; bias generation is independent.

Raises:

Type	Description
ValueError	If a configured column is missing or non-numeric, or if degree falls outside the supported (2, 3, 4). Generated product names must also be unique and absent from the input.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/interaction.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> FeatureInteractionArtifact:  # pylint: disable=arguments-differ
    """Validate the requested columns and degree, then enumerate the combinations.

    Columns are sorted into the artifact so the generated names depend only
    on the set of inputs, never on the order they were configured in.
    With ``interaction_only=True``, fewer columns than ``degree`` yields
    no products. With ``False``, one column is sufficient for self-products.
    An empty selection yields no products; bias generation is independent.

    Raises:
        ValueError: If a configured column is missing or non-numeric, or if
            ``degree`` falls outside the supported ``(2, 3, 4)``. Generated
            product names must also be unique and absent from the input.
    """
    cols = list(config.get("columns", []))
    X_pd = select_then_to_pandas(X, cols)

    _validate_interaction_columns(X_pd, cols)

    degree = config.get("degree", _DEFAULT_OPTIONS["degree"])
    _validate_interaction_degree(degree)
    interaction_only = config.get("interaction_only", _DEFAULT_OPTIONS["interaction_only"])
    include_bias = config.get("include_bias", _DEFAULT_OPTIONS["include_bias"])

    combos = _resolve_combinations(cols, degree, interaction_only)
    feature_names = _build_interaction_feature_names(combos, include_bias)
    generated = [
        name for name in feature_names if name != _BIAS_COLUMN or name not in X.columns
    ]
    _validate_generated_names(generated, list(X.columns), "FeatureInteraction")

    return cast(
        FeatureInteractionArtifact,
        {
            "type": "feature_interaction",
            "columns": sorted(cols),
            "degree": degree,
            "interaction_only": interaction_only,
            "include_bias": include_bias,
            "combinations": [list(c) for c in combos],
            "feature_names": feature_names,
        },
    )

PolynomialFeaturesApplier

Bases: BaseApplier

Append the polynomial expansion described by the artifact.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/polynomial.py

class PolynomialFeaturesApplier(BaseApplier):
    """Append the polynomial expansion described by the artifact."""

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved expansion configuration without rebuilding or fitting its terms."""
        if not local_state_fields(
            raw,
            "polynomial_features",
            {
                "type",
                "columns",
                "degree",
                "interaction_only",
                "include_bias",
                "include_input_features",
                "output_prefix",
                "feature_names",
            },
            allow_empty=True,
        ):
            return raw
        _validate_polynomial_names(raw["columns"])
        if len(set(raw["columns"])) != len(raw["columns"]):
            raise ValueError("Fitted polynomial columns must be unique.")
        for flag in ("interaction_only", "include_bias", "include_input_features"):
            local_boolean(raw[flag], flag)
        _validate_polynomial_degree(raw["degree"], raw["include_bias"])
        if not isinstance(raw["output_prefix"], str):
            raise ValueError("Fitted polynomial prefix must be a string.")
        _validate_polynomial_names(raw["feature_names"])
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe local polynomial arithmetic without granting worker execution."""
        if engine not in ("pandas", "polars"):
            return None
        PolynomialFeaturesApplier.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
        """Recompute the expansion from ``X`` and append it as new columns.

        ``PolynomialFeatures`` is refit here rather than persisted, so the
        artifact only carries configuration. Artifact columns absent from ``X``
        are filtered out first, and the default ``include_input_features=False``
        drops the degree-1 terms so the originals are not duplicated. ``X`` is
        returned unchanged when nothing survives. Emitted names must be unique
        and must not collide with existing columns.
        """
        return apply_dual_engine(
            X, params, {"polars": _polynomial_apply_polars, "pandas": _polynomial_apply_pandas}
        )

apply(X, _y, params)

Recompute the expansion from X and append it as new columns.

PolynomialFeatures is refit here rather than persisted, so the artifact only carries configuration. Artifact columns absent from X are filtered out first, and the default include_input_features=False drops the degree-1 terms so the originals are not duplicated. X is returned unchanged when nothing survives. Emitted names must be unique and must not collide with existing columns.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/polynomial.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Recompute the expansion from ``X`` and append it as new columns.

    ``PolynomialFeatures`` is refit here rather than persisted, so the
    artifact only carries configuration. Artifact columns absent from ``X``
    are filtered out first, and the default ``include_input_features=False``
    drops the degree-1 terms so the originals are not duplicated. ``X`` is
    returned unchanged when nothing survives. Emitted names must be unique
    and must not collide with existing columns.
    """
    return apply_dual_engine(
        X, params, {"polars": _polynomial_apply_polars, "pandas": _polynomial_apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe local polynomial arithmetic without granting worker execution.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/polynomial.py

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

validate_inference_state(raw) staticmethod

Inspect saved expansion configuration without rebuilding or fitting its terms.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/polynomial.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved expansion configuration without rebuilding or fitting its terms."""
    if not local_state_fields(
        raw,
        "polynomial_features",
        {
            "type",
            "columns",
            "degree",
            "interaction_only",
            "include_bias",
            "include_input_features",
            "output_prefix",
            "feature_names",
        },
        allow_empty=True,
    ):
        return raw
    _validate_polynomial_names(raw["columns"])
    if len(set(raw["columns"])) != len(raw["columns"]):
        raise ValueError("Fitted polynomial columns must be unique.")
    for flag in ("interaction_only", "include_bias", "include_input_features"):
        local_boolean(raw[flag], flag)
    _validate_polynomial_degree(raw["degree"], raw["include_bias"])
    if not isinstance(raw["output_prefix"], str):
        raise ValueError("Fitted polynomial prefix must be a string.")
    _validate_polynomial_names(raw["feature_names"])
    return raw

PolynomialFeaturesCalculator

Bases: BaseCalculator

Resolve the polynomial configuration and the names it will emit.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/polynomial.py

@NodeRegistry.register("PolynomialFeatures", PolynomialFeaturesApplier)
@NodeRegistry.register("PolynomialFeaturesNode", PolynomialFeaturesApplier)
@node_meta(
    id="PolynomialFeatures",
    name="Polynomial Features",
    category="Feature Engineering",
    description="Generate polynomial and interaction features.",
    params={"degree": 2, "interaction_only": False, "include_bias": False},
    # Automatic column discovery inspects values; fixed selections are exempted.
    learns_from_data=True,
)
class PolynomialFeaturesCalculator(BaseCalculator):
    """Resolve the polynomial configuration and the names it will emit."""

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> PolynomialFeaturesArtifact:  # pylint: disable=arguments-differ
        """Record the expansion settings and the feature names they produce.

        ``columns`` falls back to auto-detected numeric columns when
        ``auto_detect`` is set. ``PolynomialFeatures`` is fitted here purely to
        derive ``feature_names`` — the expansion itself is recomputed at apply
        time. Returns an empty artifact, a no-op passthrough, when no column
        resolves. Ambiguous generated names raise ``ValueError`` before replay.
        """
        cols = list(config.get("columns", []))
        if not cols and config.get("auto_detect", False):
            # detect_numeric_columns dispatches natively on Polars frames too,
            # so this doesn't require converting the full frame first.
            cols = detect_numeric_columns(X)
        X_np, cols = select_then_to_numpy(X, cols)
        if not cols:
            return cast(PolynomialFeaturesArtifact, {})

        degree = config.get("degree", 2)
        interaction_only = config.get("interaction_only", False)
        include_bias = config.get("include_bias", False)

        poly = PolynomialFeatures(
            degree=degree, interaction_only=interaction_only, include_bias=include_bias
        )
        poly.fit(X_np)
        _, names = _polynomial_names(poly, cols, config)
        _validate_generated_names(names, list(X.columns), "PolynomialFeatures")
        return cast(
            PolynomialFeaturesArtifact,
            {
                "type": "polynomial_features",
                "columns": cols,
                "degree": degree,
                "interaction_only": interaction_only,
                "include_bias": include_bias,
                "include_input_features": config.get("include_input_features", False),
                "output_prefix": config.get("output_prefix", "poly"),
                "feature_names": poly.get_feature_names_out(cols).tolist(),
            },
        )

fit(X, _y, config)

Record the expansion settings and the feature names they produce.

columns falls back to auto-detected numeric columns when auto_detect is set. PolynomialFeatures is fitted here purely to derive feature_names — the expansion itself is recomputed at apply time. Returns an empty artifact, a no-op passthrough, when no column resolves. Ambiguous generated names raise ValueError before replay.

Source code in skyulf-core/skyulf/preprocessing/feature_generation/polynomial.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> PolynomialFeaturesArtifact:  # pylint: disable=arguments-differ
    """Record the expansion settings and the feature names they produce.

    ``columns`` falls back to auto-detected numeric columns when
    ``auto_detect`` is set. ``PolynomialFeatures`` is fitted here purely to
    derive ``feature_names`` — the expansion itself is recomputed at apply
    time. Returns an empty artifact, a no-op passthrough, when no column
    resolves. Ambiguous generated names raise ``ValueError`` before replay.
    """
    cols = list(config.get("columns", []))
    if not cols and config.get("auto_detect", False):
        # detect_numeric_columns dispatches natively on Polars frames too,
        # so this doesn't require converting the full frame first.
        cols = detect_numeric_columns(X)
    X_np, cols = select_then_to_numpy(X, cols)
    if not cols:
        return cast(PolynomialFeaturesArtifact, {})

    degree = config.get("degree", 2)
    interaction_only = config.get("interaction_only", False)
    include_bias = config.get("include_bias", False)

    poly = PolynomialFeatures(
        degree=degree, interaction_only=interaction_only, include_bias=include_bias
    )
    poly.fit(X_np)
    _, names = _polynomial_names(poly, cols, config)
    _validate_generated_names(names, list(X.columns), "PolynomialFeatures")
    return cast(
        PolynomialFeaturesArtifact,
        {
            "type": "polynomial_features",
            "columns": cols,
            "degree": degree,
            "interaction_only": interaction_only,
            "include_bias": include_bias,
            "include_input_features": config.get("include_input_features", False),
            "output_prefix": config.get("output_prefix", "poly"),
            "feature_names": poly.get_feature_names_out(cols).tolist(),
        },
    )