turtle_quant_1.strategies.momentum.relative_strength_index

turtle_quant_1.strategies.momentum.relative_strength_index

Relative Strength Index (RSI) strategy implementation.

RelativeStrengthIndex

RelativeStrengthIndex(lookback_candles: int = 14)

Bases: BaseStrategy

A strategy that uses the RSI to generate buy and sell signals.

Leading / Lagging: Lagging in construction. Leading in interpretation. Lag period (in candles): (lookback_candles - 1) / 2 Effect: Delay in recognizing that a new trend has begun.

Initialize the RSI strategy.

Parameters:

Name	Type	Description	Default
lookback_candles	int	The number of periods to use for the RSI.	14

Source code in src/turtle_quant_1/strategies/momentum/relative_strength_index.py

def __init__(self, lookback_candles: int = 14):
    """Initialize the RSI strategy.

    Args:
        lookback_candles: The number of periods to use for the RSI.
    """
    super().__init__()
    self.lookback_candles = lookback_candles

    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 a historical score array for a symbol based on market data.

NOTE: Assume that the data is sorted by datetime.

Parameters:

Name	Type	Description	Default
data	DataFrame	DataFrame with OHLCV data.	required
symbol	str	The symbol being analyzed.	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/momentum/relative_strength_index.py

def generate_historical_scores(self, data: pd.DataFrame, symbol: str) -> pd.Series:
    """Generate a historical score array for a symbol based on market data.

    NOTE: Assume that the data is sorted by datetime.

    Args:
        data: DataFrame with OHLCV data.
        symbol: The symbol being analyzed.

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

    # TODO: This is a magic number.
    data_resampled = DataUnitConverter.convert_to_1h_data(symbol, data)

    delta = data_resampled["Close"].diff()
    gain = delta.clip(lower=0)
    loss = -delta.clip(upper=0)

    mean_gain = gain.rolling(self.lookback_candles).mean()
    mean_loss = loss.rolling(self.lookback_candles).mean()

    rs = mean_gain / (mean_loss + 1e-9)  # Prevent division by zero
    rsi = 100 - (100 / (1 + rs))  # The standard RSI formula
    if np.isnan(rsi.iloc[-1]):
        raise ValueError("Last score should not be NaN")

    rsi = rsi.reindex(data.index)
    rsi = rsi.bfill().ffill()

    return pd.Series(
        data=((50 - rsi) / 50).clip(-1, 1).fillna(0).values,  # Rescale to -1 to +1
        index=data.index,
    )

generate_prediction_score

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

Generate a score for the strategy.

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

Source code in src/turtle_quant_1/strategies/momentum/relative_strength_index.py

def generate_prediction_score(self, data: pd.DataFrame, symbol: str) -> float:
    """Generate a score for the strategy.

    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.
    """
    # RSI < 30 -> Score near +1 -> BUY
    # RSI > 70 -> Score near -1 -> SELL
    return self.generate_historical_scores(
        # TODO: Using * 2.0 here to give buffer zone for any miscalculations.
        # This depends on the resampling of the data
        data.iloc[
            -(
                round(convert_units(self.lookback_candles, "1H", CANDLE_UNIT) * 2.0)
            ) :
        ],
        symbol,
    ).iloc[-1]