from datetime import datetime from pandas import DataFrame import talib.abstract as ta from freqtrade.strategy import DecimalParameter, IStrategy, IntParameter, informative class TrendRegimeV2(IStrategy): """Stricter long/short trend strategy intended for comparison against TrendRegimeV1.""" 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.035, "120": 0.018, "300": 0.0, } stoploss = -0.045 trailing_stop = True trailing_stop_positive = 0.012 trailing_stop_positive_offset = 0.026 trailing_only_offset_is_reached = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False rsi_long_min = IntParameter(40, 55, default=46, space="buy", optimize=False) rsi_long_max = IntParameter(58, 72, default=66, space="buy", optimize=False) rsi_short_min = IntParameter(30, 45, default=38, space="buy", optimize=False) rsi_short_max = IntParameter(46, 60, default=52, 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) volume_min_factor = DecimalParameter(0.2, 1.2, default=0.5, 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.045, decimals=3, space="buy", optimize=False) pullback_atr_max = DecimalParameter(0.6, 2.5, default=1.4, decimals=1, space="buy", optimize=False) rsi_long_exit = IntParameter(72, 88, default=78, space="sell", optimize=False) rsi_short_exit = IntParameter(12, 32, default=24, 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, }, ] @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) 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) 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 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["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["plus_di_4h"] > dataframe["minus_di_4h"]) & (dataframe["adx"] > self.adx_min.value) & (dataframe["adx_1h"] > self.adx_1h_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["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["minus_di_4h"] > dataframe["plus_di_4h"]) & (dataframe["adx"] > self.adx_min.value) & (dataframe["adx_1h"] > self.adx_1h_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, "v2_long_confirmed_pullback") dataframe.loc[short_conditions, ["enter_short", "enter_tag"]] = (1, "v2_short_confirmed_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 candle["ema_50"] > candle["ema_200"]: return "short_regime_lost" if current_rate > candle["ema_50"]: return "short_ema50_break" if current_profit > 0.01 and candle["rsi"] < self.rsi_short_exit.value: return "short_rsi_profit_take" return None if candle["ema_50"] < candle["ema_200"]: return "long_regime_lost" if current_rate < candle["ema_50"]: return "long_ema50_break" if current_profit > 0.01 and candle["rsi"] > self.rsi_long_exit.value: return "long_rsi_profit_take" 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