OsirisSniper10

"""
OsirisSniper10 — High-Frequency Precision Scalper
===================================================
TARGET: 10+ trades/day | 70%+ Win Rate | 2:1 R:R

APPROACH:
  Multiple entry patterns filtered by 1h trend bias.
  Each pattern must have confluence (2+ signals) to fire.

  TREND:  1h EMA(50) direction + 5m EMA(21) alignment
  
  ENTRY PATTERNS (any one fires):
    A) PULLBACK: price touches EMA(21) in trend + RSI bounce + green/red candle
    B) BB_BOUNCE: BB(20) outer band touch + RSI extreme + reversal candle
    C) MOMENTUM: EMA(8) crosses EMA(21) + MACD histogram flips + volume spike
    
  EXITS:
    TP  = sl_pct × rr_ratio  (2:1 R:R)
    SL  = sl_pct (ATR-calibrated)
    Timeout = max_hold candles
    
  RISK: 1x leverage, single trade at a time
"""
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 OsirisSniper10(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 = 200
    process_only_new_candles = True

    # ── Risk params ──────────────────────────────────────────────────────────
    sl_pct = DecimalParameter(
        0.2, 1.0, default=0.4, decimals=2, space="sell", optimize=True
    )
    rr_ratio = DecimalParameter(
        1.5, 3.0, default=2.0, decimals=1, space="sell", optimize=True
    )
    
    # Breakeven: move SL to entry when profit reaches be_trigger × sl_pct
    be_trigger = DecimalParameter(
        0.8, 1.8, default=1.0, decimals=1, space="sell", optimize=True
    )

    # ── 1h trend filter ──────────────────────────────────────────────────────
    ema_1h_period = IntParameter(20, 100, default=50, space="buy", optimize=True)
    
    # ── 5m indicators ────────────────────────────────────────────────────────
    ema_fast = IntParameter(5, 13, default=8, space="buy", optimize=True)
    ema_slow = IntParameter(15, 30, default=21, space="buy", optimize=True)
    
    # RSI parameters
    rsi_period = IntParameter(5, 14, default=7, space="buy", optimize=True)
    rsi_oversold = IntParameter(20, 45, default=35, space="buy", optimize=True)
    rsi_overbought = IntParameter(55, 80, default=65, space="buy", optimize=True)
    
    # BB parameters
    bb_period = IntParameter(14, 30, default=20, space="buy", optimize=True)
    
    # Volume filter
    vol_mult = DecimalParameter(
        0.8, 2.0, default=1.1, decimals=1, space="buy", optimize=True
    )
    
    # ── Time management ──────────────────────────────────────────────────────
    max_hold_candles = IntParameter(
        6, 48, default=24, space="sell", optimize=True
    )
    cooldown_candles = IntParameter(1, 6, default=2, 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, "1h") for p in pairs]

    # ── Indicators ───────────────────────────────────────────────────────────
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # ── 5m core indicators ───────────────────────────────────────────────
        dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=int(self.ema_fast.value))
        dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=int(self.ema_slow.value))
        dataframe["ema_fast_prev"] = dataframe["ema_fast"].shift(1)
        dataframe["ema_slow_prev"] = dataframe["ema_slow"].shift(1)
        
        # RSI
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=int(self.rsi_period.value))
        dataframe["rsi_prev"] = dataframe["rsi"].shift(1)
        
        # Bollinger Bands
        bb = ta.BBANDS(dataframe, timeperiod=int(self.bb_period.value), nbdevup=2.0, nbdevdn=2.0)
        dataframe["bb_upper"] = bb["upperband"]
        dataframe["bb_lower"] = bb["lowerband"]
        dataframe["bb_mid"] = bb["middleband"]
        
        # MACD
        macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
        dataframe["macd_hist"] = macd["macdhist"]
        dataframe["macd_hist_prev"] = dataframe["macd_hist"].shift(1)
        
        # Volume
        dataframe["volume_sma"] = ta.SMA(dataframe["volume"], timeperiod=20)
        
        # Candle info
        dataframe["is_green"] = (dataframe["close"] > dataframe["open"]).astype(int)
        dataframe["is_red"] = (dataframe["close"] < dataframe["open"]).astype(int)
        
        # ATR for dynamic reference
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        
        # ── 1h trend (from informative pair) ─────────────────────────────────
        inf_1h = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe="1h")
        if not inf_1h.empty:
            inf_1h["ema_1h"] = ta.EMA(inf_1h, timeperiod=int(self.ema_1h_period.value))
            m1h = merge_informative_pair(
                dataframe,
                inf_1h[["date", "ema_1h", "close"]],
                self.timeframe,
                "1h",
                ffill=True,
            )
            dataframe["ema_1h"] = m1h["ema_1h_1h"].values
            dataframe["close_1h"] = m1h["close_1h"].values
        else:
            dataframe["ema_1h"] = dataframe["close"]
            dataframe["close_1h"] = dataframe["close"]

        return dataframe

    # ── Entry signals ────────────────────────────────────────────────────────
    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        oversold = int(self.rsi_oversold.value)
        overbought = int(self.rsi_overbought.value)
        vmult = float(self.vol_mult.value)

        has_data = dataframe["rsi"].notna() & dataframe["ema_1h"].notna() & (dataframe["volume"] > 0)
        vol_ok = dataframe["volume"] > (dataframe["volume_sma"] * vmult)

        # ── 1h Trend Bias ────────────────────────────────────────────────────
        trend_bull = dataframe["close_1h"] > dataframe["ema_1h"]
        trend_bear = dataframe["close_1h"] < dataframe["ema_1h"]
        
        # ── 5m Trend Alignment ───────────────────────────────────────────────
        ema_bull = dataframe["ema_fast"] > dataframe["ema_slow"]
        ema_bear = dataframe["ema_fast"] < dataframe["ema_slow"]

        # ═══════════════════════════════════════════════════════════════════════
        # PATTERN A: PULLBACK TO EMA
        # Price dips to EMA slow, RSI bounces from oversold zone, reversal candle
        # ═══════════════════════════════════════════════════════════════════════
        near_ema_long = (
            (dataframe["low"] <= dataframe["ema_slow"] * 1.002) &
            (dataframe["close"] > dataframe["ema_slow"])
        )
        near_ema_short = (
            (dataframe["high"] >= dataframe["ema_slow"] * 0.998) &
            (dataframe["close"] < dataframe["ema_slow"])
        )
        
        pullback_long = (
            has_data & trend_bull & ema_bull &
            near_ema_long &
            (dataframe["rsi"] > oversold) &
            (dataframe["rsi_prev"] <= oversold + 5) &
            (dataframe["is_green"] == 1)
        )
        
        pullback_short = (
            has_data & trend_bear & ema_bear &
            near_ema_short &
            (dataframe["rsi"] < overbought) &
            (dataframe["rsi_prev"] >= overbought - 5) &
            (dataframe["is_red"] == 1)
        )

        # ═══════════════════════════════════════════════════════════════════════
        # PATTERN B: BB BOUNCE
        # Price touches BB outer band, RSI confirms extreme, reversal candle
        # ═══════════════════════════════════════════════════════════════════════
        bb_long = (
            has_data & trend_bull &
            (dataframe["low"] <= dataframe["bb_lower"]) &
            (dataframe["close"] > dataframe["bb_lower"]) &
            (dataframe["rsi"] < oversold + 10) &
            (dataframe["is_green"] == 1)
        )
        
        bb_short = (
            has_data & trend_bear &
            (dataframe["high"] >= dataframe["bb_upper"]) &
            (dataframe["close"] < dataframe["bb_upper"]) &
            (dataframe["rsi"] > overbought - 10) &
            (dataframe["is_red"] == 1)
        )

        # ═══════════════════════════════════════════════════════════════════════
        # PATTERN C: MOMENTUM CROSS
        # EMA fast crosses slow + MACD histogram flips + volume confirmation
        # ═══════════════════════════════════════════════════════════════════════
        ema_cross_up = (
            (dataframe["ema_fast_prev"] <= dataframe["ema_slow_prev"]) &
            (dataframe["ema_fast"] > dataframe["ema_slow"])
        )
        ema_cross_down = (
            (dataframe["ema_fast_prev"] >= dataframe["ema_slow_prev"]) &
            (dataframe["ema_fast"] < dataframe["ema_slow"])
        )
        
        momentum_long = (
            has_data & trend_bull & vol_ok &
            ema_cross_up &
            (dataframe["macd_hist"] > 0) &
            (dataframe["is_green"] == 1)
        )
        
        momentum_short = (
            has_data & trend_bear & vol_ok &
            ema_cross_down &
            (dataframe["macd_hist"] < 0) &
            (dataframe["is_red"] == 1)
        )

        # ═══════════════════════════════════════════════════════════════════════
        # COMBINE ALL PATTERNS
        # ═══════════════════════════════════════════════════════════════════════
        dataframe.loc[pullback_long | bb_long | momentum_long, "enter_long"] = 1
        dataframe.loc[pullback_short | bb_short | momentum_short, "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)
        cd_seconds = int(self.cooldown_candles.value) * 300  # 5m candles
        if last is not None:
            if (current_time - last).total_seconds() < cd_seconds:
                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_threshold = sl * float(self.be_trigger.value)
        
        if current_profit >= be_threshold:
            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 * float(self.rr_ratio.value)
        
        if current_profit >= tp:
            return "sniper_tp"
            
        if trade.open_date_utc:
            elapsed_candles = (current_time - trade.open_date_utc).total_seconds() / 300
            if elapsed_candles >= self.max_hold_candles.value:
                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