from datetime import datetime from pandas import DataFrame import talib.abstract as ta from freqtrade.strategy import DecimalParameter, IStrategy, IntParameter, informative class BearMarketShortV2(IStrategy): """BTC-only research strategy for bear-market short setups. The strategy is deliberately not an always-on short bot. It looks for a bearish BTC regime, then waits for either a retest rejection or a controlled breakdown. It is meant for backtesting and dry-run research only. """ INTERFACE_VERSION = 3 can_short = True timeframe = "15m" startup_candle_count = 420 process_only_new_candles = True position_adjustment_enable = False max_entry_position_adjustment = 0 minimal_roi = { "0": 0.055, "180": 0.028, "480": 0.0, } stoploss = -0.055 trailing_stop = True trailing_stop_positive = 0.014 trailing_stop_positive_offset = 0.034 trailing_only_offset_is_reached = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False adx_min = IntParameter(18, 35, default=23, space="buy", optimize=False) rsi_min = IntParameter(24, 40, default=31, space="buy", optimize=False) rsi_max = IntParameter(45, 62, default=55, space="buy", optimize=False) daily_rsi_min = IntParameter(24, 40, default=31, space="buy", optimize=False) daily_rsi_max = IntParameter(42, 58, default=52, space="buy", optimize=False) max_daily_ema50_atr_extension = DecimalParameter(1.0, 4.0, default=2.4, decimals=1, space="buy", optimize=False) max_15m_ema50_atr_extension = DecimalParameter(0.8, 3.0, default=1.7, decimals=1, space="buy", optimize=False) min_volume_factor = DecimalParameter(0.3, 1.5, default=0.65, decimals=2, space="buy", optimize=False) retest_buffer = DecimalParameter(0.000, 0.010, default=0.003, decimals=3, space="buy", optimize=False) ema50_exit_buffer = DecimalParameter(0.000, 0.012, default=0.004, decimals=3, space="sell", optimize=False) profit_take_rsi = IntParameter(18, 35, default=26, space="sell", optimize=False) @property def protections(self) -> list[dict]: return [ { "method": "CooldownPeriod", "stop_duration_candles": 3, }, { "method": "StoplossGuard", "lookback_period_candles": 72, "trade_limit": 2, "stop_duration_candles": 16, "required_profit": 0.0, "only_per_pair": False, "only_per_side": True, }, { "method": "MaxDrawdown", "calculation_mode": "equity", "lookback_period_candles": 144, "trade_limit": 8, "stop_duration_candles": 24, "max_allowed_drawdown": 0.08, }, ] @staticmethod def _is_btc_pair(pair: str) -> bool: return pair.upper().startswith("BTC/") @informative("1h") def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20) dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50) dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200) dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["adx"] = ta.ADX(dataframe, timeperiod=14) dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14) dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14) dataframe["ema_50_slope"] = dataframe["ema_50"] - dataframe["ema_50"].shift(6) return dataframe @informative("4h") def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20) dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50) dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200) dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["adx"] = ta.ADX(dataframe, timeperiod=14) dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14) dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14) dataframe["ema_50_slope"] = dataframe["ema_50"] - dataframe["ema_50"].shift(4) return dataframe @informative("1d") def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20) dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50) dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200) dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["adx"] = ta.ADX(dataframe, timeperiod=14) dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14) dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14) dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"] dataframe["dist_ema200_pct"] = (dataframe["ema_200"] - dataframe["close"]) / dataframe["ema_200"] dataframe["ema_20_slope_atr"] = (dataframe["ema_20"] - dataframe["ema_20"].shift(5)) / dataframe["atr"] dataframe["ema_50_slope"] = dataframe["ema_50"] - dataframe["ema_50"].shift(5) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20) dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50) dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200) dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["adx"] = ta.ADX(dataframe, timeperiod=14) dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14) dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14) dataframe["volume_mean_20"] = dataframe["volume"].rolling(20, min_periods=20).mean() dataframe["donchian_low_40"] = dataframe["low"].rolling(40, min_periods=40).min().shift(1) dataframe["range_atr"] = (dataframe["high"] - dataframe["low"]) / dataframe["atr"] dataframe["ema50_extension_atr"] = (dataframe["ema_50"] - dataframe["close"]) / dataframe["atr"] dataframe["ema_50_slope"] = dataframe["ema_50"] - dataframe["ema_50"].shift(6) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["enter_long"] = 0 dataframe["enter_short"] = 0 if not self._is_btc_pair(metadata["pair"]): return dataframe volume_ok = ( (dataframe["volume"] > 0) & (dataframe["volume_mean_20"] > 0) & (dataframe["volume"] >= dataframe["volume_mean_20"] * self.min_volume_factor.value) ) daily_extension = (dataframe["ema_50_1d"] - dataframe["close_1d"]) / dataframe["atr_1d"] daily_risk_off = ( ( (dataframe["close_1d"] < dataframe["ema_50_1d"]) & (dataframe["ema_50_slope_1d"] < 0) ) | ( (dataframe["close_1d"] < dataframe["ema_200_1d"]) & (dataframe["minus_di_1d"] > dataframe["plus_di_1d"]) ) ) higher_tf_bear = ( daily_risk_off & (daily_extension < self.max_daily_ema50_atr_extension.value) & (dataframe["rsi_1d"] > self.daily_rsi_min.value) & (dataframe["rsi_1d"] < self.daily_rsi_max.value) & (dataframe["close_4h"] < dataframe["ema_50_4h"]) & (dataframe["ema_50_slope_4h"] < 0) & (dataframe["minus_di_4h"] > dataframe["plus_di_4h"]) & (dataframe["close_1h"] < dataframe["ema_200_1h"]) & (dataframe["ema_50_slope_1h"] < 0) & (dataframe["minus_di_1h"] > dataframe["plus_di_1h"]) ) not_late = ( (dataframe["ema50_extension_atr"] < self.max_15m_ema50_atr_extension.value) & (dataframe["rsi"] > self.rsi_min.value) & (dataframe["rsi"] < self.rsi_max.value) & (dataframe["range_atr"] < 3.5) ) base_short = ( volume_ok & higher_tf_bear & not_late & (dataframe["close"] < dataframe["ema_200"]) & (dataframe["ema_50"] < dataframe["ema_200"]) & (dataframe["ema_50_slope"] < 0) & (dataframe["minus_di"] > dataframe["plus_di"]) & (dataframe["adx"] > self.adx_min.value) ) retest_rejection = ( base_short & (dataframe["high"] >= dataframe["ema_50"] * (1 - self.retest_buffer.value)) & (dataframe["close"] < dataframe["ema_20"]) & (dataframe["close"] < dataframe["open"]) ) breakdown = ( base_short & (dataframe["close"] < dataframe["donchian_low_40"]) & (dataframe["close"] < dataframe["open"]) & (dataframe["adx_1h"] > self.adx_min.value) & (dataframe["rsi_1h"] > self.rsi_min.value) ) dataframe.loc[retest_rejection, ["enter_short", "enter_tag"]] = (1, "bear_v2_retest_rejection") dataframe.loc[breakdown, ["enter_short", "enter_tag"]] = (1, "bear_v2_breakdown") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["exit_long"] = 0 dataframe["exit_short"] = 0 return dataframe def custom_exit( self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> str | bool | None: if not self.dp: return None dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return None candle = dataframe.iloc[-1] if not trade.is_short: return None if current_profit > 0.025 and candle["rsi"] < self.profit_take_rsi.value: return "bear_v2_rsi_profit_take" if current_profit > 0.015 and current_rate > candle["ema_20"]: return "bear_v2_profit_ema20_reclaim" if current_rate > candle["ema_50"] * (1 + self.ema50_exit_buffer.value): return "bear_v2_ema50_reclaim" if candle["close_1h"] > candle["ema_50_1h"] and current_profit < 0.01: return "bear_v2_1h_reclaim" return None def leverage( self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str, **kwargs, ) -> float: return 1.0