Skip to content

turtle_quant_1.strategies.mean_reversion.mfi_divergence

turtle_quant_1.strategies.mean_reversion.mfi_divergence

Money Flow Index (MFI) Divergence strategy implementation.

MfiDivergence 

MfiDivergence(
    mfi_period: int = 14, local_extrema_window: int = 5
)

Bases: BaseStrategy

Initialize the MFI Divergence strategy.

Leading / Lagging: Lagging in construction. Leading in interpretation. Lag period (in candles): (mfi_period - 1) / 2 Effect: Delay in recognizing that price has deviated far enough from equilibrium.

Parameters:

Name Type Description Default
mfi_period int

Number of candles for MFI calculation.

 14
local_extrema_window int

Window for local maxima/minima detection.

 5
Source code in src/turtle_quant_1/strategies/mean_reversion/mfi_divergence.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, mfi_period: int = 14, local_extrema_window: int = 5):
    """Initialize the MFI Divergence strategy.

    Leading / Lagging: Lagging in construction. Leading in interpretation.
    Lag period (in candles): (`mfi_period` - 1) / 2
    Effect: Delay in recognizing that price has deviated far enough from equilibrium.

    Args:
        mfi_period: Number of candles for MFI calculation.
        local_extrema_window: Window for local maxima/minima detection.
    """

    super().__init__()
    self.mfi_period = mfi_period
    self.local_extrema_window = local_extrema_window

    if (
        mfi_period
        > convert_units(BACKTESTING_MAX_LOOKBACK_DAYS, "1D", CANDLE_UNIT) * 0.5
    ):
        raise ValueError(
            f"This strategy relies on too many lookback candles ({mfi_period}) for meaningful evaluation. "
            f"Maximum lookback is {BACKTESTING_MAX_LOOKBACK_DAYS} days."
        )

generate_historical_scores 

generate_historical_scores(
    data: DataFrame, symbol: str
) -> Series

Generate historical MFI divergence scores.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to use for the strategy.

 required
symbol str

The symbol to use for the strategy.

 required

Returns:

Type Description
Series

Score array with each value between -1.0 and +1.0, indexed by the original data's index.
Source code in src/turtle_quant_1/strategies/mean_reversion/mfi_divergence.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def generate_historical_scores(self, data: pd.DataFrame, symbol: str) -> pd.Series:
    """Generate historical MFI divergence scores.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to use for the strategy.
        symbol: The symbol to use for the strategy.

    Returns:
        Score array with each value between -1.0 and +1.0, indexed by the original data's index.
    """
    self.validate_data(data)

    close = data["Close"]

    mfi = self._calc_mfi(data)
    maximas, minimas = self._find_extrema(close)
    divergence_signals = self._get_divergence_signals(close, mfi, maximas, minimas)
    if np.isnan(divergence_signals.iloc[-1]):
        raise ValueError("Last score should not be NaN")

    # Optional: Smooth or decay signal over a few candles
    divergence_signals = divergence_signals.ffill(limit=3).fillna(0)

    return pd.Series(
        data=divergence_signals.clip(-1, 1).values,
        index=data.index,
    )

generate_prediction_score 

generate_prediction_score(
    data: DataFrame, symbol: str
) -> float

Generate the prediction score for the latest candle.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to use for the strategy.

 required
symbol str

The symbol to use for the strategy.

 required

Returns:

Type Description
float

Score array with each value between -1.0 and +1.0, indexed by datetime
Source code in src/turtle_quant_1/strategies/mean_reversion/mfi_divergence.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate the prediction score for the latest candle.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to use for the strategy.
        symbol: The symbol to use for the strategy.

    Returns:
        Score array with each value between -1.0 and +1.0, indexed by datetime
    """
    return self.generate_historical_scores(
        data.iloc[-(self.mfi_period + 1) :], symbol
    ).iloc[-1]