Skip to content

turtle_quant_1.strategies.mean_reversion.rsi_sup_res_divergence

turtle_quant_1.strategies.mean_reversion.rsi_sup_res_divergence

RSI Support Resistance Divergence strategy implementation.

RsiSupResDivergence

RsiSupResDivergence(
    lookback_candles: int = 60,
    local_extrema_window: int = 5,
)

Bases: BaseStrategy

Initialize the RSI strategy.

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

Parameters:

Name Type Description Default
lookback_candles int

The number of periods to use for the RSI.

60
local_extrema_window int

The size of the window for the local maxima and minima.

5
Source code in src/turtle_quant_1/strategies/mean_reversion/rsi_sup_res_divergence.py
def __init__(self, lookback_candles: int = 60, local_extrema_window: int = 5):
    """Initialize the RSI strategy.

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

    Args:
        lookback_candles: The number of periods to use for the RSI.
        local_extrema_window: The size of the window for the local maxima and minima.
    """

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

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

generate_historical_scores

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

Generate the historical scores.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to generate the scores from.

required
symbol str

The symbol to generate the scores for.

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/rsi_sup_res_divergence.py
def generate_historical_scores(self, data: pd.DataFrame, symbol: str) -> pd.Series:
    """Generate the historical scores.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to generate the scores from.
        symbol: The symbol to generate the scores for.

    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"]

    rsi = self._calc_rsi(close)
    maximas, minimas = self._find_extrema(close)
    divergence_signals = self._get_divergence_signals(close, rsi, 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.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name Type Description Default
data DataFrame

The data to generate the score from.

required
symbol str

The symbol to generate the score for.

required

Returns:

Type Description
float

The prediction score.

Source code in src/turtle_quant_1/strategies/mean_reversion/rsi_sup_res_divergence.py
def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate the prediction score.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: The data to generate the score from.
        symbol: The symbol to generate the score for.

    Returns:
        The prediction score.
    """

    return self.generate_historical_scores(
        data.iloc[-(self.lookback_candles + 1) :], symbol
    ).iloc[-1]