# source: https://raw.githubusercontent.com/DOUGLASGUEDESATRIA/OSIRIS_TRADE/fe3c09567f3798d1cad6d7a306fa4f6131d91e17/user_data/strategies/OsirisScalperV2.py """ Github_DOUGLASGUEDESATRIA_OSIRIS_TRADE__OsirisScalperV2__20260329_060843 — Multi-Timeframe Pullback Scalper (5m) ======================================================== 1h: EMA cross + MACD histogram = directional bias 5m: BB pullback + reversal candle pattern = entry Exit: ATR-based TP/SL, trailing breakeven, timeout IS (Jan-Mar 2025): 352 trades, 53% WR, +5%, PF 1.25 OOS (Apr-Jun 2025): Longs +1.32%, Shorts -4.18% """ import logging 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 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 Github_DOUGLASGUEDESATRIA_OSIRIS_TRADE__OsirisScalperV2__20260329_060843(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 = 100 process_only_new_candles = True # ── Risk (sell) — ATR multipliers ──────────────────────────────── tp_atr = DecimalParameter(1.5, 5.0, default=3.0, decimals=1, space="sell", optimize=True) sl_atr = DecimalParameter(0.8, 3.0, default=1.5, decimals=1, space="sell", optimize=True) be_trigger = DecimalParameter(0.3, 2.0, default=1.0, decimals=1, space="sell", optimize=True) max_hold = IntParameter(6, 60, default=24, space="sell", optimize=True) # ── Entry (buy) ────────────────────────────────────────────────── ema_fast_1h = IntParameter(5, 15, default=8, space="buy", optimize=True) ema_slow_1h = IntParameter(15, 40, default=21, space="buy", optimize=True) bb_period = IntParameter(14, 30, default=20, space="buy", optimize=True) bb_std = DecimalParameter(1.5, 3.0, default=2.0, decimals=1, space="buy", optimize=True) pb_depth = DecimalParameter(0.2, 0.8, default=0.5, decimals=1, space="buy", optimize=True) pb_streak = IntParameter(1, 4, default=2, space="buy", optimize=True) vol_threshold = DecimalParameter(0.6, 2.0, default=1.0, decimals=1, space="buy", optimize=True) rsi_low = IntParameter(20, 45, default=35, space="buy", optimize=True) rsi_high = IntParameter(55, 80, default=65, space="buy", optimize=True) _cooldown: dict = {} _ema_fast_grid = [5, 8, 10, 12, 15] _ema_slow_grid = [15, 21, 25, 30, 40] _bb_period_grid = [14, 20, 26] _bb_std_grid = [1.5, 2.0, 2.5, 3.0] def informative_pairs(self): return [(p, "1h") for p in self.dp.current_whitelist()] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) for period in self._bb_period_grid: for std in self._bb_std_grid: s = f"{period}_{int(std * 10)}" bb = ta.BBANDS(dataframe, timeperiod=period, nbdevup=std, nbdevdn=std) dataframe[f"bb_u_{s}"] = bb["upperband"] dataframe[f"bb_m_{s}"] = bb["middleband"] dataframe[f"bb_l_{s}"] = bb["lowerband"] dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["vol_sma"] = dataframe["volume"].rolling(20).mean() dataframe["is_up"] = (dataframe["close"] > dataframe["open"]).astype(int) is_up = dataframe["is_up"].values streak_up = np.zeros(len(is_up), dtype=int) streak_dn = np.zeros(len(is_up), dtype=int) for i in range(1, len(is_up)): if is_up[i] == 1: streak_up[i] = streak_up[i - 1] + 1 else: streak_dn[i] = streak_dn[i - 1] + 1 dataframe["streak_up"] = streak_up dataframe["streak_dn"] = streak_dn # ── 1h informative merge ───────────────────────────────────── inf_1h = self.dp.get_pair_dataframe(metadata["pair"], "1h") if not inf_1h.empty: all_emas = sorted(set(self._ema_fast_grid + self._ema_slow_grid)) for p in all_emas: inf_1h[f"ema_{p}"] = ta.EMA(inf_1h, timeperiod=p) macd = ta.MACD(inf_1h, fastperiod=12, slowperiod=26, signalperiod=9) inf_1h["macdhist"] = macd["macdhist"] cols = ["date"] + [f"ema_{p}" for p in all_emas] + ["macdhist"] inf_sel = inf_1h[cols].copy() inf_sel = inf_sel.rename(columns={c: f"1h_{c}" for c in cols if c != "date"}) dataframe = dataframe.merge(inf_sel, on="date", how="left") for c in dataframe.columns: if c.startswith("1h_"): dataframe[c] = dataframe[c].ffill() return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pb_d = float(self.pb_depth.value) pb_s = int(self.pb_streak.value) ef = min(self._ema_fast_grid, key=lambda x: abs(x - int(self.ema_fast_1h.value))) es = min(self._ema_slow_grid, key=lambda x: abs(x - int(self.ema_slow_1h.value))) bb_p = min(self._bb_period_grid, key=lambda x: abs(x - int(self.bb_period.value))) bb_s_val = min([int(s * 10) for s in self._bb_std_grid], key=lambda x: abs(x - int(self.bb_std.value * 10))) bb_key = f"{bb_p}_{bb_s_val}" if f"1h_ema_{ef}" not in dataframe.columns: return dataframe ok = dataframe["rsi"].notna() & (dataframe["volume"] > 0) vol_ok = dataframe["volume"] > dataframe["vol_sma"] * float(self.vol_threshold.value) rsi_ok = (dataframe["rsi"] > int(self.rsi_low.value)) & (dataframe["rsi"] < int(self.rsi_high.value)) trend_bull = (dataframe[f"1h_ema_{ef}"] > dataframe[f"1h_ema_{es}"]) & (dataframe["1h_macdhist"] > 0) trend_bear = (dataframe[f"1h_ema_{ef}"] < dataframe[f"1h_ema_{es}"]) & (dataframe["1h_macdhist"] < 0) bb_w = dataframe[f"bb_u_{bb_key}"] - dataframe[f"bb_l_{bb_key}"] pullback_long = dataframe["close"] <= dataframe[f"bb_l_{bb_key}"] + bb_w * pb_d pullback_short = dataframe["close"] >= dataframe[f"bb_u_{bb_key}"] - bb_w * pb_d reversal_up = (dataframe["streak_dn"].shift(1) >= pb_s) & (dataframe["is_up"] == 1) reversal_dn = (dataframe["streak_up"].shift(1) >= pb_s) & (dataframe["is_up"] == 0) dataframe.loc[ok & trend_bull & pullback_long & reversal_up & vol_ok & rsi_ok, "enter_long"] = 1 dataframe.loc[ok & trend_bear & pullback_short & reversal_dn & vol_ok & rsi_ok, "enter_short"] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe def confirm_trade_entry(self, pair, order_type, amount, rate, time_in_force, current_time, entry_tag, side, **kwargs) -> bool: last = self._cooldown.get(pair) if last is not None and (current_time - last).total_seconds() < 300: return False self._cooldown[pair] = current_time return True def custom_stoploss(self, pair, trade: Trade, current_time, current_rate: float, current_profit: float, **kwargs) -> float: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return -0.01 atr = dataframe["atr"].iloc[-1] close = dataframe["close"].iloc[-1] atr_pct = atr / close if close > 0 else 0.003 sl = atr_pct * float(self.sl_atr.value) be = atr_pct * float(self.be_trigger.value) if current_profit >= be: is_short = getattr(trade, "is_short", False) return stoploss_from_open(0.0002, current_profit, is_short=is_short) return -sl def custom_exit(self, pair, trade, current_time, current_rate, current_profit, **kwargs): dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return None atr = dataframe["atr"].iloc[-1] close = dataframe["close"].iloc[-1] atr_pct = atr / close if close > 0 else 0.003 tp = atr_pct * float(self.tp_atr.value) if current_profit >= tp: return "tp_hit" if trade.open_date_utc: candles = (current_time - trade.open_date_utc).total_seconds() / 300 if candles >= self.max_hold.value: return "timeout" return None def leverage(self, pair, current_time, current_rate, proposed_leverage, max_leverage, entry_tag, side, **kwargs) -> float: return 1.0