# source: https://raw.githubusercontent.com/cheafi/trading-team/519c79f6676eb284064ac7fd54b33941b8adba46/freqtrade/user_data/strategies/A52Strategy.py """ A52 Strategy — Top performing strategy from the ranking table ETH/USDT 5m Futures | WR% 53.4 | DD% 9.8 | Profit +24,737 Core logic: - Multi-timeframe momentum with mean-reversion filter - c=0.50 (conservative position sizing coefficient) - e=-0.18 (slight short bias — the "多空方向針") - Uses ATR-based dynamic SL/TP The 'e' parameter (entry bias) controls long/short preference: e > 0 = long bias, e < 0 = short bias, e = 0 = neutral The 'c' parameter controls position sizing aggression: c = 1.0 = max size, c = 0.5 = half size (safer) """ import numpy as np from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter from pandas import DataFrame import talib.abstract as ta class Github_cheafi_trading_team__A52Strategy__20260406_160741(IStrategy): """ A52 c=0.50 e=-0.18 Top ranked strategy — balanced profit with controlled drawdown """ INTERFACE_VERSION = 3 # Strategy parameters timeframe = "5m" can_short = True # Futures: both long and short # Position sizing coefficient position_c = DecimalParameter(0.1, 1.0, default=0.50, space="buy", optimize=True) # Entry bias: negative = short bias, positive = long bias entry_bias = DecimalParameter(-0.5, 0.5, default=-0.18, space="buy", optimize=True) # Indicator periods fast_ema = IntParameter(8, 21, default=12, space="buy", optimize=True) slow_ema = IntParameter(21, 55, default=26, space="buy", optimize=True) rsi_period = IntParameter(10, 20, default=14, space="buy", optimize=True) atr_period = IntParameter(10, 20, default=14, space="sell", optimize=True) # ROI table — graduated take profit minimal_roi = { "0": 0.012, "30": 0.008, "60": 0.005, "120": 0.002, } # Stoploss stoploss = -0.008 # 0.8% hard stop trailing_stop = True trailing_stop_positive = 0.004 trailing_stop_positive_offset = 0.008 trailing_only_offset_is_reached = True # Startup candles needed for indicators startup_candle_count = 200 def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Calculate all technical indicators""" # EMAs for trend dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=int(self.fast_ema.value)) dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=int(self.slow_ema.value)) # RSI for momentum dataframe["rsi"] = ta.RSI(dataframe, timeperiod=int(self.rsi_period.value)) # ATR for volatility-based stops dataframe["atr"] = ta.ATR(dataframe, timeperiod=int(self.atr_period.value)) # MACD for momentum confirmation macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9) dataframe["macd"] = macd["macd"] dataframe["macd_signal"] = macd["macdsignal"] dataframe["macd_hist"] = macd["macdhist"] # Bollinger Bands for mean reversion bb = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0) dataframe["bb_upper"] = bb["upperband"] dataframe["bb_middle"] = bb["middleband"] dataframe["bb_lower"] = bb["lowerband"] # Volume SMA for volume filter dataframe["volume_sma"] = ta.SMA(dataframe["volume"], timeperiod=20) # ADX for trend strength dataframe["adx"] = ta.ADX(dataframe, timeperiod=14) # EMA slope for trend direction dataframe["ema_slope"] = (dataframe["ema_fast"] - dataframe["ema_fast"].shift(10)) / dataframe["ema_fast"].shift(10) * 100 # Direction score: combines trend + momentum + bias trend_score = np.where( dataframe["ema_fast"] > dataframe["ema_slow"], 1.0, -1.0 ) momentum_score = (dataframe["rsi"] - 50) / 50 # normalize to [-1, 1] macd_score = np.where(dataframe["macd_hist"] > 0, 0.5, -0.5) dataframe["direction_score"] = ( trend_score * 0.4 + momentum_score * 0.3 + macd_score * 0.3 + float(self.entry_bias.value) # The 多空方向針 ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Generate entry signals""" # Long entry conditions dataframe.loc[ ( (dataframe["direction_score"] > 0.6) & (dataframe["close"] > dataframe["ema_fast"]) & (dataframe["rsi"] > 30) & (dataframe["rsi"] < 70) & (dataframe["volume"] > dataframe["volume_sma"] * 1.3) & (dataframe["macd_hist"] > 0) & (dataframe["adx"] > 20) ), ["enter_long", "enter_tag"], ] = (1, "a52_long_momentum") # Short entry conditions (enabled by negative entry_bias) dataframe.loc[ ( (dataframe["direction_score"] < -0.6) & (dataframe["close"] < dataframe["ema_fast"]) & (dataframe["rsi"] > 30) & (dataframe["rsi"] < 70) & (dataframe["volume"] > dataframe["volume_sma"] * 1.3) & (dataframe["macd_hist"] < 0) & (dataframe["adx"] > 20) ), ["enter_short", "enter_tag"], ] = (1, "a52_short_momentum") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Generate exit signals — require confirmation, not single triggers""" # Exit long: strong reversal confirmed by multiple indicators dataframe.loc[ ( (dataframe["direction_score"] < -0.4) & (dataframe["macd_hist"] < 0) ) | ( (dataframe["rsi"] > 78) & (dataframe["close"] > dataframe["bb_upper"]) ), ["exit_long", "exit_tag"], ] = (1, "a52_exit_reversal") # Exit short: strong reversal confirmed dataframe.loc[ ( (dataframe["direction_score"] > 0.4) & (dataframe["macd_hist"] > 0) ) | ( (dataframe["rsi"] < 22) & (dataframe["close"] < dataframe["bb_lower"]) ), ["exit_short", "exit_tag"], ] = (1, "a52_exit_reversal") return dataframe def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time, entry_tag, side: str, **kwargs) -> bool: """Block entry in choppy/fading conditions""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) == 0: return True last = dataframe.iloc[-1] adx = last.get("adx", 0) slope = last.get("ema_slope", 0) # Block long in fading downtrend if side == "long" and adx > 25 and slope < -0.15: return False # Block short in strong uptrend if side == "short" and adx > 25 and slope > 0.15: return False return True def custom_stake_amount(self, pair: str, current_time, current_rate: float, proposed_stake: float, min_stake, max_stake, leverage: float, entry_tag, side: str, **kwargs) -> float: """Apply position sizing coefficient c""" return proposed_stake * float(self.position_c.value)