Skip to content

Class FractalAdaptiveMovingAverage

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

Import

from NitroFE import FractalAdaptiveMovingAverage

FractalAdaptiveMovingAverage

Fractal Adaptive Moving Average Technical Indicator (FRAMA) was developed by John Ehlers. This indicator is constructed based on the algorithm of the Exponential Moving Average, in which the smoothing factor is calculated based on the current fractal dimension of the value series. The advantage of FRAMA is the possibility to follow strong trend movements and to sufficiently slow down at the moments of price consolidation.

Fractal adaptive moving average (FAMA) is calculated as :

\[ N(t-Length,t) = \frac{\max(dataframe[t-Length \to t]) - \min(dataframe[t-Length \to t]) }{(t)-(t-Length)} \]
\[ N1 = N(t-Length/2,t) \]
\[ N2 = N(t-Length,t-Length/2) \]
\[ N3 = N(t-Length,t) \]
\[ D[t] = (\log(N1 + N2) - \log(N3))/\log(2) \]
\[ A[t] = \exp(-4.6 * (D[t] - 1)) \]
\[ FAMA[t] = A[t] * dataframe[t] + (1 - A[t]) * FAMA[t-1] \]

Methods

Provided dataframe must be in ascending order.

__init__(self, lookback_period=8, min_periods=1) special

Parameters:

Name Type Description Default
lookback_period int

Size of the rolling window of lookback , by default 8

 8
min_periods int

Minimum number of observations in window required to have a value, by default 1

 1
Source code in nitrofe\time_based_features\moving_average_features\moving_average_features.py 
def __init__(self, lookback_period: int = 8, min_periods: int = 1):
    """
    Parameters
    ----------
    lookback_period : int, optional
        Size of the rolling window of lookback , by default 8
    min_periods : int, optional
        Minimum number of observations in window required to have a value, by default 1
    """
    self.lookback_period = lookback_period
    self.min_periods = min_periods

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._first_object = weighted_window_features()
        self._second_object = weighted_window_features()
        self._third_object = weighted_window_features()

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

    fama = pd.DataFrame(
        np.zeros(dataframe.shape), columns=dataframe.columns, index=dataframe.index
    )
    if first_fit:
        fama = pd.concat(
            [pd.DataFrame(np.zeros((1, fama.shape[1])), columns=fama.columns), fama]
        )
    else:
        fama = pd.concat([self.values_from_last_run, fama])
    fama["_iloc"] = np.arange(len(fama))
    ll = [x for x in fama.columns if x != "_iloc"]

    first_res = self._first_object._template_feature_calculation(
        function_name="_first_object",
        win_function=_identity_window,
        first_fit=first_fit,
        dataframe=dataframe,
        window=int((self.lookback_period) / 2),
        min_periods=self.min_periods,
        symmetric=None,
        operation=self._first_lb,
        operation_args=(int((self.lookback_period) / 2)),
    )

    second_res = self._second_object._template_feature_calculation(
        function_name="_second_object",
        win_function=_identity_window,
        first_fit=first_fit,
        dataframe=dataframe,
        window=self.lookback_period,
        min_periods=self.min_periods,
        symmetric=None,
        operation=self._second_lb,
        operation_args=(self.lookback_period),
    )

    third_res = self._third_object._template_feature_calculation(
        function_name="_third_object",
        win_function=_identity_window,
        first_fit=first_fit,
        dataframe=dataframe,
        window=self.lookback_period,
        min_periods=self.min_periods,
        symmetric=None,
        operation=self._first_lb,
        operation_args=(self.lookback_period),
    )
    fractal_dimension = (
        np.log(second_res + first_res) - np.log(third_res)
    ) / np.log(2)
    a_value = np.exp(-4.6 * (fractal_dimension - 1)).fillna(0)
    a_value = pd.DataFrame(
        np.where(a_value < 0.01, 0.01, a_value), columns=a_value.columns
    )

    FC = (self.lookback_period) / 2
    SC = self.lookback_period
    oldN = (2 - a_value) / a_value
    newN = ((SC - FC) * (oldN - 1) / (SC - 1)) + FC
    a_value = 2 / (newN + 1)

    for r1, r2, r3 in zip(
        dataframe.iterrows(), fama[1:].iterrows(), a_value.iterrows()
    ):
        previous_kama = fama[fama["_iloc"] == (r2[1]["_iloc"] - 1)][ll]
        fama.loc[fama["_iloc"] == r2[1]["_iloc"], ll] = (
            np.multiply(previous_kama.values, (1 - r3[1].values))
            + np.multiply(r1[1].values, r3[1].values)
        )[0]

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

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

References