from datetime import datetime from pandas import DataFrame import talib.abstract as ta from freqtrade.strategy import DecimalParameter, IStrategy, IntParameter, informative class TrendRegimeBtcCycleV1(IStrategy): """BTC-first long/short research strategy. This variant intentionally ignores non-BTC pairs. It is meant to test the thesis that BTC carries the major market regime while SOL adds execution noise. Keep it in dry-run/backtest until it has enough evidence. """ INTERFACE_VERSION = 3 can_short = True timeframe = "15m" startup_candle_count = 250 process_only_new_candles = True position_adjustment_enable = False max_entry_position_adjustment = 0 minimal_roi = { "0": 0.055, "240": 0.028, "720": 0.0, } stoploss = -0.05 trailing_stop = True trailing_stop_positive = 0.018 trailing_stop_positive_offset = 0.04 trailing_only_offset_is_reached = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False rsi_long_min = IntParameter(42, 56, default=48, space="buy", optimize=False) rsi_long_max = IntParameter(58, 72, default=64, space="buy", optimize=False) rsi_short_min = IntParameter(30, 44, default=36, space="buy", optimize=False) rsi_short_max = IntParameter(44, 58, default=50, space="buy", optimize=False) adx_min = IntParameter(18, 35, default=24, space="buy", optimize=False) adx_1h_min = IntParameter(16, 35, default=20, space="buy", optimize=False) adx_4h_min = IntParameter(12, 30, default=16, space="buy", optimize=False) volume_min_factor = DecimalParameter(0.2, 1.2, default=0.45, decimals=2, space="buy", optimize=False) ema_spread_min = DecimalParameter(0.001, 0.02, default=0.003, decimals=3, space="buy", optimize=False) atr_pct_min = DecimalParameter(0.001, 0.02, default=0.002, decimals=3, space="buy", optimize=False) atr_pct_max = DecimalParameter(0.015, 0.08, default=0.05, decimals=3, space="buy", optimize=False) pullback_atr_max = DecimalParameter(0.6, 2.5, default=1.5, decimals=1, space="buy", optimize=False) cycle_rsi_long_min = IntParameter(48, 62, default=52, space="buy", optimize=False) cycle_rsi_short_max = IntParameter(38, 52, default=48, space="buy", optimize=False) rsi_long_exit = IntParameter(68, 84, default=74, space="sell", optimize=False) rsi_short_exit = IntParameter(16, 34, default=28, space="sell", optimize=False) @property def protections(self) -> list[dict]: return [ { "method": "CooldownPeriod", "stop_duration_candles": 2, }, { "method": "StoplossGuard", "lookback_period_candles": 48, "trade_limit": 3, "stop_duration_candles": 12, "required_profit": 0.0, "only_per_pair": False, "only_per_side": False, }, { "method": "MaxDrawdown", "calculation_mode": "equity", "lookback_period_candles": 96, "trade_limit": 10, "stop_duration_candles": 24, "max_allowed_drawdown": 0.10, }, ] @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_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["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(3) 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["adx"] = ta.ADX(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(3) return dataframe def populate_indicators(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["adx"] = ta.ADX(dataframe, timeperiod=14) dataframe["plus_di"] = ta.PLUS_DI(dataframe, timeperiod=14) dataframe["minus_di"] = ta.MINUS_DI(dataframe, timeperiod=14) dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) dataframe["volume_mean_20"] = dataframe["volume"].rolling(20, min_periods=20).mean() dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"] dataframe["ema_spread"] = (dataframe["ema_50"] - dataframe["ema_200"]).abs() / dataframe["close"] dataframe["ema_50_slope"] = dataframe["ema_50"] - dataframe["ema_50"].shift(3) 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.volume_min_factor.value) ) volatility_ok = ( (dataframe["atr_pct"] >= self.atr_pct_min.value) & (dataframe["atr_pct"] <= self.atr_pct_max.value) & (dataframe["ema_spread"] >= self.ema_spread_min.value) ) long_pullback = ( (dataframe["close"] > dataframe["ema_50"]) & ( (dataframe["close"].shift(1) <= dataframe["ema_50"].shift(1)) | ((dataframe["low"] <= dataframe["ema_50"]) & (dataframe["close"] > dataframe["open"])) ) & (((dataframe["close"] - dataframe["ema_50"]) / dataframe["atr"]) <= self.pullback_atr_max.value) ) short_pullback = ( (dataframe["close"] < dataframe["ema_50"]) & ( (dataframe["close"].shift(1) >= dataframe["ema_50"].shift(1)) | ((dataframe["high"] >= dataframe["ema_50"]) & (dataframe["close"] < dataframe["open"])) ) & (((dataframe["ema_50"] - dataframe["close"]) / dataframe["atr"]) <= self.pullback_atr_max.value) ) long_conditions = ( volume_ok & volatility_ok & (dataframe["close"] > dataframe["ema_200"]) & (dataframe["ema_50"] > dataframe["ema_200"]) & (dataframe["ema_50_slope"] > 0) & (dataframe["plus_di"] > dataframe["minus_di"]) & (dataframe["close_1h"] > dataframe["ema_200_1h"]) & (dataframe["ema_50_1h"] > dataframe["ema_200_1h"]) & (dataframe["ema_50_slope_1h"] > 0) & (dataframe["plus_di_1h"] > dataframe["minus_di_1h"]) & (dataframe["rsi_1h"] > 50) & (dataframe["close_4h"] > dataframe["ema_200_4h"]) & (dataframe["ema_50_4h"] > dataframe["ema_200_4h"]) & (dataframe["ema_50_slope_4h"] > 0) & (dataframe["plus_di_4h"] > dataframe["minus_di_4h"]) & (dataframe["rsi_4h"] > self.cycle_rsi_long_min.value) & (dataframe["adx"] > self.adx_min.value) & (dataframe["adx_1h"] > self.adx_1h_min.value) & (dataframe["adx_4h"] > self.adx_4h_min.value) & (dataframe["rsi"] > self.rsi_long_min.value) & (dataframe["rsi"] < self.rsi_long_max.value) & long_pullback ) short_conditions = ( volume_ok & volatility_ok & (dataframe["close"] < dataframe["ema_200"]) & (dataframe["ema_50"] < dataframe["ema_200"]) & (dataframe["ema_50_slope"] < 0) & (dataframe["minus_di"] > dataframe["plus_di"]) & (dataframe["close_1h"] < dataframe["ema_200_1h"]) & (dataframe["ema_50_1h"] < dataframe["ema_200_1h"]) & (dataframe["ema_50_slope_1h"] < 0) & (dataframe["minus_di_1h"] > dataframe["plus_di_1h"]) & (dataframe["rsi_1h"] < 50) & (dataframe["close_4h"] < dataframe["ema_200_4h"]) & (dataframe["ema_50_4h"] < dataframe["ema_200_4h"]) & (dataframe["ema_50_slope_4h"] < 0) & (dataframe["minus_di_4h"] > dataframe["plus_di_4h"]) & (dataframe["rsi_4h"] < self.cycle_rsi_short_max.value) & (dataframe["adx"] > self.adx_min.value) & (dataframe["adx_1h"] > self.adx_1h_min.value) & (dataframe["adx_4h"] > self.adx_4h_min.value) & (dataframe["rsi"] > self.rsi_short_min.value) & (dataframe["rsi"] < self.rsi_short_max.value) & short_pullback ) dataframe.loc[long_conditions, ["enter_long", "enter_tag"]] = (1, "btc_cycle_long_pullback") dataframe.loc[short_conditions, ["enter_short", "enter_tag"]] = (1, "btc_cycle_short_pullback") 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 trade.is_short: if current_profit > 0.025 and candle["rsi"] < self.rsi_short_exit.value: return "btc_short_rsi_profit_take" if current_profit > 0.012 and current_rate > candle["ema_50"]: return "btc_short_profit_protect_ema50" if ( current_profit < -0.008 and candle["close_1h"] > candle["ema_50_1h"] and candle["plus_di_1h"] > candle["minus_di_1h"] ): return "btc_short_1h_reversal" if candle["ema_50_1h"] > candle["ema_200_1h"] and current_rate > candle["ema_200"]: return "btc_short_regime_lost" return None if current_profit > 0.025 and candle["rsi"] > self.rsi_long_exit.value: return "btc_long_rsi_profit_take" if current_profit > 0.012 and current_rate < candle["ema_50"]: return "btc_long_profit_protect_ema50" if current_rate < candle["ema_50"]: return "btc_long_ema50_break" if ( current_profit < -0.008 and candle["close_1h"] < candle["ema_50_1h"] and candle["minus_di_1h"] > candle["plus_di_1h"] ): return "btc_long_1h_reversal" if candle["ema_50_1h"] < candle["ema_200_1h"] and current_rate < candle["ema_200"]: return "btc_long_regime_lost" 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