OsirisSwing

"""
OsirisSwing — Trend-Following Mean-Reversion
=============================================
LONG: RSI<30 dip in 1h uptrend (buy the dip with the trend)
SHORT: RSI>70 rip in 1h downtrend (sell the rip against the trend)

SL=1.5% via custom_stoploss (intra-candle)
TP=3.0% via custom_stoploss (intra-candle, returns tight trail when profit >= TP)
Max hold: 72 candles (6h) via custom_exit

Simulation results (Jan 2024 - Apr 2025, with 0.08% fees):
  LONG:  242tr 54% WR PF=1.40 +37.7%
  SHORT: 231tr 47% WR PF=1.12 +13.3%
  TOTAL: 473tr ~1/day  PF~1.27  +51.0%
"""

import logging
from datetime import timedelta

import numpy as np
from pandas import DataFrame

from freqtrade.persistence import Trade
from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter, merge_informative_pair
import talib.abstract as ta

logger = logging.getLogger(__name__)


class OsirisSwing(IStrategy):
    INTERFACE_VERSION = 3
    can_short = True
    timeframe = "5m"
    informative_pairs_timeframe = "1h"

    # SL: native stoploss from ENTRY (checked intra-candle against LOW/HIGH)
    stoploss = -0.005
    use_custom_stoploss = False
    trailing_stop = False

    # TP via minimal_roi (checked intra-candle against HIGH for longs, LOW for shorts)
    minimal_roi = {"0": 0.01}

    startup_candle_count = 700
    process_only_new_candles = True

    # Hyperopt parameters
    rsi_os = IntParameter(25, 35, default=30, space="buy", optimize=True)
    rsi_ob = IntParameter(65, 78, default=70, space="buy", optimize=True)
    sl_pct = DecimalParameter(0.3, 1.5, default=0.5, decimals=1, space="sell", optimize=True)
    tp_pct = DecimalParameter(0.5, 2.0, default=1.0, decimals=1, space="sell", optimize=True)
    max_hold = IntParameter(12, 72, default=24, space="sell", optimize=True)

    def informative_pairs(self):
        return [("BTC/USDT:USDT", "1h")]

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 5m indicators
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)

        # Get 1h informative data for trend detection
        inf_1h = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe="1h")
        if len(inf_1h) > 0:
            inf_1h["sma50"] = ta.SMA(inf_1h, timeperiod=50)
            inf_1h["trend_up"] = (inf_1h["close"] > inf_1h["sma50"]).astype(int)
            dataframe = merge_informative_pair(dataframe, inf_1h, self.timeframe, "1h", ffill=True)
        else:
            dataframe["close_1h"] = np.nan
            dataframe["sma50_1h"] = np.nan
            dataframe["trend_up_1h"] = 0

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        os_thresh = int(self.rsi_os.value)
        has_data = dataframe["rsi"].notna() & dataframe["trend_up_1h"].notna() & (dataframe["volume"] > 0)

        # LONG ONLY: oversold dip in uptrend
        long_signal = (
            has_data
            & (dataframe["rsi"] < os_thresh)
            & (dataframe["trend_up_1h"] == 1)
        )

        dataframe.loc[long_signal, "enter_long"] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        return dataframe

    def confirm_trade_entry(self, pair, order_type, amount, rate, time_in_force,
                            current_time, entry_tag, side, **kwargs) -> bool:
        """Cooldown: skip entry if last trade on this pair closed < 15 min ago."""
        if not hasattr(self, "_last_exit_time"):
            self._last_exit_time = {}
        last = self._last_exit_time.get(pair)
        if last and (current_time - last).total_seconds() < 900:  # 15 min
            return False
        return True

    def custom_exit(self, pair: str, trade: Trade, current_time,
                    current_rate: float, current_profit: float, **kwargs):
        """Timeout exit after max_hold candles. Also record exit time for cooldown."""
        if trade.open_date_utc:
            minutes = (current_time - trade.open_date_utc).total_seconds() / 60
            candles = minutes / 5
            if candles >= self.max_hold.value:
                if not hasattr(self, "_last_exit_time"):
                    self._last_exit_time = {}
                self._last_exit_time[pair] = current_time
                return "timeout"
        return None

    def confirm_trade_exit(self, pair, trade, order_type, amount, rate,
                           time_in_force, exit_reason, current_time, **kwargs) -> bool:
        """Record exit time for cooldown."""
        if not hasattr(self, "_last_exit_time"):
            self._last_exit_time = {}
        self._last_exit_time[pair] = current_time
        return True

    def leverage(self, pair, current_time, current_rate, proposed_leverage,
                 max_leverage, entry_tag, side, **kwargs) -> float:
        return 1.0