# source: https://raw.githubusercontent.com/DOUGLASGUEDESATRIA/OSIRIS_TRADE/fe3c09567f3798d1cad6d7a306fa4f6131d91e17/user_data/strategies/OsirisGoldV8.py """ Github_DOUGLASGUEDESATRIA_OSIRIS_TRADE__OsirisGoldV8__20260329_060843 — 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 Github_DOUGLASGUEDESATRIA_OSIRIS_TRADE__OsirisGoldV8__20260329_060843(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