"""BTC/USDT multi-timeframe swing momentum strategy, V5.4 Lower ADX Gate. Single isolated modification relative to BTCMomentumV5_Base: - Decreased ADX_4H_THRESHOLD from 18.0 to 14.0. Hypothesis: Reducing ADX_4H_THRESHOLD from 18.0 to 14.0 permits entries during earlier trend inception phases, increasing trade frequency while preserving the macro daily and 4h moving average direction. """ from datetime import datetime from math import isfinite import pandas as pd import talib.abstract as ta from pandas import DataFrame from freqtrade.persistence import Trade from freqtrade.strategy import IStrategy, merge_informative_pair, stoploss_from_absolute class BTCMomentumV5_4_LowerADXGate(IStrategy): INTERFACE_VERSION = 3 timeframe = "1h" informative_timeframes = ("4h", "1d") can_short = False minimal_roi = {} use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # Fixed safety floor in case analyzed candle data is temporarily unavailable. stoploss = -0.10 use_custom_stoploss = True trailing_stop = False process_only_new_candles = True order_types = { "entry": "market", "exit": "market", "stoploss": "market", "stoploss_on_exchange": False, } startup_candle_count: int = 250 # 1h Breakout and volatility parameters BREAKOUT_LOOKBACK = 20 ATR_1H_PERIOD = 14 ATR_REFERENCE_PERIOD = 50 ATR_EXPANSION_MULTIPLIER = 1.05 VOLUME_SMA_PERIOD = 20 VOLUME_EXPANSION_MULTIPLIER = 1.05 BB_PERIOD = 20 BB_STDDEV = 2.0 BANDWIDTH_QUANTILE_PERIOD = 100 BANDWIDTH_QUANTILE = 0.50 # 1h Momentum indicators RSI_PERIOD = 14 RSI_MIN = 50.0 RSI_MAX = 75.0 # Macro trend parameters (ADX_4H_THRESHOLD changed from 18.0 to 14.0) EMA_4H_FAST = 20 EMA_4H_SLOW = 50 ADX_4H_PERIOD = 14 ADX_4H_THRESHOLD = 14.0 EMA_1D_FAST = 20 EMA_1D_SLOW = 50 EMA_1D_LONG = 200 # Risk and trailing ratchet parameters INITIAL_ATR_MULTIPLIER = 3.00 BE_ACTIVATION_PROFIT = 0.0100 BE_LOCK_PROFIT = 0.0020 TRAIL_ACTIVATION_PROFIT = 0.0150 TRAILING_ATR_MULTIPLIER = 2.50 _ENTRY_ATR_KEY = "btc_momentum_v5_4_entry_atr" _HIGHEST_CLOSE_KEY = "btc_momentum_v5_4_highest_close" _EFFECTIVE_STOP_KEY = "btc_momentum_v5_4_effective_stop" _BE_LOCKED_KEY = "btc_momentum_v5_4_be_locked" _TRAIL_ACTIVE_KEY = "btc_momentum_v5_4_trail_active" def informative_pairs(self): """Load native 4h and 1d candles for every configured spot pair.""" return [ (pair, timeframe) for pair in self.dp.current_whitelist() for timeframe in self.informative_timeframes ] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # -------------------------- native 1h --------------------------- dataframe["atr_14"] = ta.ATR( dataframe, timeperiod=self.ATR_1H_PERIOD, ) bb_upper, bb_middle, bb_lower = ta.BBANDS( dataframe["close"], timeperiod=self.BB_PERIOD, nbdevup=self.BB_STDDEV, nbdevdn=self.BB_STDDEV, matype=0, ) dataframe["bb_upper"] = bb_upper dataframe["bb_middle"] = bb_middle dataframe["bb_lower"] = bb_lower dataframe["bb_bandwidth"] = ( (dataframe["bb_upper"] - dataframe["bb_lower"]) / dataframe["bb_middle"] ) prior_bandwidth = dataframe["bb_bandwidth"].shift(1) dataframe["prior_bandwidth_q50"] = prior_bandwidth.rolling( self.BANDWIDTH_QUANTILE_PERIOD, min_periods=self.BANDWIDTH_QUANTILE_PERIOD, ).quantile(self.BANDWIDTH_QUANTILE) dataframe["prior_squeeze"] = ( prior_bandwidth <= dataframe["prior_bandwidth_q50"] ).fillna(False) dataframe["previous_breakout_level"] = ( dataframe["high"] .shift(1) .rolling(self.BREAKOUT_LOOKBACK, min_periods=self.BREAKOUT_LOOKBACK) .max() ) dataframe["previous_atr_reference"] = ( dataframe["atr_14"] .shift(1) .rolling( self.ATR_REFERENCE_PERIOD, min_periods=self.ATR_REFERENCE_PERIOD, ) .mean() ) dataframe["previous_volume_sma"] = ( dataframe["volume"] .shift(1) .rolling( self.VOLUME_SMA_PERIOD, min_periods=self.VOLUME_SMA_PERIOD, ) .mean() ) dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20) dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50) dataframe["rsi_14"] = ta.RSI(dataframe, timeperiod=self.RSI_PERIOD) dataframe["adx_14"] = ta.ADX(dataframe, timeperiod=14) # -------------------------- native 4h --------------------------- informative_4h = self.dp.get_pair_dataframe( pair=metadata["pair"], timeframe="4h", ).copy() informative_4h["ema_20"] = ta.EMA( informative_4h, timeperiod=self.EMA_4H_FAST, ) informative_4h["ema_50"] = ta.EMA( informative_4h, timeperiod=self.EMA_4H_SLOW, ) informative_4h["adx_14"] = ta.ADX( informative_4h, timeperiod=self.ADX_4H_PERIOD, ) dataframe = merge_informative_pair( dataframe, informative_4h, self.timeframe, "4h", ffill=True, ) # -------------------------- native 1d --------------------------- informative_1d = self.dp.get_pair_dataframe( pair=metadata["pair"], timeframe="1d", ).copy() informative_1d["ema_20"] = ta.EMA( informative_1d, timeperiod=self.EMA_1D_FAST, ) informative_1d["ema_50"] = ta.EMA( informative_1d, timeperiod=self.EMA_1D_SLOW, ) informative_1d["ema_200"] = ta.EMA( informative_1d, timeperiod=self.EMA_1D_LONG, ) dataframe = merge_informative_pair( dataframe, informative_1d, self.timeframe, "1d", ffill=True, ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Macro Daily Bullish Alignment daily_trend = ( (dataframe["close_1d"] > dataframe["ema_50_1d"]) & (dataframe["ema_20_1d"] > dataframe["ema_50_1d"]) ) # Macro 4h Trend & Directional Expansion four_hour_trend = ( (dataframe["close_4h"] > dataframe["ema_20_4h"]) & (dataframe["ema_20_4h"] > dataframe["ema_50_4h"]) & (dataframe["adx_14_4h"] > self.ADX_4H_THRESHOLD) ) # 1h Breakout with volume and RSI momentum sanity breakout = ( (dataframe["close"] > dataframe["previous_breakout_level"]) & ( dataframe["volume"] > self.VOLUME_EXPANSION_MULTIPLIER * dataframe["previous_volume_sma"] ) & (dataframe["rsi_14"] > self.RSI_MIN) & (dataframe["rsi_14"] < self.RSI_MAX) ) enter_long = daily_trend & four_hour_trend & breakout & (dataframe["volume"] > 0) dataframe.loc[enter_long, ["enter_long", "enter_tag"]] = ( 1, "mtf_swing_breakout_adx14", ) return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["exit_long"] = 0 return dataframe @classmethod def _next_stop_price( cls, entry_price: float, entry_atr: float, previous_effective_stop: float | None, highest_completed_close: float | None, current_atr: float, be_locked: bool, trail_active: bool, ) -> float: """Return monotonic absolute stop price with initial ATR, BE lock, and ATR trail.""" initial_stop = entry_price - cls.INITIAL_ATR_MULTIPLIER * entry_atr candidates = [initial_stop] if previous_effective_stop is not None and isfinite(previous_effective_stop): candidates.append(previous_effective_stop) if be_locked: candidates.append(entry_price * (1.0 + cls.BE_LOCK_PROFIT)) if ( trail_active and highest_completed_close is not None and isfinite(highest_completed_close) and isfinite(current_atr) and current_atr > 0 ): candidates.append( highest_completed_close - cls.TRAILING_ATR_MULTIPLIER * current_atr ) return max(candidates) def _completed_candles( self, pair: str, current_time: datetime, ) -> DataFrame: """Return only 1h candles completed by the callback timestamp.""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return dataframe candle_cutoff = pd.Timestamp(current_time) - pd.Timedelta(hours=1) return dataframe.loc[dataframe["date"] <= candle_cutoff] def custom_stoploss( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> float | None: """Maintain the initial ATR stop, BE ratchet, and trailing ATR stop.""" completed = self._completed_candles(pair, current_time) if completed.empty or "atr_14" not in completed.columns: return None current_atr = float(completed.iloc[-1]["atr_14"]) if not isfinite(current_atr) or current_atr <= 0: return None entry_atr = trade.get_custom_data(self._ENTRY_ATR_KEY) if entry_atr is None: entry_atr = current_atr trade.set_custom_data(self._ENTRY_ATR_KEY, float(entry_atr)) entry_atr = float(entry_atr) post_entry = completed.loc[ completed["date"] >= pd.Timestamp(trade.open_date_utc) ] stored_highest = trade.get_custom_data(self._HIGHEST_CLOSE_KEY) highest_completed_close = ( float(stored_highest) if stored_highest is not None else None ) if not post_entry.empty: observed_highest = float(post_entry["close"].max()) if isfinite(observed_highest): highest_completed_close = ( observed_highest if highest_completed_close is None else max(highest_completed_close, observed_highest) ) trade.set_custom_data( self._HIGHEST_CLOSE_KEY, highest_completed_close, ) be_locked = bool(trade.get_custom_data(self._BE_LOCKED_KEY, False)) if ( highest_completed_close is not None and highest_completed_close >= float(trade.open_rate) * (1.0 + self.BE_ACTIVATION_PROFIT) ): be_locked = True trade.set_custom_data(self._BE_LOCKED_KEY, True) trail_active = bool(trade.get_custom_data(self._TRAIL_ACTIVE_KEY, False)) if ( highest_completed_close is not None and highest_completed_close >= float(trade.open_rate) * (1.0 + self.TRAIL_ACTIVATION_PROFIT) ): trail_active = True trade.set_custom_data(self._TRAIL_ACTIVE_KEY, True) stored_stop = trade.get_custom_data(self._EFFECTIVE_STOP_KEY) previous_effective_stop = ( float(stored_stop) if stored_stop is not None else None ) effective_stop = self._next_stop_price( entry_price=float(trade.open_rate), entry_atr=entry_atr, previous_effective_stop=previous_effective_stop, highest_completed_close=highest_completed_close, current_atr=current_atr, be_locked=be_locked, trail_active=trail_active, ) trade.set_custom_data(self._EFFECTIVE_STOP_KEY, float(effective_stop)) distance = stoploss_from_absolute( stop_rate=effective_stop, current_rate=current_rate, is_short=trade.is_short, leverage=trade.leverage or 1.0, ) return distance if distance > 0 else None