OsirisSqueeze

"""
OsirisSqueeze v1 — Volatility Squeeze + Trend Momentum (1h BTC Futures)
=========================================================================
Concept: Enter during volatility contraction (squeeze) inside an uptrend,
ride the expansion phase. Price tends to continue in the trend direction
after a squeeze, producing consistent positive drift.

Walk-Forward Validation (2025-2026 BTC futures, 0.04% maker fee):
  IS  (Jan-Jul 2025): 14 trades, 50% WR, +0.58% avg, +8.0% total, 2.0% DD
  OOS (Jul 2025-Mar 2026): 8 trades, 75% WR, +1.52% avg, +12.7% total, 1.0% DD
  Full: 30 trades, 57% WR, +0.42% avg, +13.0% total, 3.6% DD

  Grid search: 6048 combos tested, 1371 IS/OOS viable.
  Top config: EMA 8/21/40, SQZ_SMA=48, SF=0.7, RSI<50, SL=2ATR, TP=12ATR, Hold=48h

Key insight: timeouts average +0.81-1.24% profit because uptrend drift
carries the position. The 6R target captures outlier breakout moves.

100% proprietary. Developed for OSIRIS.
"""

import logging
from datetime import datetime, timezone, timedelta
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

logger = logging.getLogger(__name__)


class OsirisSqueeze(IStrategy):
    """
    Volatility Squeeze + Uptrend Continuation on 1h.
    Long-only. Enters during ATR compression in a trending market.
    """
    INTERFACE_VERSION = 3
    can_short = False   # long-only (shorts showed inconsistent IS/OOS)
    timeframe = "1h"

    stoploss = -0.03    # safety net — custom_exit handles ATR-based SL via candle low
    minimal_roi = {"0": 100}   # disabled — custom exit handles TP
    trailing_stop = False
    use_custom_stoploss = False
    startup_candle_count = 100
    process_only_new_candles = True

    # ── Entry parameters (buy space) ────────────────────────────────
    squeeze_factor = DecimalParameter(0.5, 0.95, default=0.70, decimals=2, space="buy", optimize=True)
    atr_sma_period = IntParameter(0, 2, default=2, space="buy", optimize=True)  # index into grid
    ema_fast = IntParameter(0, 2, default=0, space="buy", optimize=True)        # index into grid
    ema_slow = IntParameter(0, 2, default=1, space="buy", optimize=True)        # index into grid
    ema_trend = IntParameter(0, 2, default=0, space="buy", optimize=True)       # index into grid
    rsi_max = IntParameter(40, 80, default=50, space="buy", optimize=True)

    # Grids (indexed by parameter value)
    _ema_fast_vals = [8, 10, 12]
    _ema_slow_vals = [15, 21, 25]
    _ema_trend_vals = [40, 50, 60]
    _atr_sma_vals = [24, 36, 48]

    # ── Exit parameters (sell space) ─────────────────────────────────
    sl_atr_mult = DecimalParameter(1.5, 3.0, default=2.0, decimals=1, space="sell", optimize=True)
    tp_atr_mult = DecimalParameter(6.0, 16.0, default=12.0, decimals=1, space="sell", optimize=True)
    max_hold_hours = IntParameter(12, 72, default=48, space="sell", optimize=True)

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

        # EMAs — compute all grid values
        for p in sorted(set(self._ema_fast_vals + self._ema_slow_vals + self._ema_trend_vals)):
            dataframe[f"ema_{p}"] = ta.EMA(dataframe, timeperiod=p)

        # RSI
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=7)

        # ATR SMA for squeeze detection
        for sp in self._atr_sma_vals:
            dataframe[f"atr_sma_{sp}"] = dataframe["atr"].rolling(sp).mean()

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        ef = self._ema_fast_vals[self.ema_fast.value]
        es = self._ema_slow_vals[self.ema_slow.value]
        et = self._ema_trend_vals[self.ema_trend.value]
        sp = self._atr_sma_vals[self.atr_sma_period.value]
        sf = self.squeeze_factor.value
        rm = self.rsi_max.value

        ema_f = dataframe[f"ema_{ef}"]
        ema_s = dataframe[f"ema_{es}"]
        ema_t = dataframe[f"ema_{et}"]
        atr_sma = dataframe[f"atr_sma_{sp}"]
        rsi = dataframe["rsi"]

        # Squeeze: ATR below threshold of its moving average
        squeeze = dataframe["atr"] < (atr_sma * sf)

        # Uptrend: EMA cascade
        uptrend = (ema_f > ema_s) & (ema_s > ema_t)

        # RSI filter: not overbought (optional tightening)
        rsi_ok = rsi < rm

        # Volume: at least some trading (avoid dead candles)
        vol_ok = dataframe["volume"] > 0

        dataframe.loc[
            squeeze & uptrend & rsi_ok & vol_ok,
            "enter_long",
        ] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # All exits via custom_stoploss and custom_exit
        dataframe["exit_long"] = 0
        return dataframe

    def _get_entry_atr(self, trade: Trade, pair: str) -> float:
        """Get and cache the RAW ATR from the entry candle."""
        cached = trade.get_custom_data("entry_atr")
        if cached is not None:
            return float(cached)
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe is None or len(dataframe) < 2:
            return 0.0
        entry_candle = dataframe.loc[dataframe["date"] <= trade.open_date_utc]
        if len(entry_candle) == 0:
            return 0.0
        val = entry_candle.iloc[-1]["atr"]
        if np.isnan(val) or val <= 0:
            return 0.0
        trade.set_custom_data("entry_atr", val)
        return val

    def custom_exit(self, pair: str, trade: Trade, current_time: datetime,
                    current_rate: float, current_profit: float, **kwargs):
        atr = self._get_entry_atr(trade, pair)
        if atr <= 0:
            return None

        open_rate = trade.open_rate

        # Get current candle low/high for precise SL/TP check
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe is None or len(dataframe) < 1:
            return None
        last = dataframe.iloc[-1]

        # SL: check if candle low reached the ATR-based stop price
        sl_price = open_rate - (atr * self.sl_atr_mult.value)
        if last["low"] <= sl_price:
            return "atr_stoploss"

        # TP: check if candle high reached the ATR-based target price
        tp_price = open_rate + (atr * self.tp_atr_mult.value)
        if last["high"] >= tp_price:
            return "atr_tp"

        # Timeout: max hold hours
        td = current_time - trade.open_date_utc
        if td >= timedelta(hours=self.max_hold_hours.value):
            return "timeout_exit"

        return None