OsirisXRSI_debug

"""
Minimal strategy that only uses the retest law — no extra filters.
Used to verify baseline signal count of the retest logic alone.
Then I'll add filters one by one.
"""
import logging
from datetime import datetime
import pandas as pd
from pandas import DataFrame
from freqtrade.persistence import Trade
from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter
import talib.abstract as ta

try:
    from freqtrade.strategy import stoploss_from_open
except ImportError:
    def stoploss_from_open(open_relative_stop, current_profit, is_short=False):
        if current_profit == 0: return 1
        if is_short: return -1 + ((1 - open_relative_stop) / (1 - current_profit))
        return 1 - ((1 + open_relative_stop) / (1 + current_profit))

logger = logging.getLogger(__name__)


class OsirisXRSI_debug(IStrategy):
    INTERFACE_VERSION = 3
    can_short = True
    timeframe = "5m"
    minimal_roi = {"0": 0.08}
    stoploss = -0.011
    trailing_stop = False
    use_custom_stoploss = True
    startup_candle_count = 500
    process_only_new_candles = True

    tp_pct = DecimalParameter(1.5, 4.0, default=3.0, decimals=1, space="sell", optimize=False)
    max_hold_candles = IntParameter(48, 144, default=96, space="sell", optimize=False)
    cooldown_min = IntParameter(15, 60, default=30, space="buy", optimize=False)

    _last_entry_time: dict = {}

    def informative_pairs(self):
        return []

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["is_green"] = (dataframe["close"] > dataframe["open"]).astype(int)
        dataframe["is_red"] = (dataframe["close"] < dataframe["open"]).astype(int)

        df = dataframe[["date", "close"]].copy()
        df["date"] = pd.to_datetime(df["date"])
        df["grp4h"] = df["date"].dt.floor("4h")
        df4h = df.groupby("grp4h", sort=True).agg(close4h=("close","last")).reset_index()
        df4h["rsi4h"] = ta.RSI(df4h["close4h"], timeperiod=14)
        df = df.merge(df4h[["grp4h","rsi4h"]], on="grp4h", how="left")
        dataframe["rsi_4h"] = df["rsi4h"].values

        setup_long = (
            dataframe["rsi_4h"].notna()
            & (dataframe["volume"] > 0)
            & (dataframe["rsi_4h"] > 65)
            & (dataframe["rsi"] < 35)
            & (dataframe["is_green"] == 1)
        )
        setup_short = (
            dataframe["rsi_4h"].notna()
            & (dataframe["volume"] > 0)
            & (dataframe["rsi_4h"] < 35)
            & (dataframe["rsi"] > 65)
            & (dataframe["is_red"] == 1)
        )

        n = len(dataframe)
        fill_long = [0] * n
        fill_short = [0] * n
        c = dataframe["close"].values
        h = dataframe["high"].values
        l = dataframe["low"].values
        o = dataframe["open"].values

        for i in range(n - 1):
            end = min(i + 1 + 9, n)
            if setup_long.iloc[i]:
                anchor = c[i]; tgt = anchor * 1.03
                lb, hb = anchor * 0.9997, anchor * 1.0003
                for j in range(i + 1, end):
                    if h[j] >= tgt: break
                    if (l[j] <= hb) and (h[j] >= lb):
                        if c[j] > o[j]: fill_long[j] = 1
                        break
            if setup_short.iloc[i]:
                anchor = c[i]; tgt = anchor * 0.97
                lb, hb = anchor * 0.9997, anchor * 1.0003
                for j in range(i + 1, end):
                    if l[j] <= tgt: break
                    if (l[j] <= hb) and (h[j] >= lb):
                        if c[j] < o[j]: fill_short[j] = 1
                        break

        dataframe["entry_fill_long"] = fill_long
        dataframe["entry_fill_short"] = fill_short
        logger.info(f"[DEBUG] fill_long={sum(fill_long)} fill_short={sum(fill_short)} total={n}")
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        long_cond = dataframe["entry_fill_long"] == 1
        short_cond = dataframe["entry_fill_short"] == 1
        logger.info(f"[DEBUG] enter_long_signals={long_cond.sum()} enter_short_signals={short_cond.sum()}")
        dataframe.loc[long_cond, "enter_long"] = 1
        dataframe.loc[short_cond, "enter_short"] = 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:
        last = self._last_entry_time.get(pair)
        if last is not None:
            if (current_time - last).total_seconds() < self.cooldown_min.value * 60:
                return False
        self._last_entry_time[pair] = current_time
        return True

    def custom_stoploss(self, pair, trade: Trade, current_time,
                        current_rate: float, current_profit: float, **kwargs) -> float:
        if current_profit >= 0.008:
            is_short = getattr(trade, "is_short", False)
            return stoploss_from_open(0.001, current_profit, is_short=is_short)
        return self.stoploss

    def custom_exit(self, pair, trade, current_time, current_rate, current_profit, **kwargs):
        if current_profit >= self.tp_pct.value / 100:
            return "xrsi_tp"
        if trade.open_date_utc:
            elapsed_min = (current_time - trade.open_date_utc).total_seconds() / 60
            if elapsed_min >= self.max_hold_candles.value * 5:
                return "xrsi_timeout"
        return None

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