OsirisGoldV8

"""
OsirisGoldV8 — Time-Profit Pure Edge (BTC 5m)
================================================
EVIDENCE FROM V7 BACKTEST (3286 trades, BTC 2025):

EXIT ANALYSIS — CONCLUSIVE:
  v7_time_profit:     776W /   0L = 100% WR, +3.1%  ← THE ONLY EDGE
  v7_late_profit:     311W /   0L = 100% WR, +0.7%  ← Also works
  v7_timeout:          70W / 800L =   8% WR, -3.8%  ← Main loss source
  trailing_stop_loss: 305W / 376L =  45% WR, -1.8%  ← Trailing kills
  v7_momentum_exit:    0W / 638L =   0% WR, -0.7%  ← Pure garbage
  
SCORING ANALYSIS — CONCLUSIVE:
  All scores 11-14: WR = 43-49% → NO DIRECTIONAL EDGE
  Higher score ≠ better WR → scoring is noise for direction
  
IMPLICATION:
  The edge is NOT in entry direction. It's in TIME.
  If you enter random and wait → after 40min if green, take it.
  The Research 100% WR comes from extreme selectivity (almost never
  enters), not from predicting direction.
  
V8 DESIGN — PURE TIME-PROFIT:
  - Entry: Research scoring at threshold 11 (catch everything)
  - NO trailing, NO breakeven, NO momentum exit
  - SL: Fixed wide (-3%) — just protection vs flash crash
  - TP1: After 8 candles (40min) if profit > 0.3% → exit
  - TP2: After 16 candles if profit > 0.15% → exit
  - Timeout: After 24 candles → flat exit
  - Loss cap: If loss > -1% after 12 candles → cut early
"""
import logging
import numpy as np
from pandas import DataFrame
from typing import Optional

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

logger = logging.getLogger(__name__)


