A51 c=0.35 e=0.00 High win-rate scalper with tight risk management
Timeframe
5m
Direction
Long & Short
Stoploss
-0.5%
Trailing Stop
Yes
ROI
0m: 0.8%, 15m: 0.5%, 30m: 0.3%, 60m: 0.1%
Interface Version
3
Startup Candles
100
Indicators
8
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
"""
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 A51Strategy(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)