Pipeline

Polars Native Machine Learning Pipeline

Modules:

Name	Description
pipeline	Machine Learning / Time series Pipelines with native Polars support.
transforms	This module provides classic ML dataset transforms. Note all functions here are single-use only, meaning

pipeline

Machine Learning / Time series Pipelines with native Polars support.

Classes:

Name	Description
Blueprint	Blueprints for a ML/data transformation pipeline. In other words, this is a description of
ExprStep	Container for either one of these polars transforms:
PLContext
Pipeline	A ML/data transform pipeline. Pipelines should always come from the materialize call from a

Blueprint

Blueprints for a ML/data transformation pipeline. In other words, this is a description of what a pipeline will be. No learning/fitting is done until self.materialize() is called.

If the input df is lazy, the pipeline will collect at the time of fit.

Note: although polars selectors work for most transformations and in most cases, it is still recommended that the user should use explicit expressions instead of selectors for most transformations.

Methods:

Name	Description
__init__	Creates a blueprint object.
append_fit_func	Adds a custom transform that requires a fit step in the blueprint.
append_step_from_dict	Append a step to the blueprint by taking in a dictionary with keys name, args, and kwargs, where
cast_bools	Cast all boolean columns in the dataframe to the given type.
center	Centers the columns by subtracting each with its mean.
conditional_impute	Conditionally imputes values in the given columns. This transform will collect if input is lazy.
drop	Drops the columns from the dataset.
explode	Transform that represents df.explode(columns)
filter	Filters on the dataframe using native polars expressions or SQL boolean expressions.
fit	Alias for self.materialize()
group_by_agg	Performs a group by and agg on the data.
group_by_dynamic_agg	See polars group_by_dynamic documentation for an explanation on the input arguments.
impute	Imputes null values in the given columns. Note: this doesn't fill NaN. If filling for NaN is needed,
int_to_float	Maps all integer columns to float.
iv_encode	Use Information Value to encode a discrete variable x with respect to target. This assumes x
linear_impute	Imputes the target column by training a simple linear regression using the other features. This will
materialize	Materialize the blueprint, which means that it will fit and learn all the paramters needed.
nan_to_null	Maps NaN values in all columns to null.
one_hot_encode	Find the unique values in the string/categorical columns and one-hot encode them. This will NOT
ordinal_encode	Find the unique values in the string/categorical columns and one-hot encode them. This will NOT
polynomial_features	Generates polynomial combinations out of the features given, at the given degree.
rank_hot_encode	Given a ranking, e.g. ["bad", "neutral", "good"], where "bad", "neutral" and "good" are values coming
rename	Renames the columns by the mapping.
robust_scale	Performs robust scaling on the given columns
scale	Scales values in the given columns.
select	Selects the columns from the dataset.
select_by_std	Fits and keeps columns that have standard deviation between min_ and max_.
sort	Sorts the dataframe by the columns.
sql_transform	Runs the SQL on the dataframe when it reaches this step. The user must ensure that
target_encode	Target encode the given variables.
transform	Fits the blueprint with the dataframe that it is initialized with, and
winsorize	Learns the lower and upper percentile from the columns, then clip each end at those values.
with_columns	Run Polars with_columns for the expressions.
woe_encode	Use Weight of Evidence to encode a discrete variable x with respect to target. This assumes x

Source code in python/polars_ds/pipeline/pipeline.py