class OsirisGoldV8(IStrategy):
    INTERFACE_VERSION = 3
    can_short = True
    timeframe = "5m"

    stoploss = -0.03          # Fixed 3% hard stop. Wide enough to survive noise.
    minimal_roi = {"0": 100}  # Disabled
    trailing_stop = False     # CRITICAL: No trailing. V7 proved it kills.
    use_custom_stoploss = False  # No custom stoploss either. Pure fixed SL.
    startup_candle_count = 200
    process_only_new_candles = True

    # ── ENTRY ────────────────────────────────────────────────────────────
    score_min_long = IntParameter(9, 14, default=11, space="buy", optimize=True)
    score_min_short = IntParameter(9, 14, default=11, space="buy", optimize=True)

    buy_pressure = DecimalParameter(0.45, 0.60, default=0.50, decimals=2, space="buy", optimize=True)
    buy_rvol = DecimalParameter(0.7, 1.5, default=0.9, decimals=1, space="buy", optimize=True)
    buy_cvd_bars = IntParameter(2, 5, default=3, space="buy", optimize=True)
    buy_mfi = IntParameter(15, 35, default=22, space="buy", optimize=True)
    buy_rsi_min = IntParameter(25, 40, default=29, space="buy", optimize=True)
    buy_rsi_max = IntParameter(60, 75, default=69, space="buy", optimize=True)
    buy_stochrsi = IntParameter(25, 55, default=44, space="buy", optimize=True)
    buy_cci = IntParameter(-150, -40, default=-75, space="buy", optimize=True)
    buy_adx = IntParameter(12, 30, default=15, space="buy", optimize=True)
    buy_ribbon = IntParameter(1, 4, default=1, space="buy", optimize=True)
    buy_efficiency = DecimalParameter(0.05, 0.25, default=0.11, decimals=2, space="buy", optimize=True)
    buy_vol_ratio = DecimalParameter(1.0, 2.0, default=1.6, decimals=1, space="buy", optimize=True)
    buy_rsi_15m = IntParameter(30, 50, default=36, space="buy", optimize=True)

    # ── EXIT (time-profit system) ────────────────────────────────────────
    tp1_candles = IntParameter(4, 12, default=8, space="sell", optimize=True)
    tp1_min_pct = DecimalParameter(0.1, 0.5, default=0.3, decimals=1, space="sell", optimize=True)
    tp2_candles = IntParameter(12, 24, default=16, space="sell", optimize=True)
    tp2_min_pct = DecimalParameter(0.05, 0.3, default=0.15, decimals=2, space="sell", optimize=True)
    timeout_candles = IntParameter(18, 48, default=24, space="sell", optimize=True)
    early_loss_candles = IntParameter(8, 20, default=12, space="sell", optimize=True)
    early_loss_pct = DecimalParameter(0.5, 1.5, default=1.0, decimals=1, space="sell", optimize=True)

    _last_entry: dict = {}
    _consec_losses: dict = {}

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

    # =====================================================================
    # INDICATORS (same as V7/Research)
    # =====================================================================
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        pair = metadata["pair"]

        # Standard
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
        dataframe["macd_hist"] = macd["macdhist"]

        for p in [8, 13, 21, 50, 200]:
            dataframe[f"ema{p}"] = ta.EMA(dataframe, timeperiod=p)

        dataframe["ema_ribbon"] = (
            (dataframe["ema8"] > dataframe["ema13"]).astype(int)
            + (dataframe["ema13"] > dataframe["ema21"]).astype(int)
            + (dataframe["ema21"] > dataframe["ema50"]).astype(int)
            + (dataframe["ema50"] > dataframe["ema200"]).astype(int)
        )
        dataframe["ema_ribbon_bear"] = (
            (dataframe["ema8"] < dataframe["ema13"]).astype(int)
            + (dataframe["ema13"] < dataframe["ema21"]).astype(int)
            + (dataframe["ema21"] < dataframe["ema50"]).astype(int)
            + (dataframe["ema50"] < dataframe["ema200"]).astype(int)
        )

        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)
        stoch = ta.STOCHRSI(dataframe, timeperiod=14, fastk_period=3, fastd_period=3)
        dataframe["stochrsi_k"] = stoch["fastk"]
        dataframe["cci"] = ta.CCI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)

        # VWAP
        tp = (dataframe["high"] + dataframe["low"] + dataframe["close"]) / 3
        vwap_vol = dataframe["volume"].rolling(50).sum().replace(0, 1)
        dataframe["vwap"] = (tp * dataframe["volume"]).rolling(50).sum() / vwap_vol

        # Volume Flow
        hl = (dataframe["high"] - dataframe["low"]).replace(0, np.nan)
        dataframe["buy_vol"] = (dataframe["volume"] * (dataframe["close"] - dataframe["low"]) / hl).fillna(dataframe["volume"] * 0.5)
        dataframe["sell_vol"] = (dataframe["volume"] * (dataframe["high"] - dataframe["close"]) / hl).fillna(dataframe["volume"] * 0.5)
        total = (dataframe["buy_vol"] + dataframe["sell_vol"]).replace(0, 1)
        dataframe["pressure_ratio"] = (dataframe["buy_vol"] / total).fillna(0.5)
        dataframe["volume_delta"] = dataframe["buy_vol"] - dataframe["sell_vol"]
        dataframe["cvd"] = dataframe["volume_delta"].cumsum()
        dataframe["cvd_rising"] = (dataframe["cvd"] > dataframe["cvd"].shift(1)).astype(int)
        dataframe["vol_sma20"] = dataframe["volume"].rolling(20).mean()
        dataframe["rvol"] = (dataframe["volume"] / dataframe["vol_sma20"].replace(0, 1)).fillna(1)
        dataframe["mfi"] = ta.MFI(dataframe, timeperiod=14)

        # Microstructure
        body = (dataframe["close"] - dataframe["open"]).abs()
        hl_safe = (dataframe["high"] - dataframe["low"]).replace(0, np.nan).fillna(0.0001)
        dataframe["body_ratio"] = body / hl_safe
        dataframe["is_absorption"] = ((dataframe["body_ratio"] < 0.30) & (dataframe["rvol"] > 1.5)).astype(int)
        dataframe["lower_wick_ratio"] = (dataframe[["open", "close"]].min(axis=1) - dataframe["low"]) / hl_safe
        dataframe["upper_wick_ratio"] = (dataframe["high"] - dataframe[["open", "close"]].max(axis=1)) / hl_safe
        dataframe["is_pin_bar_bull"] = ((dataframe["lower_wick_ratio"] > 0.60) & (dataframe["body_ratio"] < 0.25) & (dataframe["upper_wick_ratio"] < 0.15)).astype(int)
        dataframe["is_pin_bar_bear"] = ((dataframe["upper_wick_ratio"] > 0.60) & (dataframe["body_ratio"] < 0.25) & (dataframe["lower_wick_ratio"] < 0.15)).astype(int)
        recent_high = dataframe["high"].rolling(20).max().shift(1)
        recent_low = dataframe["low"].rolling(20).min().shift(1)
        dataframe["liq_sweep_low"] = ((dataframe["low"] < recent_low) & (dataframe["close"] > recent_low)).astype(int)
        dataframe["liq_sweep_high"] = ((dataframe["high"] > recent_high) & (dataframe["close"] < recent_high)).astype(int)

        # Innovation
        period = 20
        direction = (dataframe["close"] - dataframe["close"].shift(period)).abs()
        volatility = dataframe["close"].diff().abs().rolling(period).sum()
        dataframe["efficiency_ratio"] = (direction / volatility.replace(0, 1)).fillna(0)
        atr50 = ta.ATR(dataframe, timeperiod=50)
        dataframe["volatility_ratio"] = (dataframe["atr"] / atr50.replace(0, 1)).fillna(1)

        # Squeeze
        kelt_ema = ta.EMA(dataframe, timeperiod=20)
        kelt_atr = ta.ATR(dataframe, timeperiod=10)
        bb = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
        squeeze_on = ((bb["lowerband"] > kelt_ema - kelt_atr * 1.5) & (bb["upperband"] < kelt_ema + kelt_atr * 1.5)).astype(int)
        dataframe["squeeze_release_bull"] = ((squeeze_on.shift(1) == 1) & (squeeze_on == 0) & (dataframe["close"] > dataframe["close"].shift(1))).astype(int)
        dataframe["squeeze_release_bear"] = ((squeeze_on.shift(1) == 1) & (squeeze_on == 0) & (dataframe["close"] < dataframe["close"].shift(1))).astype(int)

        # 15m
        inf_15m = self.dp.get_pair_dataframe(pair=pair, timeframe="15m") if self.dp else DataFrame()
        if not inf_15m.empty:
            inf_15m["rsi_15m"] = ta.RSI(inf_15m, timeperiod=14)
            m15 = merge_informative_pair(
                dataframe, inf_15m[["date", "rsi_15m"]],
                self.timeframe, "15m", ffill=True,
            )
            dataframe["rsi_15m"] = m15["rsi_15m_15m"].values
        else:
            dataframe["rsi_15m"] = 50

        return dataframe

    # =====================================================================
    # SCORING — LONG (17pts)
    # =====================================================================
    def _score_long(self, df):
        s = np.zeros(len(df))
        s += (df["pressure_ratio"] > self.buy_pressure.value).astype(int)
        s += (df["rvol"] > self.buy_rvol.value).astype(int)
        cvd_ct = df["cvd_rising"].rolling(int(self.buy_cvd_bars.value)).sum()
        s += (cvd_ct >= self.buy_cvd_bars.value).astype(int)
        s += (df["mfi"] > self.buy_mfi.value).astype(int)
        s += ((df["rsi"] > self.buy_rsi_min.value) & (df["rsi"] < self.buy_rsi_max.value)).astype(int)
        s += ((df["macd_hist"] > 0) | (df["macd_hist"] > df["macd_hist"].shift(1))).astype(int)
        s += (df["stochrsi_k"] < self.buy_stochrsi.value).astype(int)
        s += ((df["cci"] > self.buy_cci.value) & (df["cci"].shift(1) <= self.buy_cci.value)).astype(int)
        s += (df["adx"] > self.buy_adx.value).astype(int)
        s += (df["ema_ribbon"] >= self.buy_ribbon.value).astype(int)
        s += (df["close"] > df["vwap"]).astype(int)
        s += (df["is_absorption"].rolling(3).sum() > 0).astype(int)
        s += ((df["is_pin_bar_bull"] == 1) | (df["liq_sweep_low"] == 1)).astype(int)
        s += (df["squeeze_release_bull"] == 1).astype(int)
        s += (df["efficiency_ratio"] > self.buy_efficiency.value).astype(int)
        s += (df["volatility_ratio"] < self.buy_vol_ratio.value).astype(int)
        s += (df["rsi_15m"] > self.buy_rsi_15m.value).astype(int)
        return s

    # =====================================================================
    # SCORING — SHORT (17pts, inverted)
    # =====================================================================
    def _score_short(self, df):
        s = np.zeros(len(df))
        s += (df["pressure_ratio"] < (1 - self.buy_pressure.value)).astype(int)
        s += (df["rvol"] > self.buy_rvol.value).astype(int)
        cvd_fall = (1 - df["cvd_rising"]).rolling(int(self.buy_cvd_bars.value)).sum()
        s += (cvd_fall >= self.buy_cvd_bars.value).astype(int)
        s += (df["mfi"] < (100 - self.buy_mfi.value)).astype(int)
        rsi_lo = 100 - self.buy_rsi_max.value
        rsi_hi = 100 - self.buy_rsi_min.value
        s += ((df["rsi"] > rsi_lo) & (df["rsi"] < rsi_hi)).astype(int)
        s += ((df["macd_hist"] < 0) | (df["macd_hist"] < df["macd_hist"].shift(1))).astype(int)
        s += (df["stochrsi_k"] > (100 - self.buy_stochrsi.value)).astype(int)
        cci_inv = -self.buy_cci.value
        s += ((df["cci"] < cci_inv) & (df["cci"].shift(1) >= cci_inv)).astype(int)
        s += (df["adx"] > self.buy_adx.value).astype(int)
        s += (df["ema_ribbon_bear"] >= self.buy_ribbon.value).astype(int)
        s += (df["close"] < df["vwap"]).astype(int)
        s += (df["is_absorption"].rolling(3).sum() > 0).astype(int)
        s += ((df["is_pin_bar_bear"] == 1) | (df["liq_sweep_high"] == 1)).astype(int)
        s += (df["squeeze_release_bear"] == 1).astype(int)
        s += (df["efficiency_ratio"] > self.buy_efficiency.value).astype(int)
        s += (df["volatility_ratio"] < self.buy_vol_ratio.value).astype(int)
        s += (df["rsi_15m"] < (100 - self.buy_rsi_15m.value)).astype(int)
        return s

    # =====================================================================
    # ENTRY
    # =====================================================================
    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        has_data = dataframe["rsi"].notna() & (dataframe["volume"] > 0)
        sl = self._score_long(dataframe)
        ss = self._score_short(dataframe)

        long_sig = has_data & (sl >= int(self.score_min_long.value))
        short_sig = has_data & (ss >= int(self.score_min_short.value))

        dataframe.loc[long_sig, "enter_long"] = 1
        dataframe.loc[long_sig, "enter_tag"] = "L" + sl[long_sig].astype(int).astype(str)
        dataframe.loc[short_sig, "enter_short"] = 1
        dataframe.loc[short_sig, "enter_tag"] = "S" + ss[short_sig].astype(int).astype(str)

        return dataframe

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

    # =====================================================================
    # ENTRY GUARD
    # =====================================================================
    def confirm_trade_entry(
        self, pair, order_type, amount, rate, time_in_force,
        current_time, entry_tag, side, **kwargs,
    ) -> bool:
        # Loss-streak cooldown (5 losses → skip 1)
        if self._consec_losses.get(pair, 0) >= 5:
            self._consec_losses[pair] = 0
            return False
        # 2 candle cooldown
        last = self._last_entry.get(pair)
        if last is not None and (current_time - last).total_seconds() < 600:
            return False
        self._last_entry[pair] = current_time
        return True

    # =====================================================================
    # EXIT — PURE TIME-PROFIT SYSTEM
    # =====================================================================
    def custom_exit(
        self, pair, trade, current_time, current_rate, current_profit, **kwargs,
    ):
        if not trade.open_date_utc:
            return None

        elapsed = (current_time - trade.open_date_utc).total_seconds() / 300

        # TP1: After N candles, take decent profit
        tp1_c = int(self.tp1_candles.value)
        tp1_p = float(self.tp1_min_pct.value) / 100
        if elapsed >= tp1_c and current_profit >= tp1_p:
            self._update_streak(pair, True)
            return "tp1_time_profit"

        # TP2: After more candles, take smaller profit
        tp2_c = int(self.tp2_candles.value)
        tp2_p = float(self.tp2_min_pct.value) / 100
        if elapsed >= tp2_c and current_profit >= tp2_p:
            self._update_streak(pair, True)
            return "tp2_late_profit"

        # Early loss cut: if deep red after some time → don't wait for full SL
        el_c = int(self.early_loss_candles.value)
        el_p = float(self.early_loss_pct.value) / 100
        if elapsed >= el_c and current_profit <= -el_p:
            self._update_streak(pair, False)
            return "early_loss_cut"

        # Hard timeout
        tc = int(self.timeout_candles.value)
        if elapsed >= tc:
            is_win = current_profit > 0
            self._update_streak(pair, is_win)
            return "timeout"

        return None

    def _update_streak(self, pair, is_win):
        if is_win:
            self._consec_losses[pair] = 0
        else:
            self._consec_losses[pair] = self._consec_losses.get(pair, 0) + 1

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