Utilities

Shared building blocks for composing and cleaning — the tools you use around the indicators rather than the indicators themselves.

• BaseIndicator — the reusable primitives every indicator is built from (sma, ema, true_range, check_fillna, get_min_max). Use these when you build a custom indicator so it inherits the same warm-up and fillna behaviour as the built-ins.
• DataCleaner — detect and repair the NaN/inf/null/absurdly-large values that real market data is full of (dropna, get_invalid_indices, approximate_invalid_values) before they cascade through a rolling window and null out a whole column.

Clean first, compute second

A single bad tick inside a rolling window can poison every value that window touches. Run DataCleaner on raw feeds before computing indicators — see the cleaning how-to.

---

polars_ta.utils

BaseIndicator

Utility functions for the Polars TA library.

check_fillna staticmethod

check_fillna(expr: Expr | str, fillna: bool, value: int = 0) -> Expr

Check if fillna flag is True and fill gaps. Replaces inf/-inf with nulls before filling.

Source code in polars_ta/utils.py

@staticmethod
def check_fillna(expr: pl.Expr | str, fillna: bool, value: int = 0) -> pl.Expr:
    """
    Check if fillna flag is True and fill gaps.
    Replaces inf/-inf with nulls before filling.
    """
    expr = as_expr(expr)

    if not fillna:
        return expr

    # Replace Inf, -Inf, and NaN with Polars Null
    clean_expr = (
        pl.when(expr.is_infinite() | expr.is_nan()).then(None).otherwise(expr)
    )

    if value == -1:
        # ffill().bfill() equivalent in Polars
        return clean_expr.forward_fill().backward_fill()
    else:
        # ffill().fillna(value) equivalent in Polars
        return clean_expr.forward_fill().fill_null(value)

true_range staticmethod

true_range(high: Expr | str, low: Expr | str, prev_close: Expr | str) -> Expr

Calculate the True Range using horizontal aggregation.

Source code in polars_ta/utils.py

@staticmethod
def true_range(
    high: pl.Expr | str, low: pl.Expr | str, prev_close: pl.Expr | str
) -> pl.Expr:
    """Calculate the True Range using horizontal aggregation."""
    high = as_expr(high)
    low = as_expr(low)
    prev_close = as_expr(prev_close)

    tr1 = high - low
    tr2 = (high - prev_close).abs()
    tr3 = (low - prev_close).abs()

    return pl.max_horizontal([tr1, tr2, tr3])

sma staticmethod

sma(expr: Expr | str, periods: int, fillna: bool = False) -> Expr

Simple Moving Average

Source code in polars_ta/utils.py

@staticmethod
def sma(expr: pl.Expr | str, periods: int, fillna: bool = False) -> pl.Expr:
    """Simple Moving Average"""
    expr = as_expr(expr)
    min_periods = 1 if fillna else periods
    return expr.rolling_mean(window_size=periods, min_samples=min_periods)

ema staticmethod

ema(expr: Expr | str, periods: int, fillna: bool = False) -> Expr

Exponential Moving Average

Source code in polars_ta/utils.py

@staticmethod
def ema(expr: pl.Expr | str, periods: int, fillna: bool = False) -> pl.Expr:
    """Exponential Moving Average"""
    expr = as_expr(expr)
    min_periods = 1 if fillna else periods
    return expr.ewm_mean(span=periods, adjust=False, min_samples=min_periods)

get_min_max staticmethod

get_min_max(expr1: Expr | str, expr2: Expr | str, function: str = 'min') -> Expr

Find min or max value between two series for each index.

Source code in polars_ta/utils.py

@staticmethod
def get_min_max(
    expr1: pl.Expr | str, expr2: pl.Expr | str, function: str = "min"
) -> pl.Expr:
    """Find min or max value between two series for each index."""
    expr1 = as_expr(expr1)
    expr2 = as_expr(expr2)

    if function == "min":
        return pl.min_horizontal([expr1, expr2])
    elif function == "max":
        return pl.max_horizontal([expr1, expr2])
    else:
        raise ValueError('"function" variable value should be "min" or "max"')

DataCleaner

Methods to handle, track, and heal invalid data before applying TA indicators.

dropna staticmethod

dropna(df: DataFrame) -> DataFrame

Drop rows with nulls, NaNs, or excessively large numbers in numeric columns (safe alternative to the original ta library).

Source code in polars_ta/utils.py

@staticmethod
def dropna(df: pl.DataFrame) -> pl.DataFrame:
    """
    Drop rows with nulls, NaNs, or excessively large numbers
    in numeric columns (safe alternative to the original ta library).
    """
    invalid_mask = DataCleaner._build_invalid_mask(df)
    # Keep rows where invalid_mask is False
    return df.filter(~invalid_mask)

get_invalid_indices staticmethod

get_invalid_indices(df: DataFrame) -> list[int]

Returns a list of integer row indices where invalid data exists. Useful for logging or inspecting anomalies.

Source code in polars_ta/utils.py

@staticmethod
def get_invalid_indices(df: pl.DataFrame) -> list[int]:
    """
    Returns a list of integer row indices where invalid data exists.
    Useful for logging or inspecting anomalies.
    """
    invalid_mask = DataCleaner._build_invalid_mask(df)

    bad_indices = (
        df.with_row_index("row_idx")
        .filter(invalid_mask)
        .get_column("row_idx")
        .to_list()
    )
    return bad_indices

approximate_invalid_values staticmethod

approximate_invalid_values(df: DataFrame) -> DataFrame

Replaces invalid values with Polars Nulls, then approximates them using linear interpolation and forward-filling based on past values.

Source code in polars_ta/utils.py

@staticmethod
def approximate_invalid_values(df: pl.DataFrame) -> pl.DataFrame:
    """
    Replaces invalid values with Polars Nulls, then approximates them
    using linear interpolation and forward-filling based on past values.
    """
    num_cols = df.select(cs.numeric()).columns
    big_number = math.exp(709)

    exprs = []
    for col in num_cols:
        clean_col = (
            pl.when(
                (pl.col(col) >= big_number)
                | pl.col(col).is_nan()
                | pl.col(col).is_infinite()
            )
            .then(None)
            .otherwise(pl.col(col))
        )

        # Interpolate (linear line) and forward_fill (carry last good value)
        imputed_col = clean_col.interpolate().forward_fill()
        exprs.append(imputed_col.alias(col))

    return df.with_columns(exprs)