class Blueprint:
    """
    Blueprints for a ML/data transformation pipeline. In other words, this is a description of
    what a pipeline will be. No learning/fitting is done until self.materialize() is called.

    If the input df is lazy, the pipeline will collect at the time of fit.

    Note: although polars selectors work for most transformations and in most cases, it is still
    recommended that the user should use explicit expressions instead of selectors for most transformations.
    """

    def __init__(
        self,
        df: PolarsFrame,
        name: str = "test",
        target: str | None = None,
        exclude: List[str] | None = None,
        lowercase: bool = False,
        uppercase: bool = False,
    ):
        """
        Creates a blueprint object.

        Parameters
        ----------
        df
            Either a lazy or an eager Polars dataframe
        name
            Name of the blueprint.
        target
            Optionally indicate the target column in the ML pipeline. This will automatically prevent any transformation
            from changing the target column. (To be implemented: this should also automatically fill any transformation
            that requires a target name)
        exclude
            Any other column to exclude from global transformation. Note: this is only needed if you are not specifiying
            the exact columns to transform. E.g. when you are using a selector like cs.numeric() for all numeric columns.
            If this is the case and target is not set nor excluded, then the transformation may be applied to the target
            as well, which is not desired in most cases. Therefore, it is highly recommended you initialize with target name.
        lowercase
            Whether to insert a lowercase column name step before all other transformations.
            This takes precedence over uppercase.
        uppercase
            Whether to insert a uppercase column name step before all other transformations.
            This only happens if lowercase is False
        """

        self._df: pl.LazyFrame = df.lazy()
        if lowercase:
            self._df = self._df.select(pl.all().name.to_lowercase())
        else:
            if uppercase:
                self._df = self._df.select(pl.all().name.to_uppercase())

        self.name: str = str(name)
        self.target = target
        self.feature_names_in_: list[str] = self._df.collect_schema().names()

        self._steps: List[ExprStep | FitStep] = []
        self.exclude: List[str] = [] if target is None else [target]
        if exclude is not None:  # dedup in case user accidentally puts the same column name twice
            self.exclude = list(set(self.exclude + exclude))

        self.lowercase = lowercase
        self.uppercase = uppercase

    def __str__(self) -> str:
        out: str = ""
        out += f"Blueprint name: {self.name}\n"
        if self.lowercase:
            out += "Column names: Lowercase all incoming columns.\n"
        elif self.uppercase:
            out += "Column names: Uppercase all incoming columns.\n"

        out += f"Blueprint current steps: {len(self._steps)}\n"
        out += f"Features Expected: {self.feature_names_in_}\n"
        return out

    def _get_target(self, target: str | pl.Expr | None = None) -> str | pl.Expr:
        if target is None:
            if self.target is None:
                raise ValueError(
                    "Target is not given and blueprint is not initialized with a target."
                )
            return self.target
        else:
            return target

    def filter(self, by: str | pl.Expr) -> Self:
        """
        Filters on the dataframe using native polars expressions or SQL boolean expressions.

        Parameters
        ----------
        by
            Native polars boolean expression or SQL strings
        """
        self._steps.append(
            ExprStep(by if isinstance(by, pl.Expr) else pl.sql_expr(by), PLContext.FILTER)
        )
        return self

    def sql_transform(self, sql: str) -> Self:
        """
        Runs the SQL on the dataframe when it reaches this step. The user must ensure that
        the SQL is valid Polars SQL and all columns referred in the SQL exist at this point.
        The name "df" should be used to refer to the current state of the dataframe in the SQL.
        E.g. select * from df where A is True.

        Parameters
        ----------
        sql
            The SQL to run on the dataframe. Note: this step doesn't immedinately check the validity of
            the SQL statement.
        """
        self._steps.append(SQLStep(sql_str=sql))
        return self

    def cast_bools(self, dtype: pl.DataType = pl.UInt8) -> Self:
        """
        Cast all boolean columns in the dataframe to the given type.
        """
        self._steps.append(ExprStep(cs.boolean().cast(dtype), PLContext.WITH_COLUMNS))
        return self

    def impute(self, cols: IntoExprColumn, method: SimpleImputeMethod = "mean") -> Self:
        """
        Imputes null values in the given columns. Note: this doesn't fill NaN. If filling for NaN is needed,
        please manually do that.

        Parameters
        ----------
        cols
            Any Polars expression that can be understood as columns.
        method
            One of `mean`, `median`, `mode`. If `mode`, a random value will be chosen if there is
            a tie.
        """
        self._steps.append(FitStep(partial(t.impute, method=method), cols, self.exclude))
        return self

    def conditional_impute(
        self, rules_dict: Dict[str, str | pl.Expr], method: SimpleImputeMethod = "mean"
    ) -> Self:
        """
        Conditionally imputes values in the given columns. This transform will collect if input is lazy.

        Parameters
        ----------
        rules_dict
            Dictionary where keys are column names (must be string), and values are SQL/Polars Conditions
            that when true, those values in the column will be imputed,
            and the value to impute will be learned on the data where the condition is false.
        method
            One of `mean`, `median`, `mode`. If `mode`, a random value will be chosen if there is
            a tie.
        """
        self._steps.append(
            FitStep(
                partial(t.conditional_impute, rules_dict=rules_dict, method=method),
                None,
                self.exclude,
            )
        )
        return self

    def nan_to_null(self) -> Self:
        """
        Maps NaN values in all columns to null.
        """
        self._steps.append(ExprStep(cs.float().fill_nan(None), PLContext.WITH_COLUMNS))
        return self

    def int_to_float(self, f32: bool = True) -> Self:
        """
        Maps all integer columns to float.

        Parameters
        ----------
        f32
            If true, map all integer columns to f32 columns. Otherwise they will be
            casted to f64 columns.
        """
        if f32:
            self._steps.append(ExprStep(cs.integer().cast(pl.Float32), PLContext.WITH_COLUMNS))
        else:
            self._steps.append(ExprStep(cs.integer().cast(pl.Float64), PLContext.WITH_COLUMNS))
        return self

    def linear_impute(
        self, features: IntoExprColumn, target: str | pl.Expr | None = None, add_bias: bool = False
    ) -> Self:
        """
        Imputes the target column by training a simple linear regression using the other features. This will
        cast the target column to f64.

        Note: The linear regression will skip nulls whenever there is a null in the features or in the target.
        Additionally, if NaN or Inf exists in data, the linear regression result may be invalid or an error
        will be thrown. It is recommended to use this only after imputing and dealing with NaN and Infs for
        all feature columns first.

        Parameters
        ----------
        features
            Any Polars expression that can be understood as numerical columns which will be used as features
        target
            The target column
        add_bias
            Whether to add a bias term to the linear regression
        """
        self._steps.append(
            FitStep(
                partial(t.linear_impute, target=self._get_target(target), add_bias=add_bias),
                features,
                self.exclude,
            )
        )
        return self

    def scale(self, cols: IntoExprColumn, method: SimpleScaleMethod = "standard") -> Self:
        """
        Scales values in the given columns.

        Parameters
        ----------
        cols
            Any Polars expression that can be understood as columns.
        method
            One of `standard`, `min_max`, `abs_max`
        """
        self._steps.append(FitStep(partial(t.scale, method=method), cols, self.exclude))
        return self

    def robust_scale(self, cols: IntoExprColumn, q_low: float, q_high: float) -> Self:
        """
        Performs robust scaling on the given columns

        Paramters
        ---------
        cols
            Any Polars expression that can be understood as columns.
        q_low
            The lower quantile value
        q_high
            The higher quantile value
        """
        self._steps.append(
            FitStep(partial(t.robust_scale, q_low=q_low, q_high=q_high), cols, self.exclude)
        )
        return self

    def center(self, cols: IntoExprColumn) -> Self:
        """
        Centers the columns by subtracting each with its mean.

        Paramters
        ---------
        cols
            Any Polars expression that can be understood as columns.
        """
        self._steps.append(FitStep(partial(t.center), cols, self.exclude))
        return self

    def select(self, *cols: IntoExprColumn) -> Self:
        """
        Selects the columns from the dataset.

        Paramters
        ---------
        cols
            Any Polars expression that can be understood as columns.
        """
        self._steps.append(ExprStep(list(cols), PLContext.SELECT))
        return self

    def select_by_std(self, min_: float, max_: float) -> Self:
        """
        Fits and keeps columns that have standard deviation between `min_` and `max_`.
        Non-numeric columns will always be selected. Only numeric columns with std outside
        the `min_` `max_` bounds will be removed. This will never exclude the target if target
        is set.

        Parameters
        ----------
        min_
            Min standard deviation to select, inclusive.
        max_
            Max standard deviation to select, exclusive
        """
        self._steps.append(
            FitStep(
                partial(t.select_by_std, min_=min_, max_=max_),
                pl.col("*"),
                self.exclude,
                PLContext.SELECT,
            )
        )
        return self

    def polynomial_features(
        self, cols: List[str], degree: int, interaction_only: bool = True
    ) -> Self:
        """
        Generates polynomial combinations out of the features given, at the given degree.

        Parameters
        ----------
        cols
            A list of strings representing column names. Input to this function cannot be Polars expressions.
        degree
            The degree of the polynomial combination
        interaction_only
            It true, only combinations that involve 2 or more variables will be used.
        """
        if not all(isinstance(s, str) for s in cols):
            raise ValueError(
                "Input columns to `polynomial_features` must all be strings represeting column names."
            )

        self._steps.append(
            ExprStep(
                t.polynomial_features(cols, degree=degree, interaction_only=interaction_only),
                PLContext.WITH_COLUMNS,
            )
        )
        return self

    def winsorize(
        self,
        cols: IntoExprColumn,
        q_low: float = 0.05,
        q_high: float = 0.95,
        method: QuantileMethod = "nearest",
    ) -> Self:
        """
        Learns the lower and upper percentile from the columns, then clip each end at those values.
        If you wish to clip by constant values, you may append expression like pl.col(c).clip(c1, c2),
        where c1 and c2 are constants decided by the user.

        Parameters
        ----------
        cols
            Any Polars expression that can be understood as columns. Columns must be numerical.
        q_low
            The lower quantile value
        q_high
            The higher quantile value
        method
            Method to compute quantile. One of `nearest`, `higher`, `lower`, `midpoint`, `linear`.
        """
        self._steps.append(
            FitStep(
                partial(t.winsorize, q_low=q_low, q_high=q_high, method=method),
                cols,
                self.exclude,
            )
        )
        return self

    def drop(self, cols: IntoExprColumn) -> Self:
        """
        Drops the columns from the dataset.

        Paramters
        ---------
        cols
            Any Polars expression that can be understood as columns.
        """
        self._steps.append(ExprStep(pl.exclude(cols), PLContext.SELECT))
        return self

    def rename(self, rename_dict: Dict[str, str]) -> Self:
        """
        Renames the columns by the mapping.

        Paramters
        ---------
        rename_dict
            The name mapping
        """
        old = list(rename_dict.keys())
        self._steps.append(
            ExprStep([pl.col(k).alias(v) for k, v in rename_dict.items()], PLContext.WITH_COLUMNS)
        )

        return self.drop([o for o in old if o not in set(rename_dict.values())])

    def one_hot_encode(
        self,
        cols: IntoExprColumn | None = None,
        separator: str = "_",
        drop_first: bool = False,
        drop_cols: bool = True,
    ) -> Self:
        """
        Find the unique values in the string/categorical columns and one-hot encode them. This will NOT
        consider nulls as one of the unique values. Append pl.col(c).is_null().cast(pl.UInt8)
        expression to the pipeline if you want null indicators.

        If `cols` is None, this will pick all string and categorical columns.

        Parameters
        ----------
        cols
            Any Polars expression that can be understood as columns. If None, all string/categorical columns will be encoded.
        separator
            E.g. if column name is `col` and `a` is an elemenet in it, then the one-hot encoded column will be called
            `col_a` where the separator `_` is used.
        drop_first
            Whether to drop the first distinct value (in terms of str/categorical order). This helps with reducing
            dimension and prevents some issues from linear dependency.
        drop_cols
            Whether to drop the original columns after the transform
        """
        # First append new columns
        self._steps.append(
            FitStep(
                partial(t.one_hot_encode, separator=separator, drop_first=drop_first),
                cols if cols is not None else cs.string() | cs.categorical(),
                self.exclude,
            )
        )
        # Whether to drop the og str columns
        if drop_cols:
            if cols is None:
                return self.drop([pl.String, pl.Categorical])
            return self.drop(cols)
        return self

    def ordinal_encode(
        self,
        cols: IntoExprColumn | None = None,
        unknown_value: float | None = None,
        null_value: float | None = None,
    ) -> Self:
        """
        Find the unique values in the string/categorical columns and one-hot encode them. This will NOT
        consider nulls as one of the unique values. Append pl.col(c).is_null().cast(pl.UInt8)
        expression to the pipeline if you want null indicators.

        If `cols` is None, this will pick all string and categorical columns.

        Parameters
        ----------
        cols
            A list of strings representing column names.
        unknown_value
            What to assign to values not seen in the initial dataset used to fit. None means null.
        null_value
            What to assign to values that are null in dataset.
        """
        self._steps.append(
            FitStep(
                partial(t.ordinal_encode, unknown_value=unknown_value, null_value=null_value),
                cols if cols is not None else cs.string() | cs.categorical(),
                self.exclude,
            )
        )
        return self

    def rank_hot_encode(
        self,
        col: str | pl.Expr,
        ranking: List[str],
        default_rank: int | None = None,
        drop_cols: bool = True,
    ) -> Self:
        """
        Given a ranking, e.g. ["bad", "neutral", "good"], where "bad", "neutral" and "good" are values coming
        from the column `col`, this will return two new columns, the first is ">=neutral", which
        will be 1 for all values in ("neutral", "good") and 0 otherwise, and the second new column is ">=good", which
        will be 1 for all values in ("good") and 0 otherwise.

        Parameters
        ----------
        col
            The name of a single column
        ranking
            A list of string representing the ranking of the values
        default_rank
            Default rank for all null/unseen values
        drop_cols
            Whether to drop the original column `col`
        """
        self._steps.append(
            ExprStep(
                t.rank_hot_encode(col=col, ranking=ranking, default_rank=default_rank),
                PLContext.WITH_COLUMNS,
            )
        )
        if drop_cols:
            return self.drop(cols=[col])
        return self

    def target_encode(
        self,
        cols: IntoExprColumn | None = None,
        target: str | pl.Expr | None = None,
        min_samples_leaf: int = 20,
        smoothing: float = 10.0,
        default: EncoderDefaultStrategy | float | None = None,
    ) -> Self:
        """
        Target encode the given variables.

        Note: nulls will be encoded as well.

        Parameters
        ----------
        cols
            Any Polars expression that can be understood as columns. Columns of type != string/categorical
            will not produce any expression. If None, all string/categorical columns will be used.
        target
            The target column
        min_samples_leaf
            A regularization factor
        smoothing
            Smoothing effect to balance categorical average vs prior
        default
            If a new value is encountered during transform (unseen in training dataset), it will be mapped to default.
            If this is a string, it can be `null`, `zero`, or `mean`, where `mean` means map them to the mean of the target.

        Reference
        ---------
        https://contrib.scikit-learn.org/category_encoders/targetencoder.html
        """
        self._steps.append(
            FitStep(
                partial(
                    t.target_encode,
                    target=self._get_target(target),
                    min_samples_leaf=min_samples_leaf,
                    smoothing=smoothing,
                    default=default,
                ),
                cols if cols is not None else cs.string() | cs.categorical(),
                self.exclude,
            )
        )
        return self

    def woe_encode(
        self,
        cols: IntoExprColumn | None = None,
        target: str | pl.Expr | None = None,
        default: EncoderDefaultStrategy | float | None = None,
    ) -> Self:
        """
        Use Weight of Evidence to encode a discrete variable x with respect to target. This assumes x
        is discrete and castable to String. A value of 1 is added to all events/non-events
        (goods/bads) to smooth the computation. This is -1 * output of the package category_encoder's WOEEncoder.

        Note: nulls will be encoded as well.

        Parameters
        ----------
        cols
            Any Polars expression that can be understood as columns. Columns of type != string/categorical
            will not produce any expression. If None, all string/categorical columns will be used.
        target
            The target column
        default
            If a new value is encountered during transform (unseen in training dataset), it will be mapped to default.
            If this is a string, it can be `null`, `zero`, or `mean`, where `mean` means map them to the mean of the target.

        Reference
        ---------
        https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html
        """
        self._steps.append(
            FitStep(
                partial(
                    t.woe_encode,
                    target=self._get_target(target),
                    default=default,
                ),
                cols if cols is not None else cs.string() | cs.categorical(),
                self.exclude,
            )
        )
        return self

    def iv_encode(
        self,
        cols: IntoExprColumn | None = None,
        target: str | pl.Expr | None = None,
        default: EncoderDefaultStrategy | float | None = None,
    ) -> Self:
        """
        Use Information Value to encode a discrete variable x with respect to target. This assumes x
        is discrete and castable to String. A value of 1 is added to all events/non-events
        (goods/bads) to smooth the computation.

        Note: nulls will be encoded as well.

        Parameters
        ----------
        cols
            Any Polars expression that can be understood as columns. Columns of type != string/categorical
            will not produce any expression. If None, all string/categorical columns will be used.
        target
            The target column
        default
            If a new value is encountered during transform (unseen in training dataset), it will be mapped to default.
            If this is a string, it can be `null`, `zero`, or `mean`, where `mean` means map them to the mean of the target.

        Reference
        ---------
        https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html
        """
        self._steps.append(
            FitStep(
                partial(
                    t.iv_encode,
                    target=self._get_target(target),
                    default=default,
                ),
                cols if cols is not None else cs.string() | cs.categorical(),
                self.exclude,
            )
        )
        return self

    def with_columns(self, *exprs: pl.Expr) -> Self:
        """
        Run Polars with_columns for the expressions.
        """
        self._steps.append(ExprStep(list(exprs), PLContext.WITH_COLUMNS))
        return self

    def sort(
        self, by: IntoExprColumn, descending: bool | List[bool], maintain_order: bool = True
    ) -> Self:
        """Sorts the dataframe by the columns.

        Parameters
        ----------
        by
            The columns to sort by
        descending
            Whether the sort should be descending for the corresponding sort column
        maintain_order
            Whether the sort should maintain order or not
        """
        self._steps.append(SortStep(by=by, descending=descending, maintain_order=maintain_order))
        return self

    def explode(self, columns: str | pl.Expr | List[str] | List[pl.Expr]) -> Self:
        """Transform that represents `df.explode(columns)`"""
        if isinstance(columns, (str, pl.Expr)):
            exprs = [to_expr(columns)]
        elif isinstance(columns, list):
            exprs = [to_expr(c) for c in columns]
        else:
            raise ValueError(
                "Input `columns` must be a string, or a pl.Expr or a list of str or pl.Expr."
            )

        self._steps.append(ExprStep(exprs, PLContext.EXPLODE))
        return self

    def group_by_agg(
        self, by: IntoExprColumn, agg: List[pl.Expr], maintain_order: bool = False
    ) -> Self:
        """
        Performs a group by and agg on the data.

        Parameters
        ----------
        by
            The columns to group by
        agg
            The aggregation functions to run
        maintain_order
            Whether to maintain the group by order
        """
        self._steps.append(
            GroupByAggStep(by=by, agg=[to_expr(a) for a in agg], maintain_order=maintain_order)
        )
        return self

    def group_by_dynamic_agg(
        self,
        index_column: str,
        agg: List[pl.Expr],
        every: str,
        period: str | None = None,
        offset: str | None = None,
        include_boundaries: bool = False,
        closed: Literal["left", "right", "both", "none"] = "left",
        label: Literal["left", "right", "datapoint"] = "left",
        group_by: IntoExprColumn | None = None,
        start_by: Literal[
            "window",
            "datapoint",
            "monday",
            "tuesday",
            "wednesday",
            "thursday",
            "friday",
            "saturday",
            "sunday",
        ] = "window",
    ) -> Self:
        """
        See polars group_by_dynamic documentation for an explanation on the input arguments.

        https://docs.pola.rs/api/python/stable/reference/dataframe/api/polars.DataFrame.group_by_dynamic.html#polars.DataFrame.group_by_dynamic
        """
        self._steps.append(
            GroupByDynAggStep(
                index_column=index_column,
                agg=[to_expr(a) for a in agg],
                every=every,
                group_by=group_by,
                period=period,
                offset=offset,
                include_boundaries=include_boundaries,
                closed=closed,
                label=label,
                start_by=start_by,
            )
        )
        return self

    # How to type this?
    def append_fit_func(self, func, cols: IntoExprColumn, **kwargs) -> Self:
        """
        Adds a custom transform that requires a fit step in the blueprint.

        Any custom function must satistfy the following function signature:
        my_func(df:Union[pl.DataFrame, pl.LazyFrame], cols: List[str], ...) -> List[pl.Expr]
        where ... means kwargs. The fit step "learns" the parameters needed to translate
        the transform into concrete expressions.

        Parameters
        ----------
        func
            A callable with signature (pl.DataFrame | pl.LazyFrame, cols: List[str], ...) -> ExprTransform,
        cols
            The columns to be fed into the func. Note that in func's signature, a list of strings
            should be expected. But here, cols can be any polars selector expression. The reason is that
            during "fit", cols is turned into concrete column names.
        **kwargs
            Any other arguments to func must be passed as kwargs
        """
        import inspect

        keywords = kwargs.copy()
        if "target" in inspect.signature(func).parameters:  # func has "target" as input
            if "target" not in kwargs:  # if target is not explicitly given
                keywords["target"] = self._get_target()
                if keywords["target"] is None:
                    raise ValueError(
                        "Target is not explicitly given and is required by the custom function."
                    )

        self._steps.append(
            FitStep(
                partial(func, **keywords),
                cols,
                self.exclude,
            )
        )
        return self

    def append_step_from_dict(self, dictionary: Dict[str, Any]) -> Self:
        """
        Append a step to the blueprint by taking in a dictionary with keys `name`, `args`, and `kwargs`, where
        the value of args must be a List[Any] and the value of kwargs must be a dict[str, Any].
        """
        step_repr: StepRepr = StepRepr.from_dict(dictionary)
        func = getattr(self, step_repr.name, None)  # Default is None
        if func is None or step_repr.name.startswith("_"):
            raise ValueError("Unknown or invalid method name.")

        return func(*step_repr.args, **step_repr.kwargs)

    def materialize(
        self, return_df: bool = False, **kwargs
    ) -> Pipeline | Tuple[pl.LazyFrame, Pipeline]:
        """
        Materialize the blueprint, which means that it will fit and learn all the paramters needed.

        Parameters
        ----------
        return_df
            If true, return the entire query plan, a lazy df, together with the Pipeline object
        **kwargs
            All kwargs here will be passed to polars.LazyFrame.collect()
        """
        transforms: List[PipelineStep] = []
        df: pl.DataFrame = self._df.collect(**kwargs)
        # Let this lazy plan go through the fit process. The frame will be collected temporarily but
        # the collect should be and optimized.
        df_lazy: pl.LazyFrame = df.lazy()
        for step in self._steps:
            if isinstance(step, FitStep):  # Need fitting, which is done here
                df_temp: pl.DataFrame = df_lazy.collect(**kwargs)
                exprs = step.fit(df_temp)
                step = ExprStep(exprs, step.context)
                transforms.append(step)
                df_lazy = step.apply_df(df_temp.lazy())
            elif isinstance(step, tuple(_SERIALIZABLE_STEPS)):
                transforms.append(step)
                df_lazy = step.apply_df(df_lazy)
            else:
                raise ValueError(f"Not a valid step: {step.__class__}")

        pipe = Pipeline(
            name=self.name,
            feature_names_in_=list(self.feature_names_in_),
            feature_names_out_=df_lazy.collect_schema().names(),
            transforms=transforms,
            lowercase=self.lowercase,
            uppercase=self.uppercase,
        )

        if return_df:
            return df_lazy, pipe
        else:
            return pipe

    def fit(self, X=None, y=None, **kwargs) -> Pipeline:
        """
        Alias for self.materialize()
        """
        return self.materialize(**kwargs)

    def transform(self, df: PolarsFrame, **kwargs) -> pl.DataFrame:
        """
        Fits the blueprint with the dataframe that it is initialized with, and
        transforms the input dataframe.

        Parameters
        ----------
        df
            Any Polars dataframe
        **kwargs
            Will be passed to Pipeline's `transform` method.
        """
        return self.materialize().transform(df, **kwargs)

__init__(df, name='test', target=None, exclude=None, lowercase=False, uppercase=False)

