from functools import reduce import talib.abstract as ta from pandas import DataFrame from freqtrade.strategy import IStrategy, merge_informative_pair from freqtrade.persistence import Trade from datetime import datetime class TrendStructureExecutor(IStrategy): """ TrendStructureExecutor — Trend-continuation strategy (spot/futures). Core concept: Identify established trends on the 1h chart (EMA52 + MACD + EMA200), then trade 5m continuation entries when the MACD histogram pulls back to zero and resumes in the trend direction. Skip low-volatility ranging markets. Partial take-profit on momentum weakening. """ INTERFACE_VERSION = 3 # ========================================================================= # CONFIGURATION # ========================================================================= timeframe = "5m" informative_timeframe = "1h" # Futures support (long + short) # Set can_short = True and switch config to futures mode (BTC/USDT:USDT) # to enable short trading. can_short = False # trading_mode = "futures" # margin_mode = "isolated" # Risk management — fixed 0.8% stoploss (tighter than the 1% ROI target) stoploss = -0.008 # Trailing stop to protect profits trailing_stop = True trailing_stop_positive = 0.004 trailing_stop_positive_offset = 0.012 trailing_only_offset_is_reached = True # Position adjustment for partial take-profits position_adjustment_enable = True # ROI disabled — exits managed by trailing stop + partial TP + EMA52 breach minimal_roi = {"0": 0.99} # General settings use_exit_signal = True exit_profit_only = False startup_candle_count = 200 process_only_new_candles = True order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": False, } # ========================================================================= # INFORMATIVE PAIRS # ========================================================================= def informative_pairs(self): pairs = self.dp.current_whitelist() return [(pair, self.informative_timeframe) for pair in pairs] # ========================================================================= # INDICATORS # ========================================================================= def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ 1h: EMA52, EMA200, MACD, slope, range/consolidation, trend flags. 5m: MACD, histogram direction helpers. """ if self.dp: informative = self.dp.get_pair_dataframe( pair=metadata["pair"], timeframe=self.informative_timeframe ) # --- EMA 52 --- informative["ema_52"] = ta.EMA(informative, timeperiod=52) # --- EMA 200 (super-trend filter) --- informative["ema_200"] = ta.EMA(informative, timeperiod=200) # EMA 52 slope (3-period ROC for noise reduction) informative["ema_52_slope"] = ( informative["ema_52"] - informative["ema_52"].shift(3) ) # --- MACD (12, 26, 9) --- macd_1h = ta.MACD(informative) informative["macd_hist_1h"] = macd_1h["macdhist"] informative["macd_hist_1h_delta"] = ( informative["macd_hist_1h"] - informative["macd_hist_1h"].shift(1) ) # --- Range / consolidation filter --- # If the 20-candle price range is less than 1.5 %, the market is # considered to be ranging and no entries are allowed. informative["range_high_20"] = informative["high"].rolling(20).max() informative["range_low_20"] = informative["low"].rolling(20).min() informative["range_pct"] = ( (informative["range_high_20"] - informative["range_low_20"]) / informative["range_low_20"] ) informative["is_ranging"] = (informative["range_pct"] < 0.015).astype(int) # --- LONG trend confirmation --- # Price above EMA52 + EMA52 sloping up + MACD histogram positive # + histogram not shrinking significantly (delta > -0.5 * rolling std) informative["trend_bull"] = ( (informative["close"] > informative["ema_52"]) & (informative["close"] > informative["ema_200"]) & (informative["ema_52_slope"] > 0) & (informative["macd_hist_1h"] > 0) & ( informative["macd_hist_1h_delta"] > -informative["macd_hist_1h"].rolling(20).std() * 0.5 ) ).astype(int) # --- SHORT trend confirmation --- # Price below EMA52 + EMA52 sloping down + MACD histogram negative # + histogram not expanding upward (delta < +0.5 * rolling std) informative["trend_bear"] = ( (informative["close"] < informative["ema_52"]) & (informative["close"] < informative["ema_200"]) & (informative["ema_52_slope"] < 0) & (informative["macd_hist_1h"] < 0) & ( informative["macd_hist_1h_delta"] < informative["macd_hist_1h"].rolling(20).std() * 0.5 ) ).astype(int) # Merge 1h → 5m (merge_informative_pair handles lookahead protection # by shifting the higher-timeframe data by one candle) dataframe = merge_informative_pair( dataframe, informative, self.timeframe, self.informative_timeframe, ffill=True, ) # --- 5m MACD --- macd_5m = ta.MACD(dataframe) dataframe["macd_hist_5m"] = macd_5m["macdhist"] # Direction helpers (avoids repeating shift logic in entry/exit methods) dataframe["macd_hist_5m_up"] = ( dataframe["macd_hist_5m"] > dataframe["macd_hist_5m"].shift(1) ) dataframe["macd_hist_5m_down"] = ( dataframe["macd_hist_5m"] < dataframe["macd_hist_5m"].shift(1) ) return dataframe # ========================================================================= # ENTRY LOGIC # ========================================================================= def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ LONG: 1h bullish + 5m MACD hist pullback-then-resumption + recent reset. SHORT: 1h bearish + 5m MACD hist pullback-then-resumption + recent reset. Both skip ranging markets. """ # Columns from merge_informative_pair carry the _1h suffix trend_bull = dataframe["trend_bull_1h"] trend_bear = dataframe["trend_bear_1h"] is_ranging = dataframe["is_ranging_1h"] # ── LONG ────────────────────────────────────────────────────────────── long_conditions = [ trend_bull == 1, is_ranging == 0, dataframe["macd_hist_5m_down"].shift(1) == True, dataframe["macd_hist_5m_up"] == True, dataframe["macd_hist_5m"] > 0, dataframe["macd_hist_5m"].rolling(3).min() < 0, ] dataframe.loc[ reduce(lambda a, b: a & b, long_conditions), ["enter_long", "enter_tag"], ] = (1, "long_continuation") # ── SHORT ───────────────────────────────────────────────────────────── short_conditions = [ trend_bear == 1, is_ranging == 0, dataframe["macd_hist_5m_up"].shift(1) == True, dataframe["macd_hist_5m_down"] == True, dataframe["macd_hist_5m"] < 0, dataframe["macd_hist_5m"].rolling(3).max() > 0, ] dataframe.loc[ reduce(lambda a, b: a & b, short_conditions), ["enter_short", "enter_tag"], ] = (1, "short_continuation") return dataframe # ========================================================================= # EXIT LOGIC # ========================================================================= def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ LONG: exit on 1h EMA52 breach (trend reversal). SHORT (futures only): exit on 1h EMA52 breach. """ long_cond = dataframe["close"] < dataframe["ema_52_1h"] dataframe.loc[long_cond, "exit_long"] = 1 dataframe.loc[long_cond, "exit_tag"] = "long_exit" if self.can_short: short_cond = dataframe["close"] > dataframe["ema_52_1h"] dataframe.loc[short_cond, "exit_short"] = 1 dataframe.loc[short_cond, "exit_tag"] = "short_exit" return dataframe # ========================================================================= # POSITION ADJUSTMENT (Partial Take-Profit) # ========================================================================= def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: float | None, max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs) -> float | None: """ Sell 50% when MACD momentum weakens while in profit. Fires once per trade (guarded by filled_exits). Exits half the position when the 5m MACD histogram starts declining toward zero while we are still above +0.5% profit. """ if current_profit <= 0.005: return None # Only one partial exit per trade filled_exits = trade.select_filled_orders(trade.exit_side) if filled_exits: return None dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) if dataframe is None or len(dataframe) < 2: return None last = dataframe.iloc[-1] prev = dataframe.iloc[-2] if trade.is_short: if last["macd_hist_5m"] < 0 and last["macd_hist_5m"] > prev["macd_hist_5m"]: return -(trade.stake_amount / 2) else: if last["macd_hist_5m"] > 0 and last["macd_hist_5m"] < prev["macd_hist_5m"]: return -(trade.stake_amount / 2) return None