# source: https://raw.githubusercontent.com/cheafi/trading-team/5a8336d6f90a166e95421141499b8b0e3e696e9e/freqtrade/user_data/strategies/A51Strategy.py """ A51 Strategy — Scalping-oriented with tight risk management ETH/USDT 5m Futures | WR% 61.7 | DD% 7.2 | Profit +12,841 Core logic: - VWAP + Order Block detection for precise entries - c=0.35 (conservative — many small wins) - e=0.00 (neutral bias — pure price action) - Tight stops with high win rate """ 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__A51Strategy__20260618_200431(IStrategy): """ A51 c=0.35 e=0.00 High win-rate scalper with tight risk management """ INTERFACE_VERSION = 3 timeframe = "5m" can_short = True position_c = DecimalParameter( 0.1, 1.0, default=0.35, space="buy", optimize=True ) entry_bias = DecimalParameter( -0.5, 0.5, default=0.00, space="buy", optimize=True ) minimal_roi = { "0": 0.008, "15": 0.005, "30": 0.003, "60": 0.001, } stoploss = -0.005 trailing_stop = True trailing_stop_positive = 0.003 trailing_stop_positive_offset = 0.005 trailing_only_offset_is_reached = True startup_candle_count = 100 def _vwap(self, df: DataFrame) -> DataFrame: """Calculate VWAP (resets each 288 candles ≈ 1 day)""" period = 288 tp = (df["high"] + df["low"] + df["close"]) / 3 cumvol = df["volume"].rolling(window=period, min_periods=1).sum() cumtp = (tp * df["volume"]).rolling( window=period, min_periods=1 ).sum() df["vwap"] = cumtp / cumvol return df def populate_indicators( self, dataframe: DataFrame, metadata: dict ) -> DataFrame: # VWAP dataframe = self._vwap(dataframe) # Short-term EMAs for scalping dataframe["ema_5"] = ta.EMA(dataframe, timeperiod=5) dataframe["ema_8"] = ta.EMA(dataframe, timeperiod=8) dataframe["ema_13"] = ta.EMA(dataframe, timeperiod=13) # RSI with shorter period for scalping dataframe["rsi"] = ta.RSI(dataframe, timeperiod=9) # Stochastic RSI stoch = ta.STOCHRSI( dataframe, timeperiod=14, fastk_period=3, fastd_period=3 ) dataframe["stoch_k"] = stoch["fastk"] dataframe["stoch_d"] = stoch["fastd"] # ATR for stop calculation dataframe["atr"] = ta.ATR(dataframe, timeperiod=10) # Volume analysis dataframe["volume_sma"] = ta.SMA( dataframe["volume"], timeperiod=20 ) dataframe["volume_ratio"] = ( dataframe["volume"] / dataframe["volume_sma"].replace(0, np.nan) ).fillna(0) # ADX for trend strength dataframe["adx"] = ta.ADX(dataframe, timeperiod=14) # EMA slope for trend direction dataframe["ema_slope"] = (dataframe["ema_5"] - dataframe["ema_5"].shift(10)) / dataframe["ema_5"].shift(10) * 100 # Order block detection (simplified) dataframe["bullish_ob"] = ( (dataframe["close"].shift(2) < dataframe["open"].shift(2)) & (dataframe["close"].shift(1) > dataframe["open"].shift(1)) & ( dataframe["close"].shift(1) > dataframe["high"].shift(2) ) ).astype(int) dataframe["bearish_ob"] = ( (dataframe["close"].shift(2) > dataframe["open"].shift(2)) & (dataframe["close"].shift(1) < dataframe["open"].shift(1)) & ( dataframe["close"].shift(1) < dataframe["low"].shift(2) ) ).astype(int) return dataframe def populate_entry_trend( self, dataframe: DataFrame, metadata: dict ) -> DataFrame: # Long: price above VWAP + bullish OB or stoch cross dataframe.loc[ ( (dataframe["close"] > dataframe["vwap"]) & (dataframe["ema_5"] > dataframe["ema_8"]) & ( (dataframe["bullish_ob"] == 1) | ( (dataframe["stoch_k"] > dataframe["stoch_d"]) & (dataframe["stoch_k"].shift(1) <= dataframe["stoch_d"].shift(1)) & (dataframe["stoch_k"] < 30) ) ) & (dataframe["volume_ratio"] > 1.5) & (dataframe["rsi"] > 40) & (dataframe["rsi"] < 60) & (dataframe["adx"] > 20) ), ["enter_long", "enter_tag"], ] = (1, "a51_vwap_ob_long") # Short: price below VWAP + bearish OB or stoch cross dataframe.loc[ ( (dataframe["close"] < dataframe["vwap"]) & (dataframe["ema_5"] < dataframe["ema_8"]) & ( (dataframe["bearish_ob"] == 1) | ( (dataframe["stoch_k"] < dataframe["stoch_d"]) & (dataframe["stoch_k"].shift(1) >= dataframe["stoch_d"].shift(1)) & (dataframe["stoch_k"] > 70) ) ) & (dataframe["volume_ratio"] > 1.5) & (dataframe["rsi"] > 40) & (dataframe["rsi"] < 60) & (dataframe["adx"] > 20) ), ["enter_short", "enter_tag"], ] = (1, "a51_vwap_ob_short") return dataframe def populate_exit_trend( self, dataframe: DataFrame, metadata: dict ) -> DataFrame: # Exit long: EMA cascade down OR extreme RSI dataframe.loc[ ( (dataframe["ema_5"] < dataframe["ema_8"]) & (dataframe["ema_8"] < dataframe["ema_13"]) & (dataframe["close"] < dataframe["vwap"]) ) | (dataframe["rsi"] > 75), ["exit_long", "exit_tag"], ] = (1, "a51_exit_cascade") # Exit short: EMA cascade up OR extreme RSI dataframe.loc[ ( (dataframe["ema_5"] > dataframe["ema_8"]) & (dataframe["ema_8"] > dataframe["ema_13"]) & (dataframe["close"] > dataframe["vwap"]) ) | (dataframe["rsi"] < 25), ["exit_short", "exit_tag"], ] = (1, "a51_exit_cascade") 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) if side == "long" and adx > 25 and slope < -0.15: return False if side == "short" and adx > 25 and slope > 0.15: return False return True def custom_stake_amount( self, pair, current_time, current_rate, proposed_stake, min_stake, max_stake, leverage, entry_tag, side, **kwargs ): return proposed_stake * float(self.position_c.value)