Creates a blueprint object.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager Polars dataframe	required
name	str	Name of the blueprint.	'test'
target	str | None	Optionally indicate the target column in the ML pipeline. This will automatically prevent any transformation from changing the target column. (To be implemented: this should also automatically fill any transformation that requires a target name)	None
exclude	List[str] | None	Any other column to exclude from global transformation. Note: this is only needed if you are not specifiying the exact columns to transform. E.g. when you are using a selector like cs.numeric() for all numeric columns. If this is the case and target is not set nor excluded, then the transformation may be applied to the target as well, which is not desired in most cases. Therefore, it is highly recommended you initialize with target name.	None
lowercase	bool	Whether to insert a lowercase column name step before all other transformations. This takes precedence over uppercase.	False
uppercase	bool	Whether to insert a uppercase column name step before all other transformations. This only happens if lowercase is False	False

Source code in python/polars_ds/pipeline/pipeline.py

def __init__(
    self,
    df: PolarsFrame,
    name: str = "test",
    target: str | None = None,
    exclude: List[str] | None = None,
    lowercase: bool = False,
    uppercase: bool = False,
):
    """
    Creates a blueprint object.

    Parameters
    ----------
    df
        Either a lazy or an eager Polars dataframe
    name
        Name of the blueprint.
    target
        Optionally indicate the target column in the ML pipeline. This will automatically prevent any transformation
        from changing the target column. (To be implemented: this should also automatically fill any transformation
        that requires a target name)
    exclude
        Any other column to exclude from global transformation. Note: this is only needed if you are not specifiying
        the exact columns to transform. E.g. when you are using a selector like cs.numeric() for all numeric columns.
        If this is the case and target is not set nor excluded, then the transformation may be applied to the target
        as well, which is not desired in most cases. Therefore, it is highly recommended you initialize with target name.
    lowercase
        Whether to insert a lowercase column name step before all other transformations.
        This takes precedence over uppercase.
    uppercase
        Whether to insert a uppercase column name step before all other transformations.
        This only happens if lowercase is False
    """

    self._df: pl.LazyFrame = df.lazy()
    if lowercase:
        self._df = self._df.select(pl.all().name.to_lowercase())
    else:
        if uppercase:
            self._df = self._df.select(pl.all().name.to_uppercase())

    self.name: str = str(name)
    self.target = target
    self.feature_names_in_: list[str] = self._df.collect_schema().names()

    self._steps: List[ExprStep | FitStep] = []
    self.exclude: List[str] = [] if target is None else [target]
    if exclude is not None:  # dedup in case user accidentally puts the same column name twice
        self.exclude = list(set(self.exclude + exclude))

    self.lowercase = lowercase
    self.uppercase = uppercase

append_fit_func(func, cols, **kwargs)

Adds a custom transform that requires a fit step in the blueprint.

Any custom function must satistfy the following function signature: my_func(df:Union[pl.DataFrame, pl.LazyFrame], cols: List[str], ...) -> List[pl.Expr] where ... means kwargs. The fit step "learns" the parameters needed to translate the transform into concrete expressions.

Parameters:

Name	Type	Description	Default
func		A callable with signature (pl.DataFrame | pl.LazyFrame, cols: List[str], ...) -> ExprTransform,	required
cols	IntoExprColumn	The columns to be fed into the func. Note that in func's signature, a list of strings should be expected. But here, cols can be any polars selector expression. The reason is that during "fit", cols is turned into concrete column names.	required
**kwargs		Any other arguments to func must be passed as kwargs	{}

Source code in python/polars_ds/pipeline/pipeline.py

def append_fit_func(self, func, cols: IntoExprColumn, **kwargs) -> Self:
    """
    Adds a custom transform that requires a fit step in the blueprint.

    Any custom function must satistfy the following function signature:
    my_func(df:Union[pl.DataFrame, pl.LazyFrame], cols: List[str], ...) -> List[pl.Expr]
    where ... means kwargs. The fit step "learns" the parameters needed to translate
    the transform into concrete expressions.

    Parameters
    ----------
    func
        A callable with signature (pl.DataFrame | pl.LazyFrame, cols: List[str], ...) -> ExprTransform,
    cols
        The columns to be fed into the func. Note that in func's signature, a list of strings
        should be expected. But here, cols can be any polars selector expression. The reason is that
        during "fit", cols is turned into concrete column names.
    **kwargs
        Any other arguments to func must be passed as kwargs
    """
    import inspect

    keywords = kwargs.copy()
    if "target" in inspect.signature(func).parameters:  # func has "target" as input
        if "target" not in kwargs:  # if target is not explicitly given
            keywords["target"] = self._get_target()
            if keywords["target"] is None:
                raise ValueError(
                    "Target is not explicitly given and is required by the custom function."
                )

    self._steps.append(
        FitStep(
            partial(func, **keywords),
            cols,
            self.exclude,
        )
    )
    return self

append_step_from_dict(dictionary)

Append a step to the blueprint by taking in a dictionary with keys name, args, and kwargs, where the value of args must be a List[Any] and the value of kwargs must be a dict[str, Any].

Source code in python/polars_ds/pipeline/pipeline.py

def append_step_from_dict(self, dictionary: Dict[str, Any]) -> Self:
    """
    Append a step to the blueprint by taking in a dictionary with keys `name`, `args`, and `kwargs`, where
    the value of args must be a List[Any] and the value of kwargs must be a dict[str, Any].
    """
    step_repr: StepRepr = StepRepr.from_dict(dictionary)
    func = getattr(self, step_repr.name, None)  # Default is None
    if func is None or step_repr.name.startswith("_"):
        raise ValueError("Unknown or invalid method name.")

    return func(*step_repr.args, **step_repr.kwargs)

cast_bools(dtype=pl.UInt8)

Cast all boolean columns in the dataframe to the given type.

Source code in python/polars_ds/pipeline/pipeline.py

def cast_bools(self, dtype: pl.DataType = pl.UInt8) -> Self:
    """
    Cast all boolean columns in the dataframe to the given type.
    """
    self._steps.append(ExprStep(cs.boolean().cast(dtype), PLContext.WITH_COLUMNS))
    return self

center(cols)

Centers the columns by subtracting each with its mean.

Paramters

cols Any Polars expression that can be understood as columns.

Source code in python/polars_ds/pipeline/pipeline.py

def center(self, cols: IntoExprColumn) -> Self:
    """
    Centers the columns by subtracting each with its mean.

    Paramters
    ---------
    cols
        Any Polars expression that can be understood as columns.
    """
    self._steps.append(FitStep(partial(t.center), cols, self.exclude))
    return self

conditional_impute(rules_dict, method='mean')

Conditionally imputes values in the given columns. This transform will collect if input is lazy.

Parameters:

Name	Type	Description	Default
rules_dict	Dict[str, str | Expr]	Dictionary where keys are column names (must be string), and values are SQL/Polars Conditions that when true, those values in the column will be imputed, and the value to impute will be learned on the data where the condition is false.	required
method	SimpleImputeMethod	One of mean, median, mode. If mode, a random value will be chosen if there is a tie.	'mean'

Source code in python/polars_ds/pipeline/pipeline.py

def conditional_impute(
    self, rules_dict: Dict[str, str | pl.Expr], method: SimpleImputeMethod = "mean"
) -> Self:
    """
    Conditionally imputes values in the given columns. This transform will collect if input is lazy.

    Parameters
    ----------
    rules_dict
        Dictionary where keys are column names (must be string), and values are SQL/Polars Conditions
        that when true, those values in the column will be imputed,
        and the value to impute will be learned on the data where the condition is false.
    method
        One of `mean`, `median`, `mode`. If `mode`, a random value will be chosen if there is
        a tie.
    """
    self._steps.append(
        FitStep(
            partial(t.conditional_impute, rules_dict=rules_dict, method=method),
            None,
            self.exclude,
        )
    )
    return self

drop(cols)

Drops the columns from the dataset.

Paramters

cols Any Polars expression that can be understood as columns.

Source code in python/polars_ds/pipeline/pipeline.py

def drop(self, cols: IntoExprColumn) -> Self:
    """
    Drops the columns from the dataset.

    Paramters
    ---------
    cols
        Any Polars expression that can be understood as columns.
    """
    self._steps.append(ExprStep(pl.exclude(cols), PLContext.SELECT))
    return self

explode(columns)

Transform that represents df.explode(columns)

Source code in python/polars_ds/pipeline/pipeline.py

def explode(self, columns: str | pl.Expr | List[str] | List[pl.Expr]) -> Self:
    """Transform that represents `df.explode(columns)`"""
    if isinstance(columns, (str, pl.Expr)):
        exprs = [to_expr(columns)]
    elif isinstance(columns, list):
        exprs = [to_expr(c) for c in columns]
    else:
        raise ValueError(
            "Input `columns` must be a string, or a pl.Expr or a list of str or pl.Expr."
        )

    self._steps.append(ExprStep(exprs, PLContext.EXPLODE))
    return self

filter(by)

Filters on the dataframe using native polars expressions or SQL boolean expressions.

Parameters:

Name	Type	Description	Default
by	str | Expr	Native polars boolean expression or SQL strings	required

Source code in python/polars_ds/pipeline/pipeline.py

def filter(self, by: str | pl.Expr) -> Self:
    """
    Filters on the dataframe using native polars expressions or SQL boolean expressions.

    Parameters
    ----------
    by
        Native polars boolean expression or SQL strings
    """
    self._steps.append(
        ExprStep(by if isinstance(by, pl.Expr) else pl.sql_expr(by), PLContext.FILTER)
    )
    return self

fit(X=None, y=None, **kwargs)

Alias for self.materialize()

Source code in python/polars_ds/pipeline/pipeline.py

def fit(self, X=None, y=None, **kwargs) -> Pipeline:
    """
    Alias for self.materialize()
    """
    return self.materialize(**kwargs)

group_by_agg(by, agg, maintain_order=False)

Performs a group by and agg on the data.

Parameters:

Name	Type	Description	Default
by	IntoExprColumn	The columns to group by	required
agg	List[Expr]	The aggregation functions to run	required
maintain_order	bool	Whether to maintain the group by order	False

Source code in python/polars_ds/pipeline/pipeline.py

def group_by_agg(
    self, by: IntoExprColumn, agg: List[pl.Expr], maintain_order: bool = False
) -> Self:
    """
    Performs a group by and agg on the data.

    Parameters
    ----------
    by
        The columns to group by
    agg
        The aggregation functions to run
    maintain_order
        Whether to maintain the group by order
    """
    self._steps.append(
        GroupByAggStep(by=by, agg=[to_expr(a) for a in agg], maintain_order=maintain_order)
    )
    return self

group_by_dynamic_agg(index_column, agg, every, period=None, offset=None, include_boundaries=False, closed='left', label='left', group_by=None, start_by='window')

See polars group_by_dynamic documentation for an explanation on the input arguments.

https://docs.pola.rs/api/python/stable/reference/dataframe/api/polars.DataFrame.group_by_dynamic.html#polars.DataFrame.group_by_dynamic

Source code in python/polars_ds/pipeline/pipeline.py

def group_by_dynamic_agg(
    self,
    index_column: str,
    agg: List[pl.Expr],
    every: str,
    period: str | None = None,
    offset: str | None = None,
    include_boundaries: bool = False,
    closed: Literal["left", "right", "both", "none"] = "left",
    label: Literal["left", "right", "datapoint"] = "left",
    group_by: IntoExprColumn | None = None,
    start_by: Literal[
        "window",
        "datapoint",
        "monday",
        "tuesday",
        "wednesday",
        "thursday",
        "friday",
        "saturday",
        "sunday",
    ] = "window",
) -> Self:
    """
    See polars group_by_dynamic documentation for an explanation on the input arguments.

    https://docs.pola.rs/api/python/stable/reference/dataframe/api/polars.DataFrame.group_by_dynamic.html#polars.DataFrame.group_by_dynamic
    """
    self._steps.append(
        GroupByDynAggStep(
            index_column=index_column,
            agg=[to_expr(a) for a in agg],
            every=every,
            group_by=group_by,
            period=period,
            offset=offset,
            include_boundaries=include_boundaries,
            closed=closed,
            label=label,
            start_by=start_by,
        )
    )
    return self

impute(cols, method='mean')

Imputes null values in the given columns. Note: this doesn't fill NaN. If filling for NaN is needed, please manually do that.

Parameters:

Name	Type	Description	Default
cols	IntoExprColumn	Any Polars expression that can be understood as columns.	required
method	SimpleImputeMethod	One of mean, median, mode. If mode, a random value will be chosen if there is a tie.	'mean'

Source code in python/polars_ds/pipeline/pipeline.py

def impute(self, cols: IntoExprColumn, method: SimpleImputeMethod = "mean") -> Self:
    """
    Imputes null values in the given columns. Note: this doesn't fill NaN. If filling for NaN is needed,
    please manually do that.

    Parameters
    ----------
    cols
        Any Polars expression that can be understood as columns.
    method
        One of `mean`, `median`, `mode`. If `mode`, a random value will be chosen if there is
        a tie.
    """
    self._steps.append(FitStep(partial(t.impute, method=method), cols, self.exclude))
    return self

int_to_float(f32=True)

Maps all integer columns to float.

Parameters:

Name	Type	Description	Default
f32	bool	If true, map all integer columns to f32 columns. Otherwise they will be casted to f64 columns.	True

Source code in python/polars_ds/pipeline/pipeline.py

def int_to_float(self, f32: bool = True) -> Self:
    """
    Maps all integer columns to float.

    Parameters
    ----------
    f32
        If true, map all integer columns to f32 columns. Otherwise they will be
        casted to f64 columns.
    """
    if f32:
        self._steps.append(ExprStep(cs.integer().cast(pl.Float32), PLContext.WITH_COLUMNS))
    else:
        self._steps.append(ExprStep(cs.integer().cast(pl.Float64), PLContext.WITH_COLUMNS))
    return self

iv_encode(cols=None, target=None, default=None)

Use Information Value to encode a discrete variable x with respect to target. This assumes x is discrete and castable to String. A value of 1 is added to all events/non-events (goods/bads) to smooth the computation.

Note: nulls will be encoded as well.

Parameters:

