from datetime import datetime from pandas import DataFrame import talib.abstract as ta from freqtrade.strategy import DecimalParameter, IStrategy, IntParameter, informative class BtcBearMarketShortV3(IStrategy): """BTC short-only bear-market research strategy. Two setups share the same priority: avoid trading unless higher timeframes are already bearish. This is a research strategy, not the active dry-run bot. """ 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.050, "180": 0.026, "540": 0.0, } stoploss = -0.060 trailing_stop = True trailing_stop_positive = 0.014 trailing_stop_positive_offset = 0.036 trailing_only_offset_is_reached = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False volume_breakdown_factor = DecimalParameter(0.8, 1.8, default=1.20, decimals=2, space="buy", optimize=False) max_candle_range_atr = DecimalParameter(1.5, 3.5, default=2.5, decimals=1, space="buy", optimize=False) atr_pct_min = DecimalParameter(0.001, 0.010, default=0.0025, decimals=4, space="buy", optimize=False) rally_rsi_1h_min = IntParameter(40, 52, default=45, space="buy", optimize=False) rally_rsi_1h_max = IntParameter(55, 70, default=65, space="buy", optimize=False) min_1h_rsi_not_oversold = IntParameter(24, 38, default=30, space="buy", optimize=False) min_4h_rsi_not_oversold = IntParameter(22, 38, default=28, space="buy", optimize=False) wick_body_factor = DecimalParameter(0.8, 2.0, default=1.2, decimals=1, space="buy", optimize=False) atr_stop_mult = DecimalParameter(1.0, 2.5, default=1.5, decimals=1, space="sell", optimize=False) profit_take_rsi = IntParameter(18, 32, default=25, space="sell", optimize=False) @property def protections(self) -> list[dict]: return [ {"method": "CooldownPeriod", "stop_duration_candles": 4}, { "method": "StoplossGuard", "lookback_period_candles": 96, "trade_limit": 2, "stop_duration_candles": 24, "required_profit": 0.0, "only_per_pair": False, "only_per_side": True, }, { "method": "MaxDrawdown", "calculation_mode": "equity", "lookback_period_candles": 192, "trade_limit": 8, "stop_duration_candles": 32, "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_100"] = ta.EMA(dataframe, timeperiod=100) dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200) dataframe["rsi"] = ta.RSI(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) return dataframe @informative("4h") def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50) dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200) dataframe["rsi"] = ta.RSI(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_50"] = ta.EMA(dataframe, timeperiod=50) dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200) dataframe["rsi"] = ta.RSI(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(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["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_20"] = dataframe["low"].rolling(20, min_periods=20).min().shift(1) bollinger = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0) dataframe["bb_upper"] = bollinger["upperband"] dataframe["bb_middle"] = bollinger["middleband"] dataframe["bb_lower"] = bollinger["lowerband"] dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"] dataframe["range_atr"] = (dataframe["high"] - dataframe["low"]) / dataframe["atr"] dataframe["body"] = (dataframe["close"] - dataframe["open"]).abs() dataframe["upper_wick"] = dataframe["high"] - dataframe[["open", "close"]].max(axis=1) dataframe["bearish_engulfing"] = ( (dataframe["close"] < dataframe["open"]) & (dataframe["close"].shift(1) > dataframe["open"].shift(1)) & (dataframe["open"] >= dataframe["close"].shift(1)) & (dataframe["close"] <= dataframe["open"].shift(1)) ) 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 daily_bear = ( (dataframe["close_1d"] < dataframe["ema_200_1d"]) | (dataframe["ema_50_1d"] < dataframe["ema_200_1d"]) ) four_h_bear = ( (dataframe["ema_50_4h"] < dataframe["ema_200_4h"]) & (dataframe["close_4h"] < dataframe["ema_50_4h"]) & (dataframe["rsi_4h"] > self.min_4h_rsi_not_oversold.value) ) one_h_breakdown_ready = ( (dataframe["ema_20_1h"] < dataframe["ema_50_1h"]) & (dataframe["rsi_1h"] < 50) & (dataframe["rsi_1h"] > self.min_1h_rsi_not_oversold.value) ) not_panic_candle = ( (dataframe["atr"] > 0) & (dataframe["atr_pct"] >= self.atr_pct_min.value) & (dataframe["range_atr"] <= self.max_candle_range_atr.value) ) volume_breakdown_ok = ( (dataframe["volume"] > 0) & (dataframe["volume_mean_20"] > 0) & (dataframe["volume"] > dataframe["volume_mean_20"] * self.volume_breakdown_factor.value) ) breakdown = ( daily_bear & four_h_bear & one_h_breakdown_ready & not_panic_candle & volume_breakdown_ok & (dataframe["close"] < dataframe["donchian_low_20"]) & (dataframe["close"] < dataframe["open"]) ) rally_regime = ( (dataframe["close_4h"] < dataframe["ema_200_4h"]) & (dataframe["ema_50_4h"] < dataframe["ema_200_4h"]) & (dataframe["close_1h"] < dataframe["ema_100_1h"]) & (dataframe["rsi_1h"] >= self.rally_rsi_1h_min.value) & (dataframe["rsi_1h"] <= self.rally_rsi_1h_max.value) ) bb_touch_recent = ( (dataframe["high"] >= dataframe["bb_upper"]) | (dataframe["high"].shift(1) >= dataframe["bb_upper"].shift(1)) ) ema20_reclaim_failed = ( (dataframe["close"] < dataframe["ema_20"]) & ( (dataframe["close"].shift(1) >= dataframe["ema_20"].shift(1)) | bb_touch_recent ) ) rejection_candle = ( (dataframe["upper_wick"] > dataframe["body"].clip(lower=dataframe["atr"] * 0.05) * self.wick_body_factor.value) | dataframe["bearish_engulfing"] ) bear_rally_rejection = ( rally_regime & not_panic_candle & bb_touch_recent & ema20_reclaim_failed & rejection_candle ) dataframe.loc[breakdown, ["enter_short", "enter_tag"]] = (1, "v3_breakdown_continuation") dataframe.loc[bear_rally_rejection, ["enter_short", "enter_tag"]] = (1, "v3_bear_rally_rejection") 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 len(dataframe) < 2 or not trade.is_short: return None candle = dataframe.iloc[-1] previous = dataframe.iloc[-2] atr_stop_rate = trade.open_rate + candle["atr"] * self.atr_stop_mult.value if current_rate > atr_stop_rate: return "v3_atr_invalidation" if current_profit > 0 and current_rate > candle["ema_20"]: return "v3_profit_ema20_reclaim" if candle["close_1h"] > candle["ema_50_1h"]: return "v3_1h_ema50_reclaim" momentum_slowing = (candle["rsi"] > previous["rsi"]) or (candle["close"] > previous["close"]) if current_profit > 0.012 and candle["rsi"] < self.profit_take_rsi.value and momentum_slowing: return "v3_oversold_momentum_slow" 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 class BtcBearMarketShortV3RallyOnly(BtcBearMarketShortV3): """Research variant: keep only the bear-rally rejection setup. The full V3 backtest showed that breakdown continuation was destructive. This variant isolates the more promising setup instead of tuning both at once. """ def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe = super().populate_entry_trend(dataframe, metadata) dataframe.loc[dataframe["enter_tag"] != "v3_bear_rally_rejection", "enter_short"] = 0 return dataframe