"""BTC/USDT multi-timeframe trend-expansion research strategy, V4. Market: Binance BTC/USDT spot, long only Execution timeframe: completed 5-minute candles Macro context: completed native 1-hour and 4-hour candles V4 deliberately abandons the 1m V1-V3 architecture. It waits for a 5m volatility/volume breakout after a squeeze, permits entries only in a strong macro trend, and lets trends run without a fixed candle-count regime cap. """ 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 BTCMomentumV4_TrendExpansion(IStrategy): INTERFACE_VERSION = 3 timeframe = "5m" informative_timeframes = ("1h", "4h") can_short = False minimal_roi = {} use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # The custom ATR stop is the normal risk control. This fixed value is a # wide emergency floor in case analyzed candle data is 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, } # Freqtrade applies this count to each requested timeframe. Two hundred # fifty native candles cover EMA(200), ADX(14), and every 5m reference. startup_candle_count: int = 250 # 5m squeeze and breakout parameters. BB_PERIOD = 20 BB_STDDEV = 2.0 BANDWIDTH_QUANTILE_PERIOD = 100 BANDWIDTH_QUANTILE = 0.25 BREAKOUT_LOOKBACK = 20 ATR_5M_PERIOD = 14 ATR_REFERENCE_PERIOD = 50 ATR_EXPANSION_MULTIPLIER = 1.25 VOLUME_SMA_PERIOD = 20 VOLUME_EXPANSION_MULTIPLIER = 1.20 # Macro trend parameters. EMA_1H_PERIOD = 200 EMA_4H_PERIOD = 50 ADX_1H_PERIOD = 14 ADX_ENTRY_THRESHOLD = 20.0 ADX_ROLLOVER_THRESHOLD = 40.0 # Risk and trend-exhaustion parameters. INITIAL_ATR_MULTIPLIER = 1.50 TRAIL_ACTIVATION_PROFIT = 0.0100 TRAILING_ATR_MULTIPLIER = 2.00 EMA_5M_EXIT_PERIOD = 50 EMA_EXIT_PROFIT_GATE = 0.0150 _ENTRY_ATR_KEY = "btc_momentum_v4_entry_atr" _HIGHEST_CLOSE_KEY = "btc_momentum_v4_highest_close" _EFFECTIVE_STOP_KEY = "btc_momentum_v4_effective_stop" _TRAIL_ACTIVE_KEY = "btc_momentum_v4_trail_active" def informative_pairs(self): """Load native 1h and 4h 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 5m --------------------------- dataframe["atr_14"] = ta.ATR( dataframe, timeperiod=self.ATR_5M_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"] ) # The breakout candle is excluded from every historical reference. prior_bandwidth = dataframe["bb_bandwidth"].shift(1) dataframe["prior_bandwidth_q25"] = 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_q25"] ).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_50"] = ta.EMA( dataframe, timeperiod=self.EMA_5M_EXIT_PERIOD, ) # -------------------------- native 1h --------------------------- informative_1h = self.dp.get_pair_dataframe( pair=metadata["pair"], timeframe="1h", ).copy() informative_1h["ema_200"] = ta.EMA( informative_1h, timeperiod=self.EMA_1H_PERIOD, ) informative_1h["adx_14"] = ta.ADX( informative_1h, timeperiod=self.ADX_1H_PERIOD, ) informative_1h["adx_14_previous"] = informative_1h["adx_14"].shift(1) dataframe = merge_informative_pair( dataframe, informative_1h, self.timeframe, "1h", ffill=True, ) # -------------------------- native 4h --------------------------- informative_4h = self.dp.get_pair_dataframe( pair=metadata["pair"], timeframe="4h", ).copy() informative_4h["ema_50"] = ta.EMA( informative_4h, timeperiod=self.EMA_4H_PERIOD, ) dataframe = merge_informative_pair( dataframe, informative_4h, self.timeframe, "4h", ffill=True, ) # merge_informative_pair exposes each higher-timeframe row only after # that native candle has closed; no unfinished 1h/4h value is visible. return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: macro_trend = ( ( (dataframe["close_1h"] > dataframe["ema_200_1h"]) | (dataframe["close_4h"] > dataframe["ema_50_4h"]) ) & (dataframe["adx_14_1h"] > self.ADX_ENTRY_THRESHOLD) ) breakout = ( dataframe["prior_squeeze"].fillna(False).astype(bool) & (dataframe["close"] > dataframe["previous_breakout_level"]) & ( dataframe["atr_14"] > self.ATR_EXPANSION_MULTIPLIER * dataframe["previous_atr_reference"] ) & ( dataframe["volume"] > self.VOLUME_EXPANSION_MULTIPLIER * dataframe["previous_volume_sma"] ) ) enter_long = macro_trend & breakout & (dataframe["volume"] > 0) dataframe.loc[enter_long, ["enter_long", "enter_tag"]] = ( 1, "mtf_squeeze_trend_expansion", ) return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Profit-dependent EMA exhaustion and the 1h ADX rollover are handled # in custom_exit, which has access to each trade's achieved profit. 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, trail_active: bool, ) -> float: """Return the monotonic absolute stop price for a long trade.""" 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 ( 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 5m 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(minutes=5) 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 and activated monotonic ATR trail.""" 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: # The first callback sees the signal candle's completed ATR, which # is the latest value known at the next-candle market entry. 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, ) 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, 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 def custom_exit( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> str | None: """Exit on gated EMA exhaustion or a completed-1h ADX rollover.""" completed = self._completed_candles(pair, current_time) if completed.empty: return None current = completed.iloc[-1] adx_now = current.get("adx_14_1h") adx_previous = current.get("adx_14_previous_1h") if ( pd.notna(adx_now) and pd.notna(adx_previous) and float(adx_previous) > self.ADX_ROLLOVER_THRESHOLD and float(adx_now) < float(adx_previous) ): return "adx_1h_rollover_from_40" if len(completed) < 2: return None previous = completed.iloc[-2] ema_cross_below = ( pd.notna(previous.get("ema_50")) and pd.notna(current.get("ema_50")) and float(previous["close"]) >= float(previous["ema_50"]) and float(current["close"]) < float(current["ema_50"]) ) peak_rate = max( float(trade.max_rate or trade.open_rate), float(current_rate), ) achieved_profit = peak_rate / float(trade.open_rate) - 1.0 if achieved_profit >= self.EMA_EXIT_PROFIT_GATE and ema_cross_below: return "ema50_exhaustion_after_1_5pct" return None