Name	Type	Description	Default
cols	IntoExprColumn | None	Any Polars expression that can be understood as columns. Columns of type != string/categorical will not produce any expression. If None, all string/categorical columns will be used.	None
target	str | Expr | None	The target column	None
default	EncoderDefaultStrategy | float | None	If a new value is encountered during transform (unseen in training dataset), it will be mapped to default. If this is a string, it can be null, zero, or mean, where mean means map them to the mean of the target.	None

Reference

https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html

Source code in python/polars_ds/pipeline/pipeline.py

def iv_encode(
    self,
    cols: IntoExprColumn | None = None,
    target: str | pl.Expr | None = None,
    default: EncoderDefaultStrategy | float | None = None,
) -> Self:
    """
    Use Information Value to encode a discrete variable x with respect to target. This assumes x
    is discrete and castable to String. A value of 1 is added to all events/non-events
    (goods/bads) to smooth the computation.

    Note: nulls will be encoded as well.

    Parameters
    ----------
    cols
        Any Polars expression that can be understood as columns. Columns of type != string/categorical
        will not produce any expression. If None, all string/categorical columns will be used.
    target
        The target column
    default
        If a new value is encountered during transform (unseen in training dataset), it will be mapped to default.
        If this is a string, it can be `null`, `zero`, or `mean`, where `mean` means map them to the mean of the target.

    Reference
    ---------
    https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html
    """
    self._steps.append(
        FitStep(
            partial(
                t.iv_encode,
                target=self._get_target(target),
                default=default,
            ),
            cols if cols is not None else cs.string() | cs.categorical(),
            self.exclude,
        )
    )
    return self

linear_impute(features, target=None, add_bias=False)

Imputes the target column by training a simple linear regression using the other features. This will cast the target column to f64.

Note: The linear regression will skip nulls whenever there is a null in the features or in the target. Additionally, if NaN or Inf exists in data, the linear regression result may be invalid or an error will be thrown. It is recommended to use this only after imputing and dealing with NaN and Infs for all feature columns first.

Parameters:

Name	Type	Description	Default
features	IntoExprColumn	Any Polars expression that can be understood as numerical columns which will be used as features	required
target	str | Expr | None	The target column	None
add_bias	bool	Whether to add a bias term to the linear regression	False

Source code in python/polars_ds/pipeline/pipeline.py

def linear_impute(
    self, features: IntoExprColumn, target: str | pl.Expr | None = None, add_bias: bool = False
) -> Self:
    """
    Imputes the target column by training a simple linear regression using the other features. This will
    cast the target column to f64.

    Note: The linear regression will skip nulls whenever there is a null in the features or in the target.
    Additionally, if NaN or Inf exists in data, the linear regression result may be invalid or an error
    will be thrown. It is recommended to use this only after imputing and dealing with NaN and Infs for
    all feature columns first.

    Parameters
    ----------
    features
        Any Polars expression that can be understood as numerical columns which will be used as features
    target
        The target column
    add_bias
        Whether to add a bias term to the linear regression
    """
    self._steps.append(
        FitStep(
            partial(t.linear_impute, target=self._get_target(target), add_bias=add_bias),
            features,
            self.exclude,
        )
    )
    return self

materialize(return_df=False, **kwargs)

Materialize the blueprint, which means that it will fit and learn all the paramters needed.

Parameters:

Name	Type	Description	Default
return_df	bool	If true, return the entire query plan, a lazy df, together with the Pipeline object	False
**kwargs		All kwargs here will be passed to polars.LazyFrame.collect()	{}

Source code in python/polars_ds/pipeline/pipeline.py

def materialize(
    self, return_df: bool = False, **kwargs
) -> Pipeline | Tuple[pl.LazyFrame, Pipeline]:
    """
    Materialize the blueprint, which means that it will fit and learn all the paramters needed.

    Parameters
    ----------
    return_df
        If true, return the entire query plan, a lazy df, together with the Pipeline object
    **kwargs
        All kwargs here will be passed to polars.LazyFrame.collect()
    """
    transforms: List[PipelineStep] = []
    df: pl.DataFrame = self._df.collect(**kwargs)
    # Let this lazy plan go through the fit process. The frame will be collected temporarily but
    # the collect should be and optimized.
    df_lazy: pl.LazyFrame = df.lazy()
    for step in self._steps:
        if isinstance(step, FitStep):  # Need fitting, which is done here
            df_temp: pl.DataFrame = df_lazy.collect(**kwargs)
            exprs = step.fit(df_temp)
            step = ExprStep(exprs, step.context)
            transforms.append(step)
            df_lazy = step.apply_df(df_temp.lazy())
        elif isinstance(step, tuple(_SERIALIZABLE_STEPS)):
            transforms.append(step)
            df_lazy = step.apply_df(df_lazy)
        else:
            raise ValueError(f"Not a valid step: {step.__class__}")

    pipe = Pipeline(
        name=self.name,
        feature_names_in_=list(self.feature_names_in_),
        feature_names_out_=df_lazy.collect_schema().names(),
        transforms=transforms,
        lowercase=self.lowercase,
        uppercase=self.uppercase,
    )

    if return_df:
        return df_lazy, pipe
    else:
        return pipe

nan_to_null()

Maps NaN values in all columns to null.

Source code in python/polars_ds/pipeline/pipeline.py

def nan_to_null(self) -> Self:
    """
    Maps NaN values in all columns to null.
    """
    self._steps.append(ExprStep(cs.float().fill_nan(None), PLContext.WITH_COLUMNS))
    return self

one_hot_encode(cols=None, separator='_', drop_first=False, drop_cols=True)

Find the unique values in the string/categorical columns and one-hot encode them. This will NOT consider nulls as one of the unique values. Append pl.col(c).is_null().cast(pl.UInt8) expression to the pipeline if you want null indicators.

If cols is None, this will pick all string and categorical columns.

Parameters:

Name	Type	Description	Default
cols	IntoExprColumn | None	Any Polars expression that can be understood as columns. If None, all string/categorical columns will be encoded.	None
separator	str	E.g. if column name is col and a is an elemenet in it, then the one-hot encoded column will be called col_a where the separator _ is used.	'_'
drop_first	bool	Whether to drop the first distinct value (in terms of str/categorical order). This helps with reducing dimension and prevents some issues from linear dependency.	False
drop_cols	bool	Whether to drop the original columns after the transform	True

Source code in python/polars_ds/pipeline/pipeline.py

def one_hot_encode(
    self,
    cols: IntoExprColumn | None = None,
    separator: str = "_",
    drop_first: bool = False,
    drop_cols: bool = True,
) -> Self:
    """
    Find the unique values in the string/categorical columns and one-hot encode them. This will NOT
    consider nulls as one of the unique values. Append pl.col(c).is_null().cast(pl.UInt8)
    expression to the pipeline if you want null indicators.

    If `cols` is None, this will pick all string and categorical columns.

    Parameters
    ----------
    cols
        Any Polars expression that can be understood as columns. If None, all string/categorical columns will be encoded.
    separator
        E.g. if column name is `col` and `a` is an elemenet in it, then the one-hot encoded column will be called
        `col_a` where the separator `_` is used.
    drop_first
        Whether to drop the first distinct value (in terms of str/categorical order). This helps with reducing
        dimension and prevents some issues from linear dependency.
    drop_cols
        Whether to drop the original columns after the transform
    """
    # First append new columns
    self._steps.append(
        FitStep(
            partial(t.one_hot_encode, separator=separator, drop_first=drop_first),
            cols if cols is not None else cs.string() | cs.categorical(),
            self.exclude,
        )
    )
    # Whether to drop the og str columns
    if drop_cols:
        if cols is None:
            return self.drop([pl.String, pl.Categorical])
        return self.drop(cols)
    return self

ordinal_encode(cols=None, unknown_value=None, null_value=None)

Find the unique values in the string/categorical columns and one-hot encode them. This will NOT consider nulls as one of the unique values. Append pl.col(c).is_null().cast(pl.UInt8) expression to the pipeline if you want null indicators.

If cols is None, this will pick all string and categorical columns.

Parameters:

Name	Type	Description	Default
cols	IntoExprColumn | None	A list of strings representing column names.	None
unknown_value	float | None	What to assign to values not seen in the initial dataset used to fit. None means null.	None
null_value	float | None	What to assign to values that are null in dataset.	None

Source code in python/polars_ds/pipeline/pipeline.py

def ordinal_encode(
    self,
    cols: IntoExprColumn | None = None,
    unknown_value: float | None = None,
    null_value: float | None = None,
) -> Self:
    """
    Find the unique values in the string/categorical columns and one-hot encode them. This will NOT
    consider nulls as one of the unique values. Append pl.col(c).is_null().cast(pl.UInt8)
    expression to the pipeline if you want null indicators.

    If `cols` is None, this will pick all string and categorical columns.

    Parameters
    ----------
    cols
        A list of strings representing column names.
    unknown_value
        What to assign to values not seen in the initial dataset used to fit. None means null.
    null_value
        What to assign to values that are null in dataset.
    """
    self._steps.append(
        FitStep(
            partial(t.ordinal_encode, unknown_value=unknown_value, null_value=null_value),
            cols if cols is not None else cs.string() | cs.categorical(),
            self.exclude,
        )
    )
    return self

polynomial_features(cols, degree, interaction_only=True)

Generates polynomial combinations out of the features given, at the given degree.

Parameters:

Name	Type	Description	Default
cols	List[str]	A list of strings representing column names. Input to this function cannot be Polars expressions.	required
degree	int	The degree of the polynomial combination	required
interaction_only	bool	It true, only combinations that involve 2 or more variables will be used.	True

Source code in python/polars_ds/pipeline/pipeline.py

def polynomial_features(
    self, cols: List[str], degree: int, interaction_only: bool = True
) -> Self:
    """
    Generates polynomial combinations out of the features given, at the given degree.

    Parameters
    ----------
    cols
        A list of strings representing column names. Input to this function cannot be Polars expressions.
    degree
        The degree of the polynomial combination
    interaction_only
        It true, only combinations that involve 2 or more variables will be used.
    """
    if not all(isinstance(s, str) for s in cols):
        raise ValueError(
            "Input columns to `polynomial_features` must all be strings represeting column names."
        )

    self._steps.append(
        ExprStep(
            t.polynomial_features(cols, degree=degree, interaction_only=interaction_only),
            PLContext.WITH_COLUMNS,
        )
    )
    return self

rank_hot_encode(col, ranking, default_rank=None, drop_cols=True)

Given a ranking, e.g. ["bad", "neutral", "good"], where "bad", "neutral" and "good" are values coming from the column col, this will return two new columns, the first is ">=neutral", which will be 1 for all values in ("neutral", "good") and 0 otherwise, and the second new column is ">=good", which will be 1 for all values in ("good") and 0 otherwise.

Parameters:

Name	Type	Description	Default
col	str | Expr	The name of a single column	required
ranking	List[str]	A list of string representing the ranking of the values	required
default_rank	int | None	Default rank for all null/unseen values	None
drop_cols	bool	Whether to drop the original column col	True

Source code in python/polars_ds/pipeline/pipeline.py

def rank_hot_encode(
    self,
    col: str | pl.Expr,
    ranking: List[str],
    default_rank: int | None = None,
    drop_cols: bool = True,
) -> Self:
    """
    Given a ranking, e.g. ["bad", "neutral", "good"], where "bad", "neutral" and "good" are values coming
    from the column `col`, this will return two new columns, the first is ">=neutral", which
    will be 1 for all values in ("neutral", "good") and 0 otherwise, and the second new column is ">=good", which
    will be 1 for all values in ("good") and 0 otherwise.

    Parameters
    ----------
    col
        The name of a single column
    ranking
        A list of string representing the ranking of the values
    default_rank
        Default rank for all null/unseen values
    drop_cols
        Whether to drop the original column `col`
    """
    self._steps.append(
        ExprStep(
            t.rank_hot_encode(col=col, ranking=ranking, default_rank=default_rank),
            PLContext.WITH_COLUMNS,
        )
    )
    if drop_cols:
        return self.drop(cols=[col])
    return self

rename(rename_dict)

Renames the columns by the mapping.

Paramters

rename_dict The name mapping

Source code in python/polars_ds/pipeline/pipeline.py

def rename(self, rename_dict: Dict[str, str]) -> Self:
    """
    Renames the columns by the mapping.

    Paramters
    ---------
    rename_dict
        The name mapping
    """
    old = list(rename_dict.keys())
    self._steps.append(
        ExprStep([pl.col(k).alias(v) for k, v in rename_dict.items()], PLContext.WITH_COLUMNS)
    )

    return self.drop([o for o in old if o not in set(rename_dict.values())])

robust_scale(cols, q_low, q_high)

Performs robust scaling on the given columns

Paramters

cols Any Polars expression that can be understood as columns. q_low The lower quantile value q_high The higher quantile value

Source code in python/polars_ds/pipeline/pipeline.py

def robust_scale(self, cols: IntoExprColumn, q_low: float, q_high: float) -> Self:
    """
    Performs robust scaling on the given columns

    Paramters
    ---------
    cols
        Any Polars expression that can be understood as columns.
    q_low
        The lower quantile value
    q_high
        The higher quantile value
    """
    self._steps.append(
        FitStep(partial(t.robust_scale, q_low=q_low, q_high=q_high), cols, self.exclude)
    )
    return self

scale(cols, method='standard')

Scales values in the given columns.

Parameters:

Name	Type	Description	Default
cols	IntoExprColumn	Any Polars expression that can be understood as columns.	required
method	SimpleScaleMethod	One of standard, min_max, abs_max	'standard'

Source code in python/polars_ds/pipeline/pipeline.py

def scale(self, cols: IntoExprColumn, method: SimpleScaleMethod = "standard") -> Self:
    """
    Scales values in the given columns.

    Parameters
    ----------
    cols
        Any Polars expression that can be understood as columns.
    method
        One of `standard`, `min_max`, `abs_max`
    """
    self._steps.append(FitStep(partial(t.scale, method=method), cols, self.exclude))
    return self

select(*cols)

Selects the columns from the dataset.

Paramters

cols Any Polars expression that can be understood as columns.

Source code in python/polars_ds/pipeline/pipeline.py

def select(self, *cols: IntoExprColumn) -> Self:
    """
    Selects the columns from the dataset.

    Paramters
    ---------
    cols
        Any Polars expression that can be understood as columns.
    """
    self._steps.append(ExprStep(list(cols), PLContext.SELECT))
    return self

select_by_std(min_, max_)

Fits and keeps columns that have standard deviation between min_ and max_. Non-numeric columns will always be selected. Only numeric columns with std outside the min_ max_ bounds will be removed. This will never exclude the target if target is set.

Parameters:

Name	Type	Description	Default
min_	float	Min standard deviation to select, inclusive.	required
max_	float	Max standard deviation to select, exclusive	required

Source code in python/polars_ds/pipeline/pipeline.py

def select_by_std(self, min_: float, max_: float) -> Self:
    """
    Fits and keeps columns that have standard deviation between `min_` and `max_`.
    Non-numeric columns will always be selected. Only numeric columns with std outside
    the `min_` `max_` bounds will be removed. This will never exclude the target if target
    is set.

    Parameters
    ----------
    min_
        Min standard deviation to select, inclusive.
    max_
        Max standard deviation to select, exclusive
    """
    self._steps.append(
        FitStep(
            partial(t.select_by_std, min_=min_, max_=max_),
            pl.col("*"),
            self.exclude,
            PLContext.SELECT,
        )
    )
    return self

sort(by, descending, maintain_order=True)

Sorts the dataframe by the columns.

Parameters:

Name	Type	Description	Default
by	IntoExprColumn	The columns to sort by	required
descending	bool | List[bool]	Whether the sort should be descending for the corresponding sort column	required
maintain_order	bool	Whether the sort should maintain order or not	True

Source code in python/polars_ds/pipeline/pipeline.py

def sort(
    self, by: IntoExprColumn, descending: bool | List[bool], maintain_order: bool = True
) -> Self:
    """Sorts the dataframe by the columns.

    Parameters
    ----------
    by
        The columns to sort by
    descending
        Whether the sort should be descending for the corresponding sort column
    maintain_order
        Whether the sort should maintain order or not
    """
    self._steps.append(SortStep(by=by, descending=descending, maintain_order=maintain_order))
    return self

sql_transform(sql)

Runs the SQL on the dataframe when it reaches this step. The user must ensure that the SQL is valid Polars SQL and all columns referred in the SQL exist at this point. The name "df" should be used to refer to the current state of the dataframe in the SQL. E.g. select * from df where A is True.

Parameters:

Name	Type	Description	Default
sql	str	The SQL to run on the dataframe. Note: this step doesn't immedinately check the validity of the SQL statement.	required

Source code in python/polars_ds/pipeline/pipeline.py

def sql_transform(self, sql: str) -> Self:
    """
    Runs the SQL on the dataframe when it reaches this step. The user must ensure that
    the SQL is valid Polars SQL and all columns referred in the SQL exist at this point.
    The name "df" should be used to refer to the current state of the dataframe in the SQL.
    E.g. select * from df where A is True.

    Parameters
    ----------
    sql
        The SQL to run on the dataframe. Note: this step doesn't immedinately check the validity of
        the SQL statement.
    """
    self._steps.append(SQLStep(sql_str=sql))
    return self

target_encode(cols=None, target=None, min_samples_leaf=20, smoothing=10.0, default=None)

Target encode the given variables.

Note: nulls will be encoded as well.

Parameters:

Name	Type	Description	Default
cols	IntoExprColumn | None	Any Polars expression that can be understood as columns. Columns of type != string/categorical will not produce any expression. If None, all string/categorical columns will be used.	None
target	str | Expr | None	The target column	None
min_samples_leaf	int	A regularization factor	20
smoothing	float	Smoothing effect to balance categorical average vs prior	10.0
default	EncoderDefaultStrategy | float | None	If a new value is encountered during transform (unseen in training dataset), it will be mapped to default. If this is a string, it can be null, zero, or mean, where mean means map them to the mean of the target.	None

Reference

https://contrib.scikit-learn.org/category_encoders/targetencoder.html

Source code in python/polars_ds/pipeline/pipeline.py

def target_encode(
    self,
    cols: IntoExprColumn | None = None,
    target: str | pl.Expr | None = None,
    min_samples_leaf: int = 20,
    smoothing: float = 10.0,
    default: EncoderDefaultStrategy | float | None = None,
) -> Self:
    """
    Target encode the given variables.

    Note: nulls will be encoded as well.

    Parameters
    ----------
    cols
        Any Polars expression that can be understood as columns. Columns of type != string/categorical
        will not produce any expression. If None, all string/categorical columns will be used.
    target
        The target column
    min_samples_leaf
        A regularization factor
    smoothing
        Smoothing effect to balance categorical average vs prior
    default
        If a new value is encountered during transform (unseen in training dataset), it will be mapped to default.
        If this is a string, it can be `null`, `zero`, or `mean`, where `mean` means map them to the mean of the target.

    Reference
    ---------
    https://contrib.scikit-learn.org/category_encoders/targetencoder.html
    """
    self._steps.append(
        FitStep(
            partial(
                t.target_encode,
                target=self._get_target(target),
                min_samples_leaf=min_samples_leaf,
                smoothing=smoothing,
                default=default,
            ),
            cols if cols is not None else cs.string() | cs.categorical(),
            self.exclude,
        )
    )
    return self

transform(df, **kwargs)

Fits the blueprint with the dataframe that it is initialized with, and transforms the input dataframe.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Any Polars dataframe	required
**kwargs		Will be passed to Pipeline's transform method.	{}

Source code in python/polars_ds/pipeline/pipeline.py

def transform(self, df: PolarsFrame, **kwargs) -> pl.DataFrame:
    """
    Fits the blueprint with the dataframe that it is initialized with, and
    transforms the input dataframe.

    Parameters
    ----------
    df
        Any Polars dataframe
    **kwargs
        Will be passed to Pipeline's `transform` method.
    """
    return self.materialize().transform(df, **kwargs)

winsorize(cols, q_low=0.05, q_high=0.95, method='nearest')

Learns the lower and upper percentile from the columns, then clip each end at those values. If you wish to clip by constant values, you may append expression like pl.col(c).clip(c1, c2), where c1 and c2 are constants decided by the user.

Parameters:

Name	Type	Description	Default
cols	IntoExprColumn	Any Polars expression that can be understood as columns. Columns must be numerical.	required
q_low	float	The lower quantile value	0.05
q_high	float	The higher quantile value	0.95
method	QuantileMethod	Method to compute quantile. One of nearest, higher, lower, midpoint, linear.	'nearest'

Source code in python/polars_ds/pipeline/pipeline.py

def winsorize(
    self,
    cols: IntoExprColumn,
    q_low: float = 0.05,
    q_high: float = 0.95,
    method: QuantileMethod = "nearest",
) -> Self:
    """
    Learns the lower and upper percentile from the columns, then clip each end at those values.
    If you wish to clip by constant values, you may append expression like pl.col(c).clip(c1, c2),
    where c1 and c2 are constants decided by the user.

    Parameters
    ----------
    cols
        Any Polars expression that can be understood as columns. Columns must be numerical.
    q_low
        The lower quantile value
    q_high
        The higher quantile value
    method
        Method to compute quantile. One of `nearest`, `higher`, `lower`, `midpoint`, `linear`.
    """
    self._steps.append(
        FitStep(
            partial(t.winsorize, q_low=q_low, q_high=q_high, method=method),
            cols,
            self.exclude,
        )
    )
    return self

with_columns(*exprs)

Run Polars with_columns for the expressions.

Source code in python/polars_ds/pipeline/pipeline.py

def with_columns(self, *exprs: pl.Expr) -> Self:
    """
    Run Polars with_columns for the expressions.
    """
    self._steps.append(ExprStep(list(exprs), PLContext.WITH_COLUMNS))
    return self

woe_encode(cols=None, target=None, default=None)

Use Weight of Evidence to encode a discrete variable x with respect to target. This assumes x is discrete and castable to String. A value of 1 is added to all events/non-events (goods/bads) to smooth the computation. This is -1 * output of the package category_encoder's WOEEncoder.

Note: nulls will be encoded as well.

Parameters:

Name	Type	Description	Default
cols	IntoExprColumn | None	Any Polars expression that can be understood as columns. Columns of type != string/categorical will not produce any expression. If None, all string/categorical columns will be used.	None
target	str | Expr | None	The target column	None
default	EncoderDefaultStrategy | float | None	If a new value is encountered during transform (unseen in training dataset), it will be mapped to default. If this is a string, it can be null, zero, or mean, where mean means map them to the mean of the target.	None

Reference

https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html

Source code in python/polars_ds/pipeline/pipeline.py

def woe_encode(
    self,
    cols: IntoExprColumn | None = None,
    target: str | pl.Expr | None = None,
    default: EncoderDefaultStrategy | float | None = None,
) -> Self:
    """
    Use Weight of Evidence to encode a discrete variable x with respect to target. This assumes x
    is discrete and castable to String. A value of 1 is added to all events/non-events
    (goods/bads) to smooth the computation. This is -1 * output of the package category_encoder's WOEEncoder.

    Note: nulls will be encoded as well.

    Parameters
    ----------
    cols
        Any Polars expression that can be understood as columns. Columns of type != string/categorical
        will not produce any expression. If None, all string/categorical columns will be used.
    target
        The target column
    default
        If a new value is encountered during transform (unseen in training dataset), it will be mapped to default.
        If this is a string, it can be `null`, `zero`, or `mean`, where `mean` means map them to the mean of the target.

    Reference
    ---------
    https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html
    """
    self._steps.append(
        FitStep(
            partial(
                t.woe_encode,
                target=self._get_target(target),
                default=default,
            ),
            cols if cols is not None else cs.string() | cs.categorical(),
            self.exclude,
        )
    )
    return self

ExprStep

Container for either one of these polars transforms:

1. df.select(...)
2. df.with_columns(...)
3. df.filter(...)

Source code in python/polars_ds/pipeline/_step.py

class ExprStep:
    """
    Container for either one of these polars transforms:

    1. df.select(...)
    2. df.with_columns(...)
    3. df.filter(...)
    """

    __slots__ = ("exprs", "context")

    def __init__(self, exprs: pl.Expr | Sequence[pl.Expr], context: str | PLContext):
        self.context: PLContext = context if isinstance(context, PLContext) else PLContext(context)
        if isinstance(exprs, pl.Expr):
            self.exprs = [exprs]
        elif isinstance(exprs, str):
            self.exprs = [pl.col(exprs)]
        elif isinstance(exprs, list):
            self.exprs = [to_expr(e) for e in exprs]
        else:
            raise ValueError(
                "A pipeline step must be either an expression or a list of expressions."
            )

    @staticmethod
    def from_partial_dict(d: dict) -> "ExprStep":
        context, json_exprs = d["context"], d["exprs"]
        step_context = PLContext(context)
        if step_context in (PLContext.SELECT, PLContext.WITH_COLUMNS, PLContext.EXPLODE):
            exprs = [pl.Expr.deserialize(StringIO(e), format="json") for e in json_exprs]
        elif step_context == PLContext.FILTER:
            exprs = [pl.Expr.deserialize(StringIO(json_exprs), format="json")]
        elif step_context == PLContext.SQL:
            exprs = str(json_exprs)  # SQL context json_exprs is just a str
        else:
            raise ValueError("Input is not a valid PDS pipeline.")

        return ExprStep(exprs=exprs, context=step_context)

    def to_json(self) -> str:
        ctx = self.context
        d = ctx.get_context_dict()
        if ctx in (PLContext.SELECT, PLContext.WITH_COLUMNS, PLContext.EXPLODE):
            d["exprs"] = [e.meta.serialize(format="json") for e in self.exprs]
        elif self.context == PLContext.SQL:
            d["exprs"] = self.exprs[0]
        elif self.context == PLContext.FILTER:
            d["exprs"] = self.exprs[0].meta.serialize(format="json")
        else:  # Should never reach here
            raise ValueError(f"Unknown context: {self.context}")

        return json.dumps(d)

    def apply_df(self, df: pl.LazyFrame | pl.DataFrame) -> pl.LazyFrame | pl.DataFrame:
        if self.context == PLContext.SELECT:
            return df.select(self.exprs)
        elif self.context == PLContext.WITH_COLUMNS:
            return df.with_columns(self.exprs)
        elif self.context == PLContext.SQL:
            return pl.SQLContext(df=df, eager=False).execute(self.exprs[0])
        elif self.context == PLContext.FILTER:
            return df.filter(self.exprs[0])
        elif self.context == PLContext.EXPLODE:
            return df.explode(self.exprs)

PLContext

Bases: Enum

Methods:

Name	Description
build_step	Build a step from an already deserialized dict. The dict may contain values of type list[str],

Source code in python/polars_ds/pipeline/_step.py

class PLContext(Enum):
    # Regular ExpreStep (purely expressed-based)
    SELECT = "select"
    WITH_COLUMNS = "with_columns"
    FILTER = "filter"
    EXPLODE = "explode"
    # SQLStep
    SQL = "sql"
    # SortStep
    SORT = "sort"
    # GROUP_BY_AGG
    GROUP_BY_AGG = "group_by_agg"
    # GROUP_BY_DYNAMIC_AGG
    GROUP_BY_DYN_AGG = "group_by_dyn_agg"

    def get_context_dict(self) -> Dict:
        return {"context": self.value}

    def build_step(self, deser_dict: Dict) -> PipelineStep:
        """
        Build a step from an already deserialized dict. The dict may contain values of type list[str],
        which requires further deserialization by polars from str to pl.Expr.
        """
        if self in (PLContext.SELECT, PLContext.WITH_COLUMNS, PLContext.FILTER, PLContext.EXPLODE):
            return ExprStep.from_partial_dict(deser_dict)
        elif self == PLContext.SORT:
            return SortStep.from_partial_dict(deser_dict)
        elif self == PLContext.GROUP_BY_AGG:
            return GroupByAggStep.from_partial_dict(deser_dict)
        elif self == PLContext.GROUP_BY_DYN_AGG:
            return GroupByDynAggStep.from_partial_dict(deser_dict)
        elif self == PLContext.SQL:
            return SQLStep.from_partial_dict(deser_dict)
        else:
            raise ValueError(f"Unknown PLContext: {self}")

build_step(deser_dict)

Build a step from an already deserialized dict. The dict may contain values of type list[str], which requires further deserialization by polars from str to pl.Expr.

Source code in python/polars_ds/pipeline/_step.py

def build_step(self, deser_dict: Dict) -> PipelineStep:
    """
    Build a step from an already deserialized dict. The dict may contain values of type list[str],
    which requires further deserialization by polars from str to pl.Expr.
    """
    if self in (PLContext.SELECT, PLContext.WITH_COLUMNS, PLContext.FILTER, PLContext.EXPLODE):
        return ExprStep.from_partial_dict(deser_dict)
    elif self == PLContext.SORT:
        return SortStep.from_partial_dict(deser_dict)
    elif self == PLContext.GROUP_BY_AGG:
        return GroupByAggStep.from_partial_dict(deser_dict)
    elif self == PLContext.GROUP_BY_DYN_AGG:
        return GroupByDynAggStep.from_partial_dict(deser_dict)
    elif self == PLContext.SQL:
        return SQLStep.from_partial_dict(deser_dict)
    else:
        raise ValueError(f"Unknown PLContext: {self}")

Pipeline dataclass

A ML/data transform pipeline. Pipelines should always come from the materialize call from a blueprint. In other words, a pipeline is a fitted blueprint.

Methods:

Name	Description
ensure_features_io	Whether or not this pipeline should check the features coming in and out during transform.
from_json	Recreates a pipeline from a dictionary created by the to_json call.
to_json	Turns self into a JSON string.
transform	Transforms the df using the learned expressions.

Source code in python/polars_ds/pipeline/pipeline.py

@dataclass
class Pipeline:
    """
    A ML/data transform pipeline. Pipelines should always come from the materialize call from a
    blueprint. In other words, a pipeline is a fitted blueprint.
    """

    name: str
    feature_names_in_: List[str]
    feature_names_out_: List[str]
    transforms: List[PipelineStep]
    ensure_features_in: bool = False
    ensure_features_out: bool = True
    lowercase: bool = False
    uppercase: bool = False

    def __str__(self) -> str:
        return self.transforms.__str__()

    def _get_init_plan(self, df: PolarsFrame) -> pl.LazyFrame:
        """
        Get an initial plan without any pipeline transforms.
        """
        if self.lowercase:
            plan = df.lazy().select(pl.all().name.to_lowercase())
        else:
            if self.uppercase:
                plan = df.lazy().select(pl.all().name.to_uppercase())
            else:
                plan = df.lazy()

        return plan

    def _generate_lazy_plan(self, df: PolarsFrame) -> pl.LazyFrame:
        """
        Generates a lazy plan for the incoming df

        Paramters
        ---------
        df
            If none, create the plan for the df that the pipe is initialized with. Otherwise,
            create the plan for the incoming df.
        """
        plan = self._get_init_plan(df)
        for step in self.transforms:
            plan = step.apply_df(plan)
        return plan

    def with_features_out(self, features: List[str], ensure_features_out: bool = True) -> Self:
        self.feature_names_out_ = list(features)
        self.ensure_features_out = ensure_features_out

    def to_json(self, path: str | None = None, **kwargs) -> str | None:
        """
        Turns self into a JSON string.

        Parameters
        ----------
        path
            If none, will return a json string. If given, this will be used as the path
            to save the pipeline and None will be returned.
        kwargs
            Keyword arguments to Python's default json
        """
        # Maybe support other json package?
        d = {
            "name": str(self.name),
            "feature_names_in_": list(self.feature_names_in_),
            "feature_names_out_": list(self.feature_names_out_),
            "transforms": [step.to_json() for step in self.transforms],
            "ensure_features_in": self.ensure_features_in,
            "ensure_features_out": self.ensure_features_out,
            "lowercase": self.lowercase,
            "uppercase": self.uppercase,
        }
        if path is None:
            return json.dumps(d, **kwargs)
        else:
            with open(path, "w") as f:
                json.dump(d, f)

            return None

    @staticmethod
    def from_json(json_str: str | bytes) -> "Pipeline":
        """
        Recreates a pipeline from a dictionary created by the `to_json` call.
        """
        pipeline_dict: Dict = json.loads(json_str)
        try:
            name: str = pipeline_dict["name"]
            transforms: List[str] = pipeline_dict["transforms"]
            feature_names_in_: List[str] = pipeline_dict["feature_names_in_"]
            feature_names_out_: List[str] = pipeline_dict["feature_names_out_"]
            ensure_features_in: bool = pipeline_dict["ensure_features_in"]
            ensure_features_out: bool = pipeline_dict["ensure_features_out"]
            lowercase: bool = pipeline_dict.get("lowercase", False)
            uppercase: bool = pipeline_dict.get("uppercase", False)
        except Exception as e:
            raise ValueError(f"Input dictionary is missing keywords. Original error: \n{e}")

        return Pipeline(
            name=name,
            feature_names_in_=feature_names_in_,
            feature_names_out_=feature_names_out_,
            transforms=[MakeStep.make(json.loads(step_str)) for step_str in transforms],
            ensure_features_in=ensure_features_in,
            ensure_features_out=ensure_features_out,
            lowercase=lowercase,
            uppercase=uppercase,
        )

    def ensure_features_io(self, ensure_in: bool = True, ensure_out: bool = True) -> Self:
        """
        Whether or not this pipeline should check the features coming in and out during transform.

        Parameters
        ----------
        ensure_in
            If true, the input df (during transform) must have the exact same features
            as when this pipeline was fitted (blueprint.materialize()). Setting this false means the input may
            have additional columns, and so this adds flexibility in the pipeline.
        ensure_out
            If true, only the output features during blueprint.materialize() will be kept at the end of the
            pipeline.
        """
        self.ensure_features_in = ensure_in
        self.ensure_features_out = ensure_out
        return self

    def transform(
        self, df: PolarsFrame, return_lazy: bool = False, set_features_out: bool = False, **kwargs
    ) -> PolarsFrame:
        """
        Transforms the df using the learned expressions.

        Paramters
        ---------
        df
            If none, transform the df that the pipe is initialized with. Otherwise, perform
            the learned transformations on the incoming df.
        return_lazy
            If true, return the lazy plan for the transformations
        set_features_out
            If true, set `self.feature_names_out_` to the output features from this transform run.
        **kwargs
            When return_lazy is True, kwargs will be passed to polars.LazyFrame.collect()
        """
        if self.ensure_features_in:
            if self.lowercase:
                columns = [c.lower() for c in df.lazy().collect_schema().names()]
            elif self.uppercase:
                columns = [c.upper() for c in df.lazy().collect_schema().names()]
            else:
                columns = df.lazy().collect_schema().names()
            extras = [c for c in columns if c not in self.feature_names_in_]
            missing = [c for c in self.feature_names_in_ if c not in columns]
            if len(extras) > 0 or len(missing) > 0:
                raise ValueError(
                    f"Input df doesn't have the features expected. Extra columns: {extras}. Missing columns: {missing}"
                )

        plan = self._generate_lazy_plan(df)
        if self.ensure_features_out:
            plan = plan.select(self.feature_names_out_)

        if set_features_out:
            self.feature_names_out_ = plan.collect_schema().names()

        # Add config here if streaming is needed
        return plan if return_lazy else plan.collect(**kwargs)

ensure_features_io(ensure_in=True, ensure_out=True)

Whether or not this pipeline should check the features coming in and out during transform.

Parameters:

Name	Type	Description	Default
ensure_in	bool	If true, the input df (during transform) must have the exact same features as when this pipeline was fitted (blueprint.materialize()). Setting this false means the input may have additional columns, and so this adds flexibility in the pipeline.	True
ensure_out	bool	If true, only the output features during blueprint.materialize() will be kept at the end of the pipeline.	True

Source code in python/polars_ds/pipeline/pipeline.py

def ensure_features_io(self, ensure_in: bool = True, ensure_out: bool = True) -> Self:
    """
    Whether or not this pipeline should check the features coming in and out during transform.

    Parameters
    ----------
    ensure_in
        If true, the input df (during transform) must have the exact same features
        as when this pipeline was fitted (blueprint.materialize()). Setting this false means the input may
        have additional columns, and so this adds flexibility in the pipeline.
    ensure_out
        If true, only the output features during blueprint.materialize() will be kept at the end of the
        pipeline.
    """
    self.ensure_features_in = ensure_in
    self.ensure_features_out = ensure_out
    return self

from_json(json_str) staticmethod

Recreates a pipeline from a dictionary created by the to_json call.

Source code in python/polars_ds/pipeline/pipeline.py

@staticmethod
def from_json(json_str: str | bytes) -> "Pipeline":
    """
    Recreates a pipeline from a dictionary created by the `to_json` call.
    """
    pipeline_dict: Dict = json.loads(json_str)
    try:
        name: str = pipeline_dict["name"]
        transforms: List[str] = pipeline_dict["transforms"]
        feature_names_in_: List[str] = pipeline_dict["feature_names_in_"]
        feature_names_out_: List[str] = pipeline_dict["feature_names_out_"]
        ensure_features_in: bool = pipeline_dict["ensure_features_in"]
        ensure_features_out: bool = pipeline_dict["ensure_features_out"]
        lowercase: bool = pipeline_dict.get("lowercase", False)
        uppercase: bool = pipeline_dict.get("uppercase", False)
    except Exception as e:
        raise ValueError(f"Input dictionary is missing keywords. Original error: \n{e}")

    return Pipeline(
        name=name,
        feature_names_in_=feature_names_in_,
        feature_names_out_=feature_names_out_,
        transforms=[MakeStep.make(json.loads(step_str)) for step_str in transforms],
        ensure_features_in=ensure_features_in,
        ensure_features_out=ensure_features_out,
        lowercase=lowercase,
        uppercase=uppercase,
    )

to_json(path=None, **kwargs)

Turns self into a JSON string.

Parameters:

Name	Type	Description	Default
path	str | None	If none, will return a json string. If given, this will be used as the path to save the pipeline and None will be returned.	None
kwargs		Keyword arguments to Python's default json	{}

Source code in python/polars_ds/pipeline/pipeline.py

def to_json(self, path: str | None = None, **kwargs) -> str | None:
    """
    Turns self into a JSON string.

    Parameters
    ----------
    path
        If none, will return a json string. If given, this will be used as the path
        to save the pipeline and None will be returned.
    kwargs
        Keyword arguments to Python's default json
    """
    # Maybe support other json package?
    d = {
        "name": str(self.name),
        "feature_names_in_": list(self.feature_names_in_),
        "feature_names_out_": list(self.feature_names_out_),
        "transforms": [step.to_json() for step in self.transforms],
        "ensure_features_in": self.ensure_features_in,
        "ensure_features_out": self.ensure_features_out,
        "lowercase": self.lowercase,
        "uppercase": self.uppercase,
    }
    if path is None:
        return json.dumps(d, **kwargs)
    else:
        with open(path, "w") as f:
            json.dump(d, f)

        return None

transform(df, return_lazy=False, set_features_out=False, **kwargs)

Transforms the df using the learned expressions.

Paramters

df If none, transform the df that the pipe is initialized with. Otherwise, perform the learned transformations on the incoming df. return_lazy If true, return the lazy plan for the transformations set_features_out If true, set self.feature_names_out_ to the output features from this transform run. **kwargs When return_lazy is True, kwargs will be passed to polars.LazyFrame.collect()

Source code in python/polars_ds/pipeline/pipeline.py

def transform(
    self, df: PolarsFrame, return_lazy: bool = False, set_features_out: bool = False, **kwargs
) -> PolarsFrame:
    """
    Transforms the df using the learned expressions.

    Paramters
    ---------
    df
        If none, transform the df that the pipe is initialized with. Otherwise, perform
        the learned transformations on the incoming df.
    return_lazy
        If true, return the lazy plan for the transformations
    set_features_out
        If true, set `self.feature_names_out_` to the output features from this transform run.
    **kwargs
        When return_lazy is True, kwargs will be passed to polars.LazyFrame.collect()
    """
    if self.ensure_features_in:
        if self.lowercase:
            columns = [c.lower() for c in df.lazy().collect_schema().names()]
        elif self.uppercase:
            columns = [c.upper() for c in df.lazy().collect_schema().names()]
        else:
            columns = df.lazy().collect_schema().names()
        extras = [c for c in columns if c not in self.feature_names_in_]
        missing = [c for c in self.feature_names_in_ if c not in columns]
        if len(extras) > 0 or len(missing) > 0:
            raise ValueError(
                f"Input df doesn't have the features expected. Extra columns: {extras}. Missing columns: {missing}"
            )

    plan = self._generate_lazy_plan(df)
    if self.ensure_features_out:
        plan = plan.select(self.feature_names_out_)

    if set_features_out:
        self.feature_names_out_ = plan.collect_schema().names()

    # Add config here if streaming is needed
    return plan if return_lazy else plan.collect(**kwargs)

StepRepr dataclass

A representation of a step

Source code in python/polars_ds/pipeline/pipeline.py

@dataclass
class StepRepr:
    """
    A representation of a step
    """

    name: str
    args: List[Any]
    kwargs: Dict[str, Any]

    @staticmethod
    def from_dict(dictionary: Dict[str, Any]) -> "StepRepr":
        try:
            name: str = dictionary["name"]
            args: List[Any] = dictionary.get("args", [])
            kwargs: Dict[str, Any] = dictionary.get("kwargs", {})
            if not isinstance(name, str):
                raise ValueError("Value of `name` must be a string.")
            if not isinstance(args, list):
                raise ValueError("Value of `args` must be a list.")
            if not isinstance(kwargs, dict):
                raise ValueError("Value of `kwargs` must be a dict.")
            if not all(isinstance(s, str) for s in kwargs.keys()):
                raise ValueError("All keys in `kwargs` must be strings.")
            return StepRepr(name=name, args=args, kwargs=kwargs)
        except Exception as e:
            raise ValueError(f"Keys missing or data type is not expected. Original error: \n{e}")

transforms

This module provides classic ML dataset transforms. Note all functions here are single-use only, meaning that the data learned (e.g. mean value in mean imputation) will not be preserved. For pipeline usage, which preserves the learned values and optimizes the transform query, see pipeline.py.

Functions:

Name	Description
center	Center the given columns so that they will have 0 mean.
conditional_impute	Conditionally imputes values in the given columns. This transform will collect if input is lazy.
impute	Impute null values in the given columns. This transform will collect if input is lazy.
iv_encode	Use Information Value to encode a discrete variable x with respect to target. This assumes x
linear_impute	Imputes the target column by training a simple linear regression using the other features. This will
one_hot_encode	Find the unique values in the string/categorical columns and one-hot encode them. This will NOT
ordinal_encode	Converts numeric / string columns into integer encodings starting from 0 to n-1, where
polynomial_features	Generates polynomial combinations out of the features given, at the given degree.
rank_hot_encode	Given a ranking, e.g. ["bad", "neutral", "good"], where "bad", "neutral" and "good" are values coming
robust_scale	Like min-max scaling, but scales each column by the quantile value at q1 and q2.
scale	Scales values in the given columns. This transform will collect if input is lazy.
select_by_std	Fits and selects from cols columns that have standard deviation between min_ and max_.
target_encode	Target encode the given variables. This will overwrite the columns that will be encoded.
winsorize	Learns the lower and upper percentile from the columns, then clip each end at those values.
woe_encode	Use Weight of Evidence to encode a discrete variable x with respect to target. This assumes x

center(df, cols)

Center the given columns so that they will have 0 mean.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names	required

Source code in python/polars_ds/pipeline/transforms.py

def center(df: PolarsFrame, cols: List[str]) -> ExprTransform:
    """
    Center the given columns so that they will have 0 mean.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names
    """
    means = df.lazy().select(pl.col(cols).mean()).collect().row(0)
    return [pl.col(c) - m for c, m in zip(cols, means) if m != 0.0]

conditional_impute(df, rules_dict, method='mean')

Conditionally imputes values in the given columns. This transform will collect if input is lazy.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
rules_dict	Dict[str, str | Expr]	Dictionary where keys are column names (must be string), and values are SQL/Polars Conditions that when true, those values in the column will be imputed, and the value to impute will be learned on the data where the condition is false.	required
method	SimpleImputeMethod	One of mean, median, mode. If mode, a random value will be chosen if there is a tie.	'mean'

Source code in python/polars_ds/pipeline/transforms.py

def conditional_impute(
    df: PolarsFrame, rules_dict: Dict[str, str | pl.Expr], method: SimpleImputeMethod = "mean"
) -> ExprTransform:
    """
    Conditionally imputes values in the given columns. This transform will collect if input is lazy.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    rules_dict
        Dictionary where keys are column names (must be string), and values are SQL/Polars Conditions
        that when true, those values in the column will be imputed,
        and the value to impute will be learned on the data where the condition is false.
    method
        One of `mean`, `median`, `mode`. If `mode`, a random value will be chosen if there is
        a tie.
    """
    rules_dict = {
        c: (r if isinstance(r, pl.Expr) else pl.sql_expr(r)) for c, r in rules_dict.items()
    }
    cols = list(rules_dict.keys())
    # Learn on the data where the condition is false
    if method == "mean":
        temp = (
            df.lazy()
            .select(*(pl.col(c).filter(rules_dict[c].not_()).mean() for c in rules_dict.keys()))
            .collect()
            .row(0)
        )
        return [
            pl.when(rules_dict[c]).then(m).otherwise(pl.col(c)).alias(c) for c, m in zip(cols, temp)
        ]
    elif method == "median":
        temp = (
            df.lazy()
            .select(*(pl.col(c).filter(rules_dict[c].not_()).median() for c in rules_dict.keys()))
            .collect()
            .row(0)
        )
        return [
            pl.when(rules_dict[c]).then(m).otherwise(pl.col(c)).alias(c) for c, m in zip(cols, temp)
        ]
    elif method == "mode":
        temp = (
            df.lazy()
            .select(
                *(
                    pl.col(c).filter(rules_dict[c].not_()).mode().list.first()
                    for c in rules_dict.keys()
                )
            )
            .collect()
            .row(0)
        )
        return [
            pl.when(rules_dict[c]).then(m).otherwise(pl.col(c)).alias(c) for c, m in zip(cols, temp)
        ]
    else:
        raise ValueError(f"Unknown impute method: `{method}`")

impute(df, cols, method='mean')

Impute null values in the given columns. This transform will collect if input is lazy.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names	required
method	SimpleImputeMethod	One of mean, median, mode. If mode, a random value will be chosen if there is a tie.	'mean'

Source code in python/polars_ds/pipeline/transforms.py

def impute(df: PolarsFrame, cols: List[str], method: SimpleImputeMethod = "mean") -> ExprTransform:
    """
    Impute null values in the given columns. This transform will collect if input is lazy.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names
    method
        One of `mean`, `median`, `mode`. If `mode`, a random value will be chosen if there is
        a tie.
    """
    if method == "mean":
        temp = df.lazy().select(pl.col(cols).mean()).collect().row(0)
        return [pl.col(c).fill_null(m) for c, m in zip(cols, temp)]
    elif method == "median":
        temp = df.lazy().select(pl.col(cols).median()).collect().row(0)
        return [pl.col(c).fill_null(m) for c, m in zip(cols, temp)]
    elif method == "mode":
        temp = df.lazy().select(pl.col(cols).mode().list.first()).collect().row(0)
        return [pl.col(c).fill_null(m) for c, m in zip(cols, temp)]
    else:
        raise ValueError(f"Unknown impute method: `{method}`")

iv_encode(df, cols, /, target, default='null')

Use Information Value to encode a discrete variable x with respect to target. This assumes x is discrete and castable to String. A value of 1 is added to all events/non-events (goods/bads) to smooth the computation.

Note: Nulls will always be mapped to the default.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names. Columns of type != string/categorical will not produce any expression.	required
target	str | Expr | Series	The target column	required
default	EncoderDefaultStrategy | float | None	If a new value is encountered during transform (unseen in training dataset), it will be mapped to default. If this is a string, it can be null, zero, or mean, where mean means map them to the mean of the target.	'null'

Reference

https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html

Source code in python/polars_ds/pipeline/transforms.py

def iv_encode(
    df: PolarsFrame,
    cols: List[str],
    /,
    target: str | pl.Expr | pl.Series,
    default: EncoderDefaultStrategy | float | None = "null",
) -> ExprTransform:
    """
    Use Information Value to encode a discrete variable x with respect to target. This assumes x
    is discrete and castable to String. A value of 1 is added to all events/non-events
    (goods/bads) to smooth the computation.

    Note: Nulls will always be mapped to the default.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names. Columns of type != string/categorical will not produce any expression.
    target
        The target column
    default
        If a new value is encountered during transform (unseen in training dataset), it will be mapped to default.
        If this is a string, it can be `null`, `zero`, or `mean`, where `mean` means map them to the mean of the target.

    Reference
    ---------
    https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html
    """
    temp = df.lazy()
    valid_cols = temp.select(cols).select(cs.string() | cs.categorical()).collect_schema().names()

    if len(valid_cols) == 0:
        raise ValueError(
            "The provided columns are either not string/categorical type, or are not in df."
        )

    default_value = _encoder_default_value(temp, default=default, target=target)

    temp = temp.select(
        pds_num.info_value_discrete(c, target, return_sum=False).implode() for c in valid_cols
    ).collect()  # add collect config..
    # POLARS_V1
    return [
        # c[0] will be a series of struct because of the implode above.
        pl.col(c.name).replace_strict(
            old=c[0].struct.field("value"), new=c[0].struct.field("iv"), default=default_value
        )
        for c in temp.get_columns()
    ]

linear_impute(df, features, target, add_bias=False)

Imputes the target column by training a simple linear regression using the other features. This will cast the target column to f64.

Note: The linear regression will skip nulls whenever there is a null in the features or in the target. Additionally, if NaN or Inf exists in data, the linear regression result may be invalid or an error will be thrown. It is recommended to use this only after imputing and dealing with NaN and Infs for all feature columns first.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
features	List[str]	A list of strings representing column names that will be used as features in the linear regression	required
target	str | Expr	The target column	required
add_bias	bool	Whether to add a bias term to the linear regression	False

Source code in python/polars_ds/pipeline/transforms.py

def linear_impute(
    df: PolarsFrame, features: List[str], target: str | pl.Expr, add_bias: bool = False
) -> ExprTransform:
    """
    Imputes the target column by training a simple linear regression using the other features. This will
    cast the target column to f64.

    Note: The linear regression will skip nulls whenever there is a null in the features or in the target.
    Additionally, if NaN or Inf exists in data, the linear regression result may be invalid or an error
    will be thrown. It is recommended to use this only after imputing and dealing with NaN and Infs for
    all feature columns first.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    features
        A list of strings representing column names that will be used as features in the linear regression
    target
        The target column
    add_bias
        Whether to add a bias term to the linear regression
    """
    target_name = df.lazy().select(target).collect_schema().names()[0]
    features_as_expr = [pl.col(f) for f in features]
    target_as_expr = pl.col(target_name)
    temp = (
        df.lazy()
        .select(
            lr.lin_reg(
                *features_as_expr, target=target_as_expr, add_bias=add_bias, null_policy="skip"
            )
        )
        .collect()
    )  # Add streaming config
    coeffs = temp.item(0, 0)  # coeffs is a series
    linear_eq = [f * coeffs[i] for i, f in enumerate(features_as_expr)]
    if add_bias:
        linear_eq.append(pl.lit(coeffs[-1], dtype=pl.Float64))

    return [pl.col(target_name).fill_null(pl.sum_horizontal(linear_eq)).alias(target_name)]

one_hot_encode(df, cols, separator='_', drop_first=False)

Find the unique values in the string/categorical columns and one-hot encode them. This will NOT consider nulls as one of the unique values. Append a one-hot null indicator if you want to encode nulls.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names. Columns of type != string/categorical will not produce any expression.	required
separator	str	E.g. if column name is col and a is an elemenet in it, then the one-hot encoded column will be called col_a where the separator _ is used.	'_'
drop_first	bool	Whether to drop the first distinct value (in terms of str/categorical order). This helps with reducing dimension and prevents some issues from linear dependency.	False

Source code in python/polars_ds/pipeline/transforms.py

def one_hot_encode(
    df: PolarsFrame, cols: List[str], separator: str = "_", drop_first: bool = False
) -> ExprTransform:
    """
    Find the unique values in the string/categorical columns and one-hot encode them. This will NOT
    consider nulls as one of the unique values. Append a one-hot null indicator if you want to encode nulls.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names. Columns of type != string/categorical will not produce any expression.
    separator
        E.g. if column name is `col` and `a` is an elemenet in it, then the one-hot encoded column will be called
        `col_a` where the separator `_` is used.
    drop_first
        Whether to drop the first distinct value (in terms of str/categorical order). This helps with reducing
        dimension and prevents some issues from linear dependency.
    """

    temp = (
        df.lazy()
        .select(cols)
        .select(
            (cs.string() | cs.categorical())
            .unique()
            .drop_nulls()
            .cast(pl.String)
            .implode()
            .list.sort()
        )
    )
    exprs: list[pl.Expr] = []
    for t in temp.collect().get_columns():
        u: pl.Series = t[0]  # t is a Series which contains a single series, so u is a series
        if len(u) > 1:
            # Need to take care of the case where null == 1 is null. Need only True and False, not null
            exprs.extend(
                pl.col(t.name).eq_missing(u[i]).cast(pl.UInt8).alias(t.name + separator + u[i])
                for i in range(int(drop_first), len(u))
            )

    if len(exprs) == 0:
        raise ValueError(
            "Provided columns either do not exist or are not string/categorical types."
        )

    return exprs

ordinal_encode(df, cols, unknown_value=None, null_value=None)

Converts numeric / string columns into integer encodings starting from 0 to n-1, where n is the number of distinct values in the column.

Note: if assigning an integer based on sorting does not make sense for your use case, then please directly use pl.col().replace_strict(mapping)

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names.	required
unknown_value	float | None	What to assign to values not present in training dataset. None means null. Must be a number.	None
null_value	float | None	What to assign to values that are null. Must be a number.	None

Source code in python/polars_ds/pipeline/transforms.py

def ordinal_encode(
    df: PolarsFrame,
    cols: List[str],
    unknown_value: float | None = None,
    null_value: float | None = None,
) -> ExprTransform:
    """
    Converts numeric / string columns into integer encodings starting from 0 to n-1, where
    n is the number of distinct values in the column.

    Note: if assigning an integer based on sorting does not make sense for
    your use case, then please directly use `pl.col().replace_strict(mapping)`

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names.
    unknown_value
        What to assign to values not present in training dataset. None means null.
        Must be a number.
    null_value
        What to assign to values that are null. Must be a number.
    """
    temp = (
        df.lazy()
        .select(cols)
        .select(
            (cs.string() | cs.categorical() | cs.numeric() | cs.boolean())
            .unique()
            .drop_nulls()
            .sort()
            .implode()
        )
    )
    exprs: list[pl.Expr] = []
    for t in temp.collect().get_columns():
        # u is a sorted series
        u: pl.Series = t[0]  # old
        new = pl.int_range(0, len(u), eager=True).cast(pl.Float64)
        c = (
            pl.when(pl.col(t.name).is_null())
            .then(pl.lit(null_value, dtype=pl.Float64))
            .otherwise(
                pl.col(t.name).replace_strict(
                    old=u,
                    new=new,
                    default=pl.lit(unknown_value, dtype=pl.Float64),
                    return_dtype=pl.Float64,
                )
            )
            .alias(t.name)
        )

        exprs.append(c)

    return exprs

polynomial_features(cols, /, degree, interaction_only=False)

Generates polynomial combinations out of the features given, at the given degree.

Parameters:

Name	Type	Description	Default
cols	List[str]	A list of strings representing column names.	required
degree	int	The degree of the polynomial combination	required
interaction_only	bool	It true, only combinations that involve 2 or more variables will be used.	False

Source code in python/polars_ds/pipeline/transforms.py

def polynomial_features(
    cols: List[str],
    /,
    degree: int,
    interaction_only: bool = False,
) -> ExprTransform:
    """
    Generates polynomial combinations out of the features given, at the given degree.

    Parameters
    ----------
    cols
        A list of strings representing column names.
    degree
        The degree of the polynomial combination
    interaction_only
        It true, only combinations that involve 2 or more variables will be used.
    """
    from itertools import combinations_with_replacement

    if degree <= 1:
        raise ValueError("Degree should be > 1.")

    return list(
        pl.reduce(function=lambda acc, x: acc * x, exprs=list(comb)).alias("*".join(comb))
        for comb in combinations_with_replacement(cols, degree)
        if ((not interaction_only) or len(set(comb)) > 1)
    )

rank_hot_encode(col, ranking, default_rank=None)

Given a ranking, e.g. ["bad", "neutral", "good"], where "bad", "neutral" and "good" are values coming from the column col, this will return two new columns, the first is ">=neutral", which will be 1 for all values in ("neutral", "good") and 0 otherwise, and the second new column is ">=good", which will be 1 for all values in ("good") and 0 otherwise.

This currently only works on string columns.

Values not in the provided ranking will have -1 in all the new columns.

Parameters:

Name	Type	Description	Default
col	str	The name of a single column	required
ranking	List[str]	A list of string representing the ranking of the values	required
default_rank	int | None	Default rank for all null/unseen values	None

Source code in python/polars_ds/pipeline/transforms.py

def rank_hot_encode(
    col: str,
    ranking: List[str],
    default_rank: int | None = None,
) -> ExprTransform:
    """
    Given a ranking, e.g. ["bad", "neutral", "good"], where "bad", "neutral" and "good" are values coming
    from the column `col`, this will return two new columns, the first is ">=neutral", which
    will be 1 for all values in ("neutral", "good") and 0 otherwise, and the second new column is ">=good", which
    will be 1 for all values in ("good") and 0 otherwise.

    This currently only works on string columns.

    Values not in the provided ranking will have -1 in all the new columns.

    Parameters
    ----------
    col
        The name of a single column
    ranking
        A list of string representing the ranking of the values
    default_rank
        Default rank for all null/unseen values
    """

    n_ranks = len(ranking)
    if n_ranks <= 1:
        raise ValueError("Rank hot encoding does not work with single value ranking.")

    number_rank = pl.int_range(0, n_ranks, eager=True, dtype=pl.Int32)
    ranked_expr = pl.col(col).replace_strict(
        old=ranking, new=number_rank, default=default_rank, return_dtype=pl.Int32
    )

    return [
        (ranked_expr >= i).cast(pl.Int8).alias(f"{col}>={c}")
        for i, c in zip(range(1, n_ranks), ranking[1:])
    ]

robust_scale(df, cols, q_low=0.25, q_high=0.75, method='midpoint')

Like min-max scaling, but scales each column by the quantile value at q1 and q2.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names	required
q_low	float	The lower quantile value	0.25
q_high	float	The higher quantile value	0.75
method	QuantileMethod	Method to compute quantile. One of nearest, higher, lower, midpoint, linear.	'midpoint'

Source code in python/polars_ds/pipeline/transforms.py

def robust_scale(
    df: PolarsFrame,
    cols: List[str],
    q_low: float = 0.25,
    q_high: float = 0.75,
    method: QuantileMethod = "midpoint",
) -> ExprTransform:
    """
    Like min-max scaling, but scales each column by the quantile value at q1 and q2.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names
    q_low
        The lower quantile value
    q_high
        The higher quantile value
    method
        Method to compute quantile. One of `nearest`, `higher`, `lower`, `midpoint`, `linear`.
    """
    if q_low > 1.0 or q_low < 0.0 or q_high > 1.0 or q_high < 0.0 or q_low >= q_high:
        raise ValueError(
            "Input `q_low` and `q_high` must be between 0 and 1 and q_low must be < than q_high."
        )

    temp = (
        df.lazy()
        .select(
            pl.col(cols).quantile(q_low, interpolation=method).name.prefix("q1:"),
            pl.col(cols).quantile(q_high, interpolation=method).name.prefix("q2:"),
        )
        .collect()
        .row(0)
    )
    n = len(cols)
    return [(pl.col(c) - temp[i]) / (temp[n + i] - temp[i]) for i, c in enumerate(cols)]

scale(df, cols, method='standard')

Scales values in the given columns. This transform will collect if input is lazy.

Note: if method = 'min_max' and min = max, or method = 'standard' and std = 0.0, or method = 'abs_max' and |min| = |max| = 0.0, then the column(s) will not be transformed.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names	required
method	SimpleScaleMethod	One of standard, min_max, abs_max	'standard'

Source code in python/polars_ds/pipeline/transforms.py

def scale(
    df: PolarsFrame,
    cols: List[str],
    method: SimpleScaleMethod = "standard",
) -> ExprTransform:
    """
    Scales values in the given columns. This transform will collect if input is lazy.

    Note: if method = 'min_max' and min = max, or method = 'standard' and std = 0.0, or method = 'abs_max' and |min| = |max| = 0.0,
    then the column(s) will not be transformed.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names
    method
        One of `standard`, `min_max`, `abs_max`
    """
    if method == "standard":
        temp = (
            df.lazy()
            .select(
                pl.col(cols).mean().name.prefix("mean:"),
                pl.col(cols).std(ddof=0).name.prefix("std:"),
            )
            .collect()
            .row(0)
        )
        n = len(cols)
        # do nothing if std = 0.0
        return [
            (pl.col(c) - temp[i]) / temp[i + n] for i, c in enumerate(cols) if temp[i + n] != 0.0
        ]
    elif method == "min_max":
        temp = (
            df.lazy()
            .select(
                pl.col(cols).min().name.prefix("min:"),
                pl.col(cols).max().name.prefix("max:"),
            )
            .collect()
            .row(0)
        )
        n = len(cols)
        # If input is constant, this will return all NaNs.
        # If min = max we don't do anything to the column
        return [
            (pl.col(c) - temp[i]) / (temp[n + i] - temp[i])
            for i, c in enumerate(cols)
            if temp[n + i] != temp[i]
        ]
    elif method == "abs_max":
        temp = (
            df.lazy()
            .select(pl.max_horizontal(pl.col(c).min().abs(), pl.col(c).max().abs()) for c in cols)
            .collect()
            .row(0)
        )
        return [pl.col(c) / m for c, m in zip(cols, temp) if m != 0.0]
    else:
        raise ValueError(f"Unknown input method: {method}")

select_by_std(df, cols, /, min_, max_)

Fits and selects from cols columns that have standard deviation between min_ and max_. Non-numeric columns in cols will always be selected.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names.	required
min_	float	Min standard deviation to select, inclusive.	required
max_	float	Max standard deviation to select, exclusive	required

Source code in python/polars_ds/pipeline/transforms.py

def select_by_std(
    df: PolarsFrame,
    cols: List[str],
    /,
    min_: float,
    max_: float,
) -> ExprTransform:
    """
    Fits and selects from `cols` columns that have standard deviation between `min_` and `max_`.
    Non-numeric columns in `cols` will always be selected.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names.
    min_
        Min standard deviation to select, inclusive.
    max_
        Max standard deviation to select, exclusive
    """
    numeric = df.lazy().select(cs.by_name(cols) & cs.numeric()).collect_schema().names()
    std_vals = df.lazy().select(pl.col(numeric).std()).collect().row(0)
    # Do an exclusion instead
    return pl.exclude([c for c, v in zip(numeric, std_vals) if (v < min_ or v > max_)])

target_encode(df, cols, /, target, min_samples_leaf=20, smoothing=10.0, default='null')

Target encode the given variables. This will overwrite the columns that will be encoded.

Note: Nulls will always be mapped to the default.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names. Columns of type != string/categorical will not produce any expression.	required
target	str | Expr | Series	The target column	required
min_samples_leaf	int	A regularization factor	20
smoothing	float	Smoothing effect to balance categorical average vs prior	10.0
default	EncoderDefaultStrategy | float | None	If a new value is encountered during transform (unseen in training dataset), it will be mapped to default. If this is a string, it can be null, zero, or mean, where mean means map them to the mean of the target.	'null'

Reference

https://contrib.scikit-learn.org/category_encoders/targetencoder.html

Source code in python/polars_ds/pipeline/transforms.py

def target_encode(
    df: PolarsFrame,
    cols: List[str],
    /,
    target: str | pl.Expr | pl.Series,
    min_samples_leaf: int = 20,
    smoothing: float = 10.0,
    default: EncoderDefaultStrategy | float | None = "null",
) -> ExprTransform:
    """
    Target encode the given variables. This will overwrite the columns that will be encoded.

    Note: Nulls will always be mapped to the default.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names. Columns of type != string/categorical will not produce any expression.
    target
        The target column
    min_samples_leaf
        A regularization factor
    smoothing
        Smoothing effect to balance categorical average vs prior
    default
        If a new value is encountered during transform (unseen in training dataset), it will be mapped to default.
        If this is a string, it can be `null`, `zero`, or `mean`, where `mean` means map them to the mean of the target.

    Reference
    ---------
    https://contrib.scikit-learn.org/category_encoders/targetencoder.html
    """
    temp = df.lazy()
    valid_cols = temp.select(cols).select(cs.string() | cs.categorical()).collect_schema().names()

    if len(valid_cols) == 0:
        raise ValueError(
            "The provided columns are either not string/categorical type, or are not in df."
        )

    default_value = _encoder_default_value(temp, default=default, target=target)

    temp = temp.select(
        pds_num.target_encode(
            c, target, min_samples_leaf=min_samples_leaf, smoothing=smoothing
        ).implode()
        for c in valid_cols
    ).collect()  # add collect config..
    return [
        # c[0] will be a series of struct because of the implode above.
        pl.col(c.name).replace_strict(
            old=c[0].struct.field("value"), new=c[0].struct.field("to"), default=default_value
        )
        for c in temp.get_columns()
    ]

winsorize(df, cols, q_low=0.05, q_high=0.95, method='nearest')

Learns the lower and upper percentile from the columns, then clip each end at those values.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names	required
q_low	float	The lower quantile value	0.05
q_high	float	The higher quantile value	0.95
method	QuantileMethod	Method to compute quantile. One of nearest, higher, lower, midpoint, linear.	'nearest'

Source code in python/polars_ds/pipeline/transforms.py

def winsorize(
    df: PolarsFrame,
    cols: List[str],
    q_low: float = 0.05,
    q_high: float = 0.95,
    method: QuantileMethod = "nearest",
) -> ExprTransform:
    """
    Learns the lower and upper percentile from the columns, then clip each end at those values.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names
    q_low
        The lower quantile value
    q_high
        The higher quantile value
    method
        Method to compute quantile. One of `nearest`, `higher`, `lower`, `midpoint`, `linear`.
    """
    if q_low > 1.0 or q_low < 0.0 or q_high > 1.0 or q_high < 0.0 or q_low >= q_high:
        raise ValueError(
            "Input `q_low` and `q_high` must be between 0 and 1 and q_low must be < than q_high."
        )

    temp = (
        df.lazy()
        .select(
            pl.col(cols).quantile(q_low, interpolation=method).name.prefix("l"),
            pl.col(cols).quantile(q_high, interpolation=method).name.prefix("u"),
        )
        .collect()
        .row(0)
    )
    n = len(cols)
    return [pl.col(c).clip(temp[i], temp[n + i]) for i, c in enumerate(cols)]

woe_encode(df, cols, /, target, default='null')

Use Weight of Evidence to encode a discrete variable x with respect to target. This assumes x is discrete and castable to String. A value of 1 is added to all events/non-events (goods/bads) to smooth the computation. This is -1 * output of the package category_encoder's WOEEncoder.

Note: Nulls will always be mapped to the default.

Parameters:

Name	Type	Description	Default
df	PolarsFrame	Either a lazy or an eager dataframe	required
cols	List[str]	A list of strings representing column names. Columns of type != string/categorical will not produce any expression.	required
target	str | Expr | Series	The target column	required
default	EncoderDefaultStrategy | float | None	If a new value is encountered during transform (unseen in training dataset), it will be mapped to default. If this is a string, it can be null, zero, or mean, where mean means map them to the mean of the target.	'null'

Reference

https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html

Source code in python/polars_ds/pipeline/transforms.py

def woe_encode(
    df: PolarsFrame,
    cols: List[str],
    /,
    target: str | pl.Expr | pl.Series,
    default: EncoderDefaultStrategy | float | None = "null",
) -> ExprTransform:
    """
    Use Weight of Evidence to encode a discrete variable x with respect to target. This assumes x
    is discrete and castable to String. A value of 1 is added to all events/non-events
    (goods/bads) to smooth the computation. This is -1 * output of the package category_encoder's WOEEncoder.

    Note: Nulls will always be mapped to the default.

    Parameters
    ----------
    df
        Either a lazy or an eager dataframe
    cols
        A list of strings representing column names. Columns of type != string/categorical will not produce any expression.
    target
        The target column
    default
        If a new value is encountered during transform (unseen in training dataset), it will be mapped to default.
        If this is a string, it can be `null`, `zero`, or `mean`, where `mean` means map them to the mean of the target.

    Reference
    ---------
    https://www.listendata.com/2015/03/weight-of-evidence-woe-and-information.html
    """
    temp = df.lazy()
    valid_cols = temp.select(cols).select(cs.string() | cs.categorical()).collect_schema().names()

    if len(valid_cols) == 0:
        raise ValueError(
            "The provided columns are either not string/categorical type, or are not in df."
        )

    default_value = _encoder_default_value(temp, default=default, target=target)

    temp = temp.select(
        pds_num.woe_discrete(c, target).implode() for c in valid_cols
    ).collect()  # add collect config..

    return [
        # c[0] will be a series of struct because of the implode above.
        pl.col(c.name).replace_strict(
            old=c[0].struct.field("value"), new=c[0].struct.field("woe"), default=default_value
        )
        for c in temp.get_columns()
    ]