AutoEvolve Strategy Generation 27 — Multi‑timeframe EMA/RSI confluence with volume spike, recent pivot break and ATR‑based dynamic stop‑loss.
Timeframe
5m
Direction
Long & Short
Stoploss
-2.0%
Trailing Stop
Yes
ROI
0m: 15.0%, 45m: 8.0%, 90m: 4.0%, 180m: 0.0%
Interface Version
3
Startup Candles
300
Indicators
4
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# GENERATION_METADATA = {
# "generation": 27,
# "parent": 26,
# "created_at": "2026-02-27 19:50:15 +02:00",
# "trigger": "profit_drawdown_23.0pct_from_peak",
# "metrics": {"win_rate": 0.7, "profit_factor": 2.763, "sharpe": 88.5517, "max_drawdown": 0.032563},
# "changelog": "Added multi‑timeframe confluence entry filter with pivot break and volume spike"
# }
# WHAT_CHANGED: Introduced 1h and 4h EMA/RSI confluence, volume spike, and recent pivot break as mandatory entry criteria.
# WHY: Current drawdown spikes (max_drawdown 0.032) suggest stop‑losses are triggered; tightening entries should reduce false signals.
# HYPOTHESIS: Win rate ↑ to ≥0.55, profit factor ↑ to ≥1.9, Sharpe ↑ by ~0.2 due to fewer premature exits.
# RISK: Over‑filtering may lower trade count, increasing idle capital and reducing total PnL in strong trends.
import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime
from typing import Optional, Dict, Any
import talib.abstract as ta
from freqtrade.strategy import IStrategy, Trade, stoploss_from_open
class AutoEvolveStrategy(IStrategy):
"""
AutoEvolve Strategy Generation 27 — Multi‑timeframe EMA/RSI confluence with volume spike,
recent pivot break and ATR‑based dynamic stop‑loss.
"""
INTERFACE_VERSION = 3
can_short = True
# Strategy parameters
timeframe = "5m"
startup_candle_count = 300 # enough for higher‑timeframe calculations
minimal_roi = {
"0": 0.15,
"45": 0.08,
"90": 0.04,
"180": 0
}
stoploss = -0.02
trailing_stop = True
trailing_stop_positive = 0.01
trailing_stop_positive_offset = 0.03
trailing_only_offset_is_reached = True
use_exit_signal = True
exit_profit_only = False
# Indicator parameters
ema_fast_period = 20
ema_slow_period = 50
rsi_period = 14
atr_period = 14
volume_ma_period = 20
volume_spike_factor = 1.5
atr_stop_factor = 1.5
# ----------------------------------------------------------------------
# Indicator calculation
# ----------------------------------------------------------------------
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Calculate all technical indicators for base and higher timeframes."""
# Base timeframe indicators
dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=self.ema_fast_period)
dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=self.ema_slow_period)
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=self.rsi_period)
dataframe["atr"] = ta.ATR(dataframe, timeperiod=self.atr_period)
dataframe["volume_ma"] = dataframe["volume"].rolling(self.volume_ma_period).mean()
# Recent pivot high/low over last 3 candles (excluding current)
pivot_high = dataframe["high"].shift(1).rolling(3).max()
pivot_low = dataframe["low"].shift(1).rolling(3).min()
dataframe["pivot_high"] = pivot_high
dataframe["pivot_low"] = pivot_low
# Higher timeframe indicators (1h and 4h)
pair = metadata.get("pair")
if not pair:
return dataframe
# 1h
df_1h = self.dp.get_pair_dataframe(pair=pair, timeframe="1h")
if df_1h is not None and not df_1h.empty:
df_1h["ema_fast_1h"] = ta.EMA(df_1h, timeperiod=self.ema_fast_period)
df_1h["ema_slow_1h"] = ta.EMA(df_1h, timeperiod=self.ema_slow_period)
df_1h["rsi_1h"] = ta.RSI(df_1h, timeperiod=self.rsi_period)
df_1h["volume_ma_1h"] = df_1h["volume"].rolling(self.volume_ma_period).mean()
# Align to base timeframe timestamps (use last known value)
df_1h = df_1h.reindex(dataframe.index, method="ffill")
dataframe["ema_fast_1h"] = df_1h["ema_fast_1h"]
dataframe["ema_slow_1h"] = df_1h["ema_slow_1h"]
dataframe["rsi_1h"] = df_1h["rsi_1h"]
dataframe["volume_ma_1h"] = df_1h["volume_ma_1h"]
# 4h
df_4h = self.dp.get_pair_dataframe(pair=pair, timeframe="4h")
if df_4h is not None and not df_4h.empty:
df_4h["ema_fast_4h"] = ta.EMA(df_4h, timeperiod=self.ema_fast_period)
df_4h["ema_slow_4h"] = ta.EMA(df_4h, timeperiod=self.ema_slow_period)
df_4h["rsi_4h"] = ta.RSI(df_4h, timeperiod=self.rsi_period)
df_4h["volume_ma_4h"] = df_4h["volume"].rolling(self.volume_ma_period).mean()
df_4h = df_4h.reindex(dataframe.index, method="ffill")
dataframe["ema_fast_4h"] = df_4h["ema_fast_4h"]
dataframe["ema_slow_4h"] = df_4h["ema_slow_4h"]
dataframe["rsi_4h"] = df_4h["rsi_4h"]
dataframe["volume_ma_4h"] = df_4h["volume_ma_4h"]
return dataframe
# ----------------------------------------------------------------------
# Entry signal generation
# ----------------------------------------------------------------------
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Set entry signals only when all confluence conditions are satisfied."""
dataframe["enter_long"] = 0
dataframe["enter_short"] = 0
# Ensure higher timeframe columns exist; otherwise skip
required_cols = [
"ema_fast_1h", "ema_slow_1h",
"ema_fast_4h", "ema_slow_4h",
"rsi_1h", "rsi_4h",
"volume_ma"
]
if not all(col in dataframe.columns for col in required_cols):
return dataframe
# EMA cross on 1h (current candle)
ema_cross_up_1h = (dataframe["ema_fast_1h"] > dataframe["ema_slow_1h"]) & (
dataframe["ema_fast_1h"].shift(1) <= dataframe["ema_slow_1h"].shift(1)
)
ema_cross_down_1h = (dataframe["ema_fast_1h"] < dataframe["ema_slow_1h"]) & (
dataframe["ema_fast_1h"].shift(1) >= dataframe["ema_slow_1h"].shift(1)
)
# Directional bias on 4h
bias_long_4h = dataframe["ema_fast_4h"] > dataframe["ema_slow_4h"]
bias_short_4h = dataframe["ema_fast_4h"] < dataframe["ema_slow_4h"]
# RSI extremes on all timeframes
rsi_long_cond = (
(dataframe["rsi"] < 30) &
(dataframe["rsi_1h"] < 30) &
(dataframe["rsi_4h"] < 30)
)
rsi_short_cond = (
(dataframe["rsi"] > 70) &
(dataframe["rsi_1h"] > 70) &
(dataframe["rsi_4h"] > 70)
)
# Volume spike on base timeframe
vol_spike = dataframe["volume"] > (self.volume_spike_factor * dataframe["volume_ma"])
# Pivot break
break_up = dataframe["close"] > dataframe["pivot_high"]
break_down = dataframe["close"] < dataframe["pivot_low"]
# Long entry condition: all True
long_condition = (
ema_cross_up_1h &
bias_long_4h &
rsi_long_cond &
vol_spike &
break_up
)
dataframe.loc[long_condition, "enter_long"] = 1
# Short entry condition: all True
short_condition = (
ema_cross_down_1h &
bias_short_4h &
rsi_short_cond &
vol_spike &
break_down
)
dataframe.loc[short_condition, "enter_short"] = 1
return dataframe
# ----------------------------------------------------------------------
# Exit signal generation (simple EMA crossover)
# ----------------------------------------------------------------------
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Generate exit signals based on EMA crossovers on the base timeframe."""
dataframe["exit_long"] = 0
dataframe["exit_short"] = 0
ema_cross_down = (dataframe["ema_fast"] < dataframe["ema_slow"]) & (
dataframe["ema_fast"].shift(1) >= dataframe["ema_slow"].shift(1)
)
ema_cross_up = (dataframe["ema_fast"] > dataframe["ema_slow"]) & (
dataframe["ema_fast"].shift(1) <= dataframe["ema_slow"].shift(1)
)
dataframe.loc[ema_cross_down, "exit_long"] = 1
dataframe.loc[ema_cross_up, "exit_short"] = 1
return dataframe
# ----------------------------------------------------------------------
# Custom stop‑loss (ATR based)
# ----------------------------------------------------------------------
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
current_rate: float, entry_rate: float, **kwargs) -> float:
"""Dynamic stop‑loss: entry_price - ATR * factor for longs, +ATR*factor for shorts."""
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if df is None or df.empty:
return self.stoploss
latest = df.iloc[-1]
atr = latest.get("atr")
if pd.isna(atr):
return self.stoploss
if trade.is_short:
stop_price = entry_rate + atr * self.atr_stop_factor
return (stop_price - current_rate) / current_rate
else:
stop_price = entry_rate - atr * self.atr_stop_factor
return (stop_price - current_rate) / current_rate
# ----------------------------------------------------------------------
# Leverage definition
# ----------------------------------------------------------------------
def leverage(self, pair: str, is_short: bool, **kwargs) -> float:
"""Fixed leverage of 5× for all trades."""
return 5.0
# ----------------------------------------------------------------------
# Optional custom exit logic (placeholder)
# ----------------------------------------------------------------------
def custom_exit(self,
pair: str,
trade: Trade,
current_time: datetime,
**kwargs) -> Optional[Dict[str, Any]]:
"""No custom exit – rely on exit signals and ROI."""
return None
# ----------------------------------------------------------------------
# Confirm entry – additional safety checks
# ----------------------------------------------------------------------
def confirm_trade_entry(self,
pair: str,
order_type: str,
amount: float,
rate: float,
time_in_force: str,
current_time: datetime,
entry_tag: Optional[str],
side: str,
**kwargs) -> bool:
"""Validate candle range, ATR size and candle direction before allowing entry."""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe is None or dataframe.empty:
return False
last = dataframe.iloc[-1]
# Candle range check (max 4% of price)
candle_range = last["high"] - last["low"]
if (candle_range / rate) > 0.04:
return False
# ATR should not be excessive (>8% of price)
if pd.isna(last.get("atr")) or (last["atr"] / rate) > 0.08:
return False
# Direction consistency with candle body
bullish = last["close"] > last["open"]
bearish = last["close"] < last["open"]
if side == "long" and not bullish:
return False
if side == "short" and not bearish:
return False
return True
# ----------------------------------------------------------------------
# Custom fee – default to exchange handling
# ----------------------------------------------------------------------
def custom_fee(self,
pair: str,
trade: Trade,
order_type: str,
amount: float,
rate: float,
is_short: bool,
**kwargs) -> float:
"""Use exchange default fee handling."""
return 0.0