okx-codex-trader/freqtrade/user_data/strategies/BtcRsi2Guarded.py

from datetime import datetime

import pandas as pd
from freqtrade.persistence import Trade
from freqtrade.strategy import IStrategy


class BtcRsi2Guarded(IStrategy):
    INTERFACE_VERSION = 3

    timeframe = "15m"
    can_short = False
    startup_candle_count = 240
    process_only_new_candles = True

    minimal_roi = {"0": 100.0}
    stoploss = -0.024
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    trend_sma = 240
    rsi_length = 2
    rsi_threshold = 2.0
    exit_rsi = 55.0
    max_hold_bars = 48
    leverage_value = 3.0

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        dataframe["trend"] = dataframe["close"].rolling(self.trend_sma).mean()
        dataframe["rsi2"] = self._rsi(dataframe["close"], self.rsi_length)
        return dataframe

    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        dataframe.loc[
            (dataframe["close"] > dataframe["trend"]) & (dataframe["rsi2"] <= self.rsi_threshold),
            ["enter_long", "enter_tag"],
        ] = (1, "rsi2_guarded")
        return dataframe

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        dataframe.loc[dataframe["rsi2"] >= self.exit_rsi, ["exit_long", "exit_tag"]] = (1, "rsi_exit")
        return dataframe

    def custom_exit(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> str | None:
        held_bars = int((current_time - trade.open_date_utc).total_seconds() // (15 * 60))
        if held_bars >= self.max_hold_bars:
            return "max_hold"
        return None

    def leverage(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_leverage: float,
        max_leverage: float,
        entry_tag: str | None,
        side: str,
        **kwargs,
    ) -> float:
        return min(self.leverage_value, max_leverage)

    @staticmethod
    def _rsi(close: pd.Series, length: int) -> pd.Series:
        deltas = close.diff()
        gains = deltas.clip(lower=0.0)
        losses = -deltas.clip(upper=0.0)
        values = [float("nan")] * len(close)
        if len(close) <= length:
            return pd.Series(values, index=close.index)

        average_gain = float(gains.iloc[1 : length + 1].mean())
        average_loss = float(losses.iloc[1 : length + 1].mean())
        for index in range(length, len(close)):
            if index > length:
                average_gain = ((average_gain * (length - 1)) + float(gains.iloc[index])) / length
                average_loss = ((average_loss * (length - 1)) + float(losses.iloc[index])) / length
            if pd.isna(average_gain) or pd.isna(average_loss):
                continue
            if average_loss == 0.0:
                values[index] = 100.0 if average_gain > 0.0 else 50.0
                continue
            relative_strength = average_gain / average_loss
            values[index] = 100.0 - (100.0 / (1.0 + relative_strength))
        return pd.Series(values, index=close.index)