Skip to content

Class KaufmanAdaptiveMovingAverage

  • Jump right in for a hands-on Open In Colab

Import

from NitroFE import KaufmanAdaptiveMovingAverage

KaufmanAdaptiveMovingAverage

Kaufman's Adaptive Moving Average (KAMA) is a moving average designed to account for noise or volatility.

KAMA will closely follow values when the values swings are relatively small and the noise is low.

KAMA will adjust when the values swings widen and follow values from a greater distance.

This trend-following indicator can be used to identify the overall trend, time turning points and filter price movements.

Kaufman adaptive moving average (KAMA) is calculated as :

\[ Current\_value[t] = dataframe[t] \]
\[ ER[t] = Kaufman\_Efficiency\_Ratio[t] \]
\[ fast \ ema \ alpha = \frac{2}{1+fast\_ema\_span} \]
\[ slow \ ema \ alpha = \frac{2}{1+slow\_ema\_span} \]
\[ Smoothing\_Constant[t] = (ER[t] * (fast\_ema\_alpha - slow\_ema\_alpha ) + slow\_ema\_alpha)^{2} \]
\[ KAMA[t]= KAMA[t-1] + Smoothing\_Constant[t] * ( Current\_value[t] - KAMA[t-1] ) \]

Methods

Provided dataframe must be in ascending order.

__init__(self, kaufman_efficiency_lookback_period=4, kaufman_efficiency_min_periods=None, fast_ema_span=2, slow_ema_span=5) special

Parameters:

Name Type Description Default
kaufman_efficiency_lookback_period int

Size of the rolling window of lookback for the caluculation of kaufman efficiency ratio , by default 4

 4
kaufman_efficiency_min_periods int

Minimum number of observations in window required to have a value for kaufman efficiency ratio, by default 1

 None
fast_ema_span int

fast span length, by default 2

 2
slow_ema_span int

slow span length, by default 5

 5
Source code in nitrofe\time_based_features\moving_average_features\moving_average_features.py 
def __init__(
    self,
    kaufman_efficiency_lookback_period: int = 4,
    kaufman_efficiency_min_periods: int = None,
    fast_ema_span: int = 2,
    slow_ema_span: int = 5,
):
    """
    Parameters
    ----------
    kaufman_efficiency_lookback_period : int, optional
        Size of the rolling window of lookback for the caluculation of kaufman efficiency ratio , by default 4
    kaufman_efficiency_min_periods : int, optional
        Minimum number of observations in window required to have a value for kaufman efficiency ratio, by default 1
    fast_ema_span : int, optional
        fast span length, by default 2
    slow_ema_span : int, optional
        slow span length, by default 5
    """

    self.kaufman_efficiency_lookback_period = kaufman_efficiency_lookback_period
    self.kaufman_efficiency_min_periods = kaufman_efficiency_min_periods
    self.fast_ema_span = fast_ema_span
    self.slow_ema_span = slow_ema_span

fit(self, dataframe, first_fit=True)

For your training/initial fit phase (very first fit) use fit_first=True, and for any production/test implementation pass fit_first=False

Parameters:

Name Type Description Default
dataframe Union[pandas.core.frame.DataFrame, pandas.core.series.Series]

dataframe containing column values to create feature over

 required
first_fit bool

Moving features require past values for calculation. Use True, when calculating for training data (very first fit) Use False, when calculating for subsequent testing/production data { in which case the values, which were saved during the last phase, will be utilized for calculation }, by default True

 True
Source code in nitrofe\time_based_features\moving_average_features\moving_average_features.py 
def fit(self, dataframe: Union[pd.DataFrame, pd.Series], first_fit: bool = True):
    """
    For your training/initial fit phase (very first fit) use fit_first=True, and for any production/test implementation pass fit_first=False

    Parameters
    ----------
    dataframe : Union[pd.DataFrame, pd.Series]
        dataframe containing column values to create feature over
    first_fit : bool, optional
        Moving features require past values for calculation.
        Use True, when calculating for training data (very first fit)
        Use False, when calculating for subsequent testing/production data { in which case the values, which
        were saved during the last phase, will be utilized for calculation }, by default True
    """

    if first_fit:
        self._kaufman_object = _a_kaufman_efficiency()

    if isinstance(dataframe, pd.Series):
        dataframe = dataframe.to_frame()

    kma = pd.DataFrame(
        np.zeros(dataframe.shape), columns=dataframe.columns, index=dataframe.index
    )
    if first_fit:
        kma = pd.concat(
            [pd.DataFrame(np.zeros((1, kma.shape[1])), columns=kma.columns), kma]
        )
        if self.kaufman_efficiency_min_periods == None:
            _first_pervious = self.kaufman_efficiency_lookback_period - 2
        elif self.kaufman_efficiency_min_periods > 1:
            _first_pervious = self.kaufman_efficiency_min_periods - 2
        else:
            _first_pervious = 0

    else:
        kma = pd.concat([self.values_from_last_run, kma])
        _first_pervious = -1
    kma["_iloc"] = np.arange(len(kma))

    _kaufman_efficiency = self._kaufman_object.fit(
        dataframe=dataframe,
        first_fit=first_fit,
        lookback_period=self.kaufman_efficiency_lookback_period,
        min_periods=self.kaufman_efficiency_min_periods,
    )

    ll = [x for x in kma.columns if x != "_iloc"]
    SC = (
        _kaufman_efficiency.copy()
        * (2 / (self.fast_ema_span + 1) - 2 / (self.slow_ema_span + 1))
        + 2 / (self.slow_ema_span + 1)
    ) ** 2

    if first_fit:
        SC.iloc[_first_pervious] = [0] * SC.shape[1]

    for r1, r2, r3 in zip(
        dataframe[(_first_pervious + 1) :].iterrows(),
        kma[(1 + _first_pervious + 1) :].iterrows(),
        SC[(_first_pervious + 1) :].iterrows(),
    ):

        previous_kama = kma[kma["_iloc"] == (r2[1]["_iloc"] - 1)][ll]

        kma.loc[kma["_iloc"] == r2[1]["_iloc"], ll] = (
            previous_kama
            + np.multiply(r3[1].values, (r1[1].values - previous_kama.values))
        ).values[0]

    res = kma.iloc[1:][ll]

    self.values_from_last_run = res.iloc[-1:]

    return res

References