Skip to content

Super Smoother

SuperSmoother algorithm for time series smoothing.

Implementation based on John Ehlers' SuperSmoother filter, which is a two-pole Butterworth filter that provides excellent smoothing while minimizing lag. The filter uses an adaptive approach with cross-validation to determine the optimal smoothing parameters.

Reference: Ehlers, J. (2013). "Cycle Analytics for Traders"

quantflow.ta.SuperSmoother pydantic-model

Bases: BaseModel

SuperSmoother filter for time series data.

This implementation uses a two-pole Butterworth filter with adaptive smoothing. The SuperSmoother filter is designed to remove high-frequency noise while preserving the underlying trend with minimal lag. The filter is defined by the following recurrence relation:

\[ y_t = c_1 \frac{x_t + x_{t-1}}{2} + c_2 y_{t-1} + c_3 y_{t-2} \]

where the coefficients are calculated as:

\[ \begin{align} \lambda &= \frac{\pi \sqrt{2}}{N} \\ a &= \exp(-\lambda) \\ c_2 &= 2 a \cos(\lambda) \\ c_3 &= -a^2 \\ c_1 &= 1 - c_2 - c_3 \end{align} \]

and \(N\) is the period.

Example

import pandas as pd
smoother = SuperSmoother(period=10)
df = pd.DataFrame({"value": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]})
df["smoothed"] = df["value"].apply(smoother.update)

For online updates:

smoother = SuperSmoother(period=10)
for value in [1, 2, 3, 4, 5]:
    smoothed = smoother(value)
    print(smoothed)

Fields:

period pydantic-field 

period = 10

Number of periods for the smoothing filter (must be >= 2)

raw_value 

raw_value()

Get the most recent raw input value, if available.

Source code in quantflow/ta/supersmoother.py 
68
69
70
def raw_value(self) -> float | None:
    """Get the most recent raw input value, if available."""
    return self._prev_value

value 

value()

Get the most recent smoothed value, if available.

Source code in quantflow/ta/supersmoother.py 
72
73
74
def value(self) -> float | None:
    """Get the most recent smoothed value, if available."""
    return self._prev_smooth1

update 

update(value)

Update the filter with a new value and return the smoothed result.

PARAMETER DESCRIPTION
value

New data point to add to the filter

TYPE: float

Source code in quantflow/ta/supersmoother.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def update(
    self,
    value: Annotated[float, Doc("New data point to add to the filter")],
) -> float:
    """Update the filter with a new value and return the smoothed result."""
    if self._prev_value is None:
        self._prev_value = value
        return value
    elif self._prev_smooth1 is None:
        # Second value, use simple average
        smoothed = (value + self._prev_value) / 2.0
        self._prev_smooth2 = self._prev_value
        self._prev_smooth1 = smoothed
    else:
        # Calculate the average of current and previous value
        avg = (value + self._prev_value) / 2.0
        # Apply filter equation
        smoothed = (
            self._c1 * avg
            + self._c2 * self._prev_smooth1  # type: ignore
            + self._c3 * self._prev_smooth2  # type: ignore
        )
        # Update state for next iteration
        self._prev_smooth2 = self._prev_smooth1
        self._prev_smooth1 = smoothed
    self._prev_value = value
    return smoothed