RAP-v1 (Regime-Adaptive Pullback), spot long-only.
Timeframe
5m
Direction
Long Only
Stoploss
-10.0%
Trailing Stop
No
ROI
0m: 3.0%, 60m: 1.0%, 120m: 0.0%
Interface Version
3
Startup Candles
N/A
Indicators
3
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# pragma pylint: disable=missing-docstring, invalid-name
from __future__ import annotations
from datetime import datetime, timedelta
from typing import Any
import pandas as pd
import talib.abstract as ta
from pandas import DataFrame
from technical import qtpylib
from freqtrade.persistence import Trade
from freqtrade.strategy import DecimalParameter, IStrategy, merge_informative_pair, stoploss_from_open
class RAPv1Strategy(IStrategy):
"""
RAP-v1 (Regime-Adaptive Pullback), spot long-only.
Core logic:
- 1h regime filter: trade only when market is above rising EMA200.
- 5m entry: pullback to EMA20 inside local trend + RSI recovery.
- Risk: ATR-based custom stop with R-multiple profit protection.
"""
INTERFACE_VERSION = 3
can_short: bool = False
timeframe = "5m"
informative_timeframe = "1h"
process_only_new_candles = True
startup_candle_count: int = 2500
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = False
use_custom_stoploss = True
# Conservative defaults. Runtime config can override these.
minimal_roi = {"120": 0.0, "60": 0.01, "0": 0.03}
stoploss = -0.10
trailing_stop = False
order_types = {
"entry": "limit",
"exit": "limit",
"emergency_exit": "market",
"force_exit": "market",
"force_entry": "market",
"stoploss": "market",
"stoploss_on_exchange": False,
}
order_time_in_force = {"entry": "GTC", "exit": "GTC"}
# Entry control knobs (hyperopt-ready).
pullback_tolerance = DecimalParameter(
0.0, 0.010, decimals=4, default=0.0062, space="buy", optimize=True
)
min_volume_ratio = DecimalParameter(
0.4, 1.0, decimals=2, default=0.62, space="buy", optimize=True
)
max_range_proxy = DecimalParameter(
0.005, 0.025, decimals=3, default=0.018, space="buy", optimize=True
)
max_1h_atr_ratio = DecimalParameter(
0.015, 0.060, decimals=3, default=0.050, space="buy", optimize=True
)
min_rsi_reclaim = DecimalParameter(40, 55, decimals=0, default=49, space="buy", optimize=True)
min_trend_buffer = DecimalParameter(
0.0, 0.010, decimals=4, default=0.0004, space="buy", optimize=True
)
min_ema50_slope = DecimalParameter(
0.0, 0.005, decimals=4, default=0.0001, space="buy", optimize=True
)
max_5m_atr_pct = DecimalParameter(
0.002, 0.020, decimals=3, default=0.010, space="buy", optimize=True
)
min_5m_atr_pct = DecimalParameter(
0.0005, 0.006, decimals=4, default=0.0012, space="buy", optimize=True
)
test_mode_rsi_buffer = DecimalParameter(
2.0, 10.0, decimals=1, default=4.0, space="buy", optimize=True
)
test_mode_ema20_buffer = DecimalParameter(
0.0005, 0.0040, decimals=4, default=0.0010, space="buy", optimize=True
)
risk_atr_mult = DecimalParameter(
1.2, 2.0, decimals=2, default=1.70, space="sell", optimize=True
)
trend_fail_risk_mult = DecimalParameter(
0.12, 0.55, decimals=2, default=0.22, space="sell", optimize=True
)
stagnation_risk_mult = DecimalParameter(
0.08, 0.35, decimals=2, default=0.14, space="sell", optimize=True
)
weak_momentum_rsi = DecimalParameter(
44, 54, decimals=0, default=50, space="sell", optimize=True
)
# Test mode intentionally allows continuation entries in strong trend momentum.
test_mode_relaxed_entry: bool = True
@property
def protections(self) -> list[dict[str, Any]]:
return [
{"method": "CooldownPeriod", "stop_duration_candles": 3},
{
"method": "StoplossGuard",
"lookback_period_candles": 24,
"trade_limit": 3,
"stop_duration_candles": 5,
"only_per_pair": True,
},
{
"method": "MaxDrawdown",
"lookback_period_candles": 48,
"trade_limit": 20,
"stop_duration_candles": 10,
"max_allowed_drawdown": 0.08,
},
{
"method": "LowProfitPairs",
"lookback_period_candles": 72,
"trade_limit": 4,
"stop_duration_candles": 12,
"required_profit": 0.003,
},
]
def informative_pairs(self) -> list[tuple[str, str]]:
if not self.dp:
return []
return [(pair, self.informative_timeframe) for pair in self.dp.current_whitelist()]
def populate_indicators(self, dataframe: DataFrame, metadata: dict[str, Any]) -> DataFrame:
dataframe["ema20"] = ta.EMA(dataframe, timeperiod=20)
dataframe["ema50"] = ta.EMA(dataframe, timeperiod=50)
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
dataframe["atr_pct"] = dataframe["atr"] / dataframe["close"]
dataframe["ema50_slope"] = dataframe["ema50"].pct_change(periods=3)
dataframe["vol_sma20"] = dataframe["volume"].rolling(20).mean()
# Proxy for spread/instability in backtest-safe form.
dataframe["range_proxy"] = (dataframe["high"] - dataframe["low"]) / dataframe["close"]
if self.dp:
inf = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe=self.informative_timeframe)
if inf is not None and not inf.empty:
inf["ema200"] = ta.EMA(inf, timeperiod=200)
inf["atr_1h"] = ta.ATR(inf, timeperiod=14)
inf["atr_ratio"] = inf["atr_1h"] / inf["close"]
inf["regime_up"] = (inf["close"] > inf["ema200"]) & (
inf["ema200"] > inf["ema200"].shift(3)
)
dataframe = merge_informative_pair(
dataframe,
inf[["date", "close", "ema200", "atr_ratio", "regime_up"]],
self.timeframe,
self.informative_timeframe,
ffill=True,
)
if "regime_up_1h" not in dataframe.columns:
dataframe["regime_up_1h"] = False
if "atr_ratio_1h" not in dataframe.columns:
dataframe["atr_ratio_1h"] = 999.0
dataframe["volatility_ok_1h"] = dataframe["atr_ratio_1h"] <= float(self.max_1h_atr_ratio.value)
dataframe["regime_up_1h"] = dataframe["regime_up_1h"].fillna(False)
dataframe["ema50_slope"] = dataframe["ema50_slope"].fillna(0.0)
dataframe["atr_pct"] = dataframe["atr_pct"].fillna(0.0)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict[str, Any]) -> DataFrame:
rsi_reclaim = qtpylib.crossed_above(dataframe["rsi"], float(self.min_rsi_reclaim.value))
ema20_reclaim = qtpylib.crossed_above(dataframe["close"], dataframe["ema20"])
momentum_continuation = (
dataframe["rsi"] >= (float(self.min_rsi_reclaim.value) + float(self.test_mode_rsi_buffer.value))
) & (
dataframe["close"] >= (dataframe["ema20"] * (1.0 + float(self.test_mode_ema20_buffer.value)))
)
trigger_ok = (rsi_reclaim | ema20_reclaim)
if self.test_mode_relaxed_entry:
trigger_ok = trigger_ok | momentum_continuation
entry_conditions = [
dataframe["volume"] > 0,
dataframe["regime_up_1h"],
dataframe["volatility_ok_1h"],
dataframe["close"] >= (dataframe["ema50"] * (1.0 + float(self.min_trend_buffer.value))),
dataframe["ema50_slope"] >= float(self.min_ema50_slope.value),
dataframe["low"] <= (dataframe["ema20"] * (1.0 + float(self.pullback_tolerance.value))),
dataframe["close"] >= dataframe["ema20"],
trigger_ok,
dataframe["volume"] >= (dataframe["vol_sma20"] * float(self.min_volume_ratio.value)),
dataframe["range_proxy"] <= float(self.max_range_proxy.value),
dataframe["atr_pct"] >= float(self.min_5m_atr_pct.value),
dataframe["atr_pct"] <= float(self.max_5m_atr_pct.value),
]
if entry_conditions:
dataframe.loc[
pd.concat(entry_conditions, axis=1).all(axis=1),
["enter_long", "enter_tag"],
] = (1, "rap_pullback_reclaim")
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict[str, Any]) -> DataFrame:
dataframe["exit_long"] = 0
dataframe["exit_tag"] = None
regime_break = ~dataframe["regime_up_1h"]
momentum_fade = (dataframe["rsi"] > 72) & qtpylib.crossed_below(dataframe["close"], dataframe["ema20"])
dataframe.loc[regime_break, ["exit_long", "exit_tag"]] = (1, "regime_break")
dataframe.loc[momentum_fade, ["exit_long", "exit_tag"]] = (1, "momentum_fade")
return dataframe
def _latest_pair_row(self, pair: str) -> pd.Series | None:
if not self.dp:
return None
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe is None or dataframe.empty:
return None
return dataframe.iloc[-1]
def _risk_pct(self, pair: str, trade: Trade | None = None) -> float:
if trade is not None:
stored = trade.get_custom_data("risk_pct")
if stored is not None:
return float(stored)
latest = self._latest_pair_row(pair)
atr_pct = float(latest["atr_pct"]) if latest is not None and "atr_pct" in latest else 0.01
risk_pct = min(float(self.risk_atr_mult.value) * atr_pct, 0.016)
risk_pct = max(risk_pct, 0.005)
if trade is not None:
trade.set_custom_data("risk_pct", risk_pct)
return risk_pct
def custom_stoploss(
self,
pair: str,
trade: Trade,
current_time: datetime,
current_rate: float,
current_profit: float,
after_fill: bool,
**kwargs: Any,
) -> float | None:
risk_pct = self._risk_pct(pair, trade)
if current_profit >= (risk_pct * 1.8):
# Lock profit at roughly +0.8R once +1.8R is reached.
open_relative_stop = max(0.001, current_profit - risk_pct)
return stoploss_from_open(
open_relative_stop,
current_profit,
is_short=trade.is_short,
leverage=trade.leverage,
)
if current_profit >= risk_pct:
# Move to tiny positive stop once trade has moved +1R.
return stoploss_from_open(
0.001,
current_profit,
is_short=trade.is_short,
leverage=trade.leverage,
)
return stoploss_from_open(
-risk_pct,
current_profit,
is_short=trade.is_short,
leverage=trade.leverage,
)
def custom_exit(
self,
pair: str,
trade: Trade,
current_time: datetime,
current_rate: float,
current_profit: float,
**kwargs: Any,
) -> str | bool | None:
open_time = getattr(trade, "open_date_utc", None) or getattr(trade, "open_date", None)
trade_age = (current_time - open_time) if open_time is not None else None
risk_pct = self._risk_pct(pair, trade)
if open_time is not None and (current_time - open_time) >= timedelta(hours=3) and current_profit < 0.003:
return "time_stop_3h"
latest = self._latest_pair_row(pair)
if latest is not None:
regime_ok = bool(latest.get("regime_up_1h", True))
if not regime_ok:
return "regime_break_exit"
rsi_value = float(latest.get("rsi", 50.0))
close_value = float(latest.get("close", 0.0))
ema20_value = float(latest.get("ema20", 0.0))
ema50_value = float(latest.get("ema50", 0.0))
# Cut weak trades before full stop when momentum and trend fail together.
if (
trade_age is not None
and trade_age >= timedelta(minutes=10)
and current_profit < -(risk_pct * float(self.trend_fail_risk_mult.value))
and rsi_value < (float(self.weak_momentum_rsi.value) + 1.0)
and close_value < ema20_value
):
return "trend_fail_early"
if (
trade_age is not None
and trade_age >= timedelta(minutes=25)
and current_profit < -(risk_pct * float(self.stagnation_risk_mult.value))
and rsi_value < (float(self.weak_momentum_rsi.value) + 2.0)
and close_value < ema20_value
):
return "stagnation_cut"
if (
trade_age is not None
and trade_age >= timedelta(minutes=50)
and current_profit < 0.001
and rsi_value < 52.0
and close_value < ema20_value
):
return "time_decay_exit"
if current_profit >= (risk_pct * 1.8):
return "tp_1_8r"
return None