"""BTC/USDT multi-timeframe swing momentum strategy, V5.2 Wide Trail. Single isolated modification relative to BTCMomentumV5_Base: - Increased TRAILING_ATR_MULTIPLIER from 2.50 to 3.50. Hypothesis: Widening the trailing stop buffer from 2.5x to 3.5x ATR gives swing positions more breathing room during multi-day trend expansions, capturing larger winning swings and boosting net profit and average trade return. """ 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_2_WideTrail(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 EMA_4H_FAST = 20 EMA_4H_SLOW = 50 ADX_4H_PERIOD = 14 ADX_4H_THRESHOLD = 18.0 EMA_1D_FAST = 20 EMA_1D_SLOW = 50 EMA_1D_LONG = 200 # Risk and trailing ratchet parameters (TRAILING_ATR_MULTIPLIER changed from 2.50 to 3.50) INITIAL_ATR_MULTIPLIER = 3.00 BE_ACTIVATION_PROFIT = 0.0100 BE_LOCK_PROFIT = 0.0020 TRAIL_ACTIVATION_PROFIT = 0.0150 TRAILING_ATR_MULTIPLIER = 3.50 _ENTRY_ATR_KEY = "btc_momentum_v5_2_entry_atr" _HIGHEST_CLOSE_KEY = "btc_momentum_v5_2_highest_close" _EFFECTIVE_STOP_KEY = "btc_momentum_v5_2_effective_stop" _BE_LOCKED_KEY = "btc_momentum_v5_2_be_locked" _TRAIL_ACTIVE_KEY = "btc_momentum_v5_2_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", ) 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