OsirisScalperV2

"""
OsirisScalperV2 — Multi-Timeframe Pullback Scalper (5m)
========================================================
1h: EMA cross + MACD histogram = directional bias
5m: BB pullback + reversal candle pattern = entry
Exit: ATR-based TP/SL, trailing breakeven, timeout

IS (Jan-Mar 2025): 352 trades, 53% WR, +5%, PF 1.25
OOS (Apr-Jun 2025): Longs +1.32%, Shorts -4.18%
"""

import logging
from pandas import DataFrame
import numpy as np

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 OsirisScalperV2(IStrategy):
    INTERFACE_VERSION = 3
    can_short = True
    timeframe = "5m"

    stoploss = -0.05
    minimal_roi = {"0": 100}
    trailing_stop = False
    use_custom_stoploss = True
    startup_candle_count = 100
    process_only_new_candles = True

    # ── Risk (sell) — ATR multipliers ────────────────────────────────
    tp_atr = DecimalParameter(1.5, 5.0, default=3.0, decimals=1, space="sell", optimize=True)
    sl_atr = DecimalParameter(0.8, 3.0, default=1.5, decimals=1, space="sell", optimize=True)
    be_trigger = DecimalParameter(0.3, 2.0, default=1.0, decimals=1, space="sell", optimize=True)
    max_hold = IntParameter(6, 60, default=24, space="sell", optimize=True)

    # ── Entry (buy) ──────────────────────────────────────────────────
    ema_fast_1h = IntParameter(5, 15, default=8, space="buy", optimize=True)
    ema_slow_1h = IntParameter(15, 40, default=21, space="buy", optimize=True)
    bb_period = IntParameter(14, 30, default=20, space="buy", optimize=True)
    bb_std = DecimalParameter(1.5, 3.0, default=2.0, decimals=1, space="buy", optimize=True)
    pb_depth = DecimalParameter(0.2, 0.8, default=0.5, decimals=1, space="buy", optimize=True)
    pb_streak = IntParameter(1, 4, default=2, space="buy", optimize=True)
    vol_threshold = DecimalParameter(0.6, 2.0, default=1.0, decimals=1, space="buy", optimize=True)
    rsi_low = IntParameter(20, 45, default=35, space="buy", optimize=True)
    rsi_high = IntParameter(55, 80, default=65, space="buy", optimize=True)

    _cooldown: dict = {}
    _ema_fast_grid = [5, 8, 10, 12, 15]
    _ema_slow_grid = [15, 21, 25, 30, 40]
    _bb_period_grid = [14, 20, 26]
    _bb_std_grid = [1.5, 2.0, 2.5, 3.0]

    def informative_pairs(self):
        return [(p, "1h") for p in self.dp.current_whitelist()]

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)

        for period in self._bb_period_grid:
            for std in self._bb_std_grid:
                s = f"{period}_{int(std * 10)}"
                bb = ta.BBANDS(dataframe, timeperiod=period, nbdevup=std, nbdevdn=std)
                dataframe[f"bb_u_{s}"] = bb["upperband"]
                dataframe[f"bb_m_{s}"] = bb["middleband"]
                dataframe[f"bb_l_{s}"] = bb["lowerband"]

        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["vol_sma"] = dataframe["volume"].rolling(20).mean()
        dataframe["is_up"] = (dataframe["close"] > dataframe["open"]).astype(int)

        is_up = dataframe["is_up"].values
        streak_up = np.zeros(len(is_up), dtype=int)
        streak_dn = np.zeros(len(is_up), dtype=int)
        for i in range(1, len(is_up)):
            if is_up[i] == 1:
                streak_up[i] = streak_up[i - 1] + 1
            else:
                streak_dn[i] = streak_dn[i - 1] + 1
        dataframe["streak_up"] = streak_up
        dataframe["streak_dn"] = streak_dn

        # ── 1h informative merge ─────────────────────────────────────
        inf_1h = self.dp.get_pair_dataframe(metadata["pair"], "1h")
        if not inf_1h.empty:
            all_emas = sorted(set(self._ema_fast_grid + self._ema_slow_grid))
            for p in all_emas:
                inf_1h[f"ema_{p}"] = ta.EMA(inf_1h, timeperiod=p)
            macd = ta.MACD(inf_1h, fastperiod=12, slowperiod=26, signalperiod=9)
            inf_1h["macdhist"] = macd["macdhist"]

            cols = ["date"] + [f"ema_{p}" for p in all_emas] + ["macdhist"]
            inf_sel = inf_1h[cols].copy()
            inf_sel = inf_sel.rename(columns={c: f"1h_{c}" for c in cols if c != "date"})
            dataframe = dataframe.merge(inf_sel, on="date", how="left")
            for c in dataframe.columns:
                if c.startswith("1h_"):
                    dataframe[c] = dataframe[c].ffill()

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        pb_d = float(self.pb_depth.value)
        pb_s = int(self.pb_streak.value)

        ef = min(self._ema_fast_grid, key=lambda x: abs(x - int(self.ema_fast_1h.value)))
        es = min(self._ema_slow_grid, key=lambda x: abs(x - int(self.ema_slow_1h.value)))
        bb_p = min(self._bb_period_grid, key=lambda x: abs(x - int(self.bb_period.value)))
        bb_s_val = min([int(s * 10) for s in self._bb_std_grid],
                       key=lambda x: abs(x - int(self.bb_std.value * 10)))
        bb_key = f"{bb_p}_{bb_s_val}"

        if f"1h_ema_{ef}" not in dataframe.columns:
            return dataframe

        ok = dataframe["rsi"].notna() & (dataframe["volume"] > 0)
        vol_ok = dataframe["volume"] > dataframe["vol_sma"] * float(self.vol_threshold.value)
        rsi_ok = (dataframe["rsi"] > int(self.rsi_low.value)) & (dataframe["rsi"] < int(self.rsi_high.value))

        trend_bull = (dataframe[f"1h_ema_{ef}"] > dataframe[f"1h_ema_{es}"]) & (dataframe["1h_macdhist"] > 0)
        trend_bear = (dataframe[f"1h_ema_{ef}"] < dataframe[f"1h_ema_{es}"]) & (dataframe["1h_macdhist"] < 0)

        bb_w = dataframe[f"bb_u_{bb_key}"] - dataframe[f"bb_l_{bb_key}"]
        pullback_long = dataframe["close"] <= dataframe[f"bb_l_{bb_key}"] + bb_w * pb_d
        pullback_short = dataframe["close"] >= dataframe[f"bb_u_{bb_key}"] - bb_w * pb_d
        reversal_up = (dataframe["streak_dn"].shift(1) >= pb_s) & (dataframe["is_up"] == 1)
        reversal_dn = (dataframe["streak_up"].shift(1) >= pb_s) & (dataframe["is_up"] == 0)

        dataframe.loc[ok & trend_bull & pullback_long & reversal_up & vol_ok & rsi_ok, "enter_long"] = 1
        dataframe.loc[ok & trend_bear & pullback_short & reversal_dn & vol_ok & rsi_ok, "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._cooldown.get(pair)
        if last is not None and (current_time - last).total_seconds() < 300:
            return False
        self._cooldown[pair] = current_time
        return True

    def custom_stoploss(self, pair, trade: Trade, current_time,
                        current_rate: float, current_profit: float, **kwargs) -> float:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return -0.01

        atr = dataframe["atr"].iloc[-1]
        close = dataframe["close"].iloc[-1]
        atr_pct = atr / close if close > 0 else 0.003

        sl = atr_pct * float(self.sl_atr.value)
        be = atr_pct * float(self.be_trigger.value)

        if current_profit >= be:
            is_short = getattr(trade, "is_short", False)
            return stoploss_from_open(0.0002, current_profit, is_short=is_short)
        return -sl

    def custom_exit(self, pair, trade, current_time, current_rate, current_profit, **kwargs):
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return None

        atr = dataframe["atr"].iloc[-1]
        close = dataframe["close"].iloc[-1]
        atr_pct = atr / close if close > 0 else 0.003
        tp = atr_pct * float(self.tp_atr.value)

        if current_profit >= tp:
            return "tp_hit"
        if trade.open_date_utc:
            candles = (current_time - trade.open_date_utc).total_seconds() / 300
            if candles >= self.max_hold.value:
                return "timeout"
        return None

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