OsirisSniper70

"""
OsirisSniper70 — Precision Day Trade (BTC/USDT)
================================================
TARGET: 70%+ Win Rate | 2:1 R:R | min 10 trades

MECHANISM:
  Multi-timeframe trend following with surgical 5m entries.

  4h:  RSI(14) trend + EMA(20) direction (shifted — no lookahead)
  15m: EMA(50) structure / pullback zone
  5m:  RSI(14) crossback from oversold/overbought + volume spike + candle color

  LONG:  4h bull trend confirmed → 5m RSI dipped below oversold and CROSSED BACK
         above it → green candle + above-average volume → enter
  SHORT: 4h bear trend confirmed → 5m RSI spiked above overbought and CROSSED BACK
         below it → red candle + above-average volume → enter

  EXITS:
    TP  = 2 × SL (guaranteed 2:1 R:R in rate terms)
    SL  = hyperopt-optimized fixed %
    Timeout = max hold time (close at market)
    NO trailing stop. NO breakeven. NO ROI table.
"""

import logging
from datetime import datetime

import numpy as np
import pandas as pd
from pandas import DataFrame

from freqtrade.persistence import Trade
from freqtrade.strategy import (
    IStrategy,
    DecimalParameter,
    IntParameter,
    merge_informative_pair,
)
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 OsirisSniper70(IStrategy):
    INTERFACE_VERSION = 3
    can_short = True
    timeframe = "5m"

    # Safety-net stoploss (actual SL controlled by custom_stoploss)
    stoploss = -0.10
    # Disable ROI exits — all exits via custom logic
    minimal_roi = {"0": 100}
    trailing_stop = False
    use_custom_stoploss = True
    startup_candle_count = 800
    process_only_new_candles = True

    # ── Risk: SL + TP linked at 2:1 ──────────────────────────────────────────
    sl_pct = DecimalParameter(
        0.3, 1.5, default=1.5, decimals=1, space="sell", optimize=True
    )
    # TP = 2 × sl_pct  (enforced in custom_exit)

    # Breakeven: once profit reaches be_mult × sl_pct, move stop to entry +0.1%
    # Values > 2.0 effectively disable BE (TP fires at 2x before BE triggers)
    be_mult = DecimalParameter(
        0.5, 2.5, default=1.1, decimals=1, space="sell", optimize=True
    )

    # ── 4h trend filters ─────────────────────────────────────────────────────
    rsi_4h_bull = IntParameter(55, 75, default=73, space="buy", optimize=True)
    rsi_4h_bear = IntParameter(25, 45, default=33, space="buy", optimize=True)

    # ── 5m entry trigger ─────────────────────────────────────────────────────
    rsi_5m_oversold = IntParameter(20, 40, default=39, space="buy", optimize=True)
    rsi_5m_overbought = IntParameter(60, 80, default=76, space="buy", optimize=True)

    # ── Volume filter ────────────────────────────────────────────────────────
    vol_mult = DecimalParameter(
        0.8, 3.0, default=2.6, decimals=1, space="buy", optimize=True
    )

    # ── 15m structure tolerance ──────────────────────────────────────────────
    ema_tolerance = DecimalParameter(
        0.0, 3.0, default=0.2, decimals=1, space="buy", optimize=True
    )

    # ── Time management ──────────────────────────────────────────────────────
    max_hold_candles = IntParameter(
        12, 96, default=62, space="sell", optimize=True
    )  # 1h–8h in 5m candles
    cooldown_min = IntParameter(10, 120, default=53, space="buy", optimize=True)

    _last_entry_time: dict = {}

    # ── Informative pairs ────────────────────────────────────────────────────
    def informative_pairs(self):
        pairs = self.dp.current_whitelist() if self.dp else []
        return [(p, "15m") for p in pairs]

    # ── Indicators ───────────────────────────────────────────────────────────
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # 5m indicators
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["rsi_prev"] = dataframe["rsi"].shift(1)
        dataframe["volume_sma"] = ta.SMA(dataframe["volume"], timeperiod=20)
        dataframe["is_green"] = (dataframe["close"] > dataframe["open"]).astype(int)
        dataframe["is_red"] = (dataframe["close"] < dataframe["open"]).astype(int)

        # ── 4h indicators (from 5m data, no lookahead via shift) ─────────────
        df = dataframe[["date", "open", "high", "low", "close"]].copy()
        df["date"] = pd.to_datetime(df["date"])
        df["grp4h"] = df["date"].dt.floor("4h")
        df4h = (
            df.groupby("grp4h", sort=True)
            .agg(
                open4h=("open", "first"),
                high4h=("high", "max"),
                low4h=("low", "min"),
                close4h=("close", "last"),
            )
            .reset_index()
        )
        # Compute on 4h bars, then shift(1): each 5m bar sees PREVIOUS completed 4h
        df4h["rsi4h"] = ta.RSI(df4h["close4h"], timeperiod=14)
        df4h["ema20_4h"] = ta.EMA(df4h["close4h"], timeperiod=20)
        df4h["close4h_prev"] = df4h["close4h"]
        # Shift all → no lookahead
        df4h["rsi4h"] = df4h["rsi4h"].shift(1)
        df4h["ema20_4h"] = df4h["ema20_4h"].shift(1)
        df4h["close4h_prev"] = df4h["close4h_prev"].shift(1)

        df = df.merge(
            df4h[["grp4h", "rsi4h", "ema20_4h", "close4h_prev"]],
            on="grp4h",
            how="left",
        )
        dataframe["rsi_4h"] = df["rsi4h"].values
        dataframe["ema20_4h"] = df["ema20_4h"].values
        dataframe["close_4h"] = df["close4h_prev"].values

        # ── 15m EMA50 structure ──────────────────────────────────────────────
        inf_15m = self.dp.get_pair_dataframe(
            pair=metadata["pair"], timeframe="15m"
        )
        if not inf_15m.empty:
            inf_15m["ema50_15"] = ta.EMA(inf_15m, timeperiod=50)
            m15 = merge_informative_pair(
                dataframe,
                inf_15m[["date", "ema50_15"]],
                self.timeframe,
                "15m",
                ffill=True,
            )
            dataframe["ema50_15m"] = m15["ema50_15_15m"].values
        else:
            dataframe["ema50_15m"] = dataframe["close"]

        return dataframe

    # ── Entry signals ────────────────────────────────────────────────────────
    def populate_entry_trend(
        self, dataframe: DataFrame, metadata: dict
    ) -> DataFrame:
        bull = int(self.rsi_4h_bull.value)
        bear = int(self.rsi_4h_bear.value)
        oversold = int(self.rsi_5m_oversold.value)
        overbought = int(self.rsi_5m_overbought.value)
        vmult = float(self.vol_mult.value)
        tol = float(self.ema_tolerance.value) / 100.0

        ema50 = dataframe["ema50_15m"]
        ema20_4h = dataframe["ema20_4h"]
        close_4h = dataframe["close_4h"]

        # Common filters
        has_data = dataframe["rsi_4h"].notna() & (dataframe["volume"] > 0)
        vol_ok = dataframe["volume"] > (dataframe["volume_sma"] * vmult)

        # ── RSI crossback: was oversold/overbought, now recovering ───────────
        rsi_crossback_up = (
            (dataframe["rsi_prev"] < oversold) & (dataframe["rsi"] >= oversold)
        )
        rsi_crossback_down = (
            (dataframe["rsi_prev"] > overbought)
            & (dataframe["rsi"] <= overbought)
        )

        # ── LONG ─────────────────────────────────────────────────────────────
        long_cond = (
            has_data
            & vol_ok
            & (dataframe["rsi_4h"] > bull)                   # 4h RSI bullish
            & (close_4h > ema20_4h)                           # prev 4h close above EMA20
            & (dataframe["close"] >= ema50 * (1.0 - tol))    # 15m structure intact
            & rsi_crossback_up                                 # 5m RSI bounced from oversold
            & (dataframe["is_green"] == 1)                    # green candle
        )

        # ── SHORT ────────────────────────────────────────────────────────────
        short_cond = (
            has_data
            & vol_ok
            & (dataframe["rsi_4h"] < bear)                    # 4h RSI bearish
            & (close_4h < ema20_4h)                            # prev 4h close below EMA20
            & (dataframe["close"] <= ema50 * (1.0 + tol))     # 15m structure intact
            & rsi_crossback_down                               # 5m RSI rejected from overbought
            & (dataframe["is_red"] == 1)                       # red candle
        )

        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

    # ── Trade management ─────────────────────────────────────────────────────
    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:
        sl = float(self.sl_pct.value) / 100
        be_trigger = sl * float(self.be_mult.value)
        if current_profit >= be_trigger:
            is_short = getattr(trade, "is_short", False)
            return stoploss_from_open(0.001, current_profit, is_short=is_short)
        return -sl

    def custom_exit(
        self, pair, trade, current_time, current_rate, current_profit, **kwargs,
    ):
        sl = float(self.sl_pct.value) / 100
        tp = sl * 2  # 2:1 R:R guaranteed
        if current_profit >= tp:
            return "sniper_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 "sniper_timeout"
        return None

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