# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
import numpy as np
import pandas as pd
from functools import reduce
from pandas import DataFrame
from freqtrade.strategy import (
IStrategy,
Trade,
DecimalParameter,
IntParameter,
)
import talib
import talib.abstract as ta
from technical import qtpylib
class HybridAdaptiveStrategy(IStrategy):
"""
Regime-adaptive hybrid v8 — 1h EMA200 master filter + 4-regime detection.
Architecture:
- 1h EMA200 from resampled 5m data → master trend direction
- 5m indicators → precise entry/exit timing
- Four regimes: UPTREND, DOWNTREND, SIDEWAYS, VOLATILE
- No entries in DOWNTREND — stay flat in crashes
"""
INTERFACE_VERSION = 3
timeframe = "5m"
can_short: bool = False
minimal_roi = {}
stoploss = -0.10
trailing_stop = True
trailing_only_offset_is_reached = True
trailing_stop_positive = 0.025
trailing_stop_positive_offset = 0.04
process_only_new_candles = True
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = False
position_adjustment_enable = True
max_entry_position_adjustment = 0
startup_candle_count: int = 480
# --- Parameters ---
adx_slow = IntParameter(15, 25, default=20, space="buy", optimize=True)
adx_fast = IntParameter(20, 35, default=25, space="buy", optimize=True)
trend_rsi_low = IntParameter(30, 45, default=38, space="buy", optimize=True)
trend_rsi_high = IntParameter(45, 60, default=55, space="buy", optimize=True)
side_rsi_max = IntParameter(25, 40, default=32, space="buy", optimize=True)
side_bb_pct_max = DecimalParameter(0.0, 0.3, default=0.2, space="buy", optimize=True)
exit_rsi_trend = IntParameter(65, 85, default=72, space="sell", optimize=True)
exit_rsi_side = IntParameter(50, 65, default=55, space="sell", optimize=True)
atr_stop_mult = DecimalParameter(1.5, 4.0, default=2.5, space="buy", optimize=True)
vol_factor = DecimalParameter(0.8, 2.0, default=1.0, space="buy", optimize=True)
bb_squeeze_thresh = DecimalParameter(0.1, 0.5, default=0.35, space="buy", optimize=True)
order_types = {
"entry": "limit",
"exit": "limit",
"stoploss": "limit",
"stoploss_on_exchange": False,
}
order_time_in_force = {"entry": "GTC", "exit": "GTC"}
# ------------------------------------------------------------------
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# ---- 1h RESAMPLING for master trend filter ----
# Keep 'date' column intact for freqtrade validator
has_date = "date" in dataframe.columns
date_col = dataframe["date"].copy() if has_date else None
if has_date:
dataframe = dataframe.set_index("date")
if isinstance(dataframe.index, pd.DatetimeIndex):
# Resample close to 1h, compute EMA200
temp_1h = dataframe[["close"]].resample("1h").last()
# Fill NaN to prevent talib from returning all-NaN
close_1h = temp_1h["close"].ffill().bfill().values
ema_values = talib.EMA(close_1h, timeperiod=200)
ema_series = pd.Series(ema_values, index=temp_1h.index)
# EMA50 of EMA200_1h: higher-order trend filter (crash detection)
ema50_values = talib.EMA(ema_values, timeperiod=50)
ema50_series = pd.Series(ema50_values, index=temp_1h.index)
higher_order_bull = ema_series > ema50_s...[truncated]
).ffill()
dataframe["ema200_1h_slope"] = pd.Series(
hour_floor.map(rising), index=dataframe.index
).ffill()
else:
# Fallback if no datetime index
dataframe["ema200_1h"] = np.nan
# Restore 'date' column
if has_date and date_col is not None:
dataframe = dataframe.reset_index()
if "date" not in dataframe.columns:
dataframe["date"] = date_col
elif date_col is not None:
dataframe["date"] = date_col
dataframe["master_bull"] = (
dataframe["close"] > dataframe["ema200_1h"]
)
# ---- 5m Trend ----
dataframe["ema20"] = ta.EMA(dataframe, timeperiod=20)
dataframe["ema50"] = ta.EMA(dataframe, timeperiod=50)
dataframe["ema100"] = ta.EMA(dataframe, timeperiod=100)
dataframe["ema200"] = ta.EMA(dataframe, timeperiod=200)
dataframe["ema_stack"] = 0
dataframe["ema_stack"] += (dataframe["close"] > dataframe["ema20"]).astype(int)
dataframe["ema_stack"] += (dataframe["close"] > dataframe["ema50"]).astype(int)
dataframe["ema_stack"] += (dataframe["close"] > dataframe["ema100"]).astype(int)
dataframe["ema_stack"] += (dataframe["close"] > dataframe["ema200"]).astype(int)
# ---- 5m Momentum ----
dataframe["rsi"] = ta.RSI(dataframe)
dataframe["adx"] = ta.ADX(dataframe)
macd = ta.MACD(dataframe)
dataframe["macd"] = macd["macd"]
dataframe["macdsignal"] = macd["macdsignal"]
dataframe["macdhist"] = macd["macdhist"]
stoch = ta.STOCHF(dataframe)
dataframe["fastk"] = stoch["fastk"]
dataframe["fastd"] = stoch["fastd"]
# ---- 5m Volatility ----
dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
bollinger = qtpylib.bollinger_bands(
qtpylib.typical_price(dataframe), window=20, stds=2
)
dataframe["bb_lowerband"] = bollinger["lower"]
dataframe["bb_middleband"] = bollinger["mid"]
dataframe["bb_upperband"] = bollinger["upper"]
dataframe["bb_percent"] = (
(dataframe["close"] - dataframe["bb_lowerband"])
/ (dataframe["bb_upperband"] - dataframe["bb_lowerband"])
)
dataframe["bb_width"] = (
(dataframe["bb_upperband"] - dataframe["bb_lowerband"])
/ dataframe["bb_middleband"]
)
# BB squeeze
dataframe["bb_width_min_40"] = dataframe["bb_width"].rolling(40).min()
dataframe["bb_squeeze"] = (
dataframe["bb_width"]
< (dataframe["bb_width_min_40"] * 1.1 + self.bb_squeeze_thresh.value)
)
# Volume
dataframe["volume_mean_20"] = dataframe["volume"].rolling(20).mean()
dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_mean_20"]
# BB expansion
dataframe["bb_width_ma"] = dataframe["bb_width"].rolling(20).mean()
dataframe["bb_expanding"] = dataframe["bb_width"] > dataframe["bb_width_ma"] * 1.5
# EMA slopes
dataframe["ema20_slope"] = dataframe["ema20"] > dataframe["ema20"].shift(20)
dataframe["ema50_slope"] = dataframe["ema50"] > dataframe["ema50"].shift(20)
# ---- REGIME DETECTION (1h master filter) ----
dataframe["regime"] = "SIDEWAYS"
# 1. DOWNTREND: price below 1h EMA200
dataframe.loc[~dataframe["master_bull"].fillna(False), "regime"] = "DOWNTREND"
# 2. VOLATILE: BB expanding rapidly (only when above 1h EMA)
dataframe.loc[
dataframe["bb_expanding"] & dataframe["master_bull"],
"regime",
] = "VOLATILE"
# 4. UPTREND: above 1h EMA + EMA200 rising + ADX strong + EMAs sloping up
dataframe.loc[
dataframe["master_bull"]
& dataframe["ema200_1h_slope"]
& (dataframe["adx"] > self.adx_fast.value)
& dataframe["ema20_slope"]
& dataframe["ema50_slope"],
"regime",
] = "UPTREND"
return dataframe
# ------------------------------------------------------------------
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
entry_masks = []
# UPTREND: buy pullbacks (1h confirms uptrend)
uptrend_mask = reduce(
lambda x, y: x & y,
[
dataframe["regime"] == "UPTREND",
dataframe["ema_stack"] >= 3,
dataframe["rsi"] > self.trend_rsi_low.value,
dataframe["rsi"] < self.trend_rsi_high.value,
dataframe["rsi"] > dataframe["rsi"].shift(1),
dataframe["volume_ratio"] > self.vol_factor.value,
],
)
dataframe.loc[uptrend_mask, "enter_tag"] = "uptrend_pullback"
entry_masks.append(uptrend_mask)
# SIDEWAYS: mean reversion at BB lower
side_mask = reduce(
lambda x, y: x & y,
[
dataframe["regime"] == "SIDEWAYS",
dataframe["bb_percent"] < self.side_bb_pct_max.value,
dataframe["rsi"] < self.side_rsi_max.value,
dataframe["macdhist"] > dataframe["macdhist"].shift(1),
dataframe["volume"] > 0,
],
)
dataframe.loc[side_mask, "enter_tag"] = "mean_reversion"
entry_masks.append(side_mask)
# BREAKOUT: BB squeeze release (not in downtrend)
breakout_mask = reduce(
lambda x, y: x & y,
[
dataframe["bb_squeeze"].shift(1),
~dataframe["bb_squeeze"],
dataframe["close"] > dataframe["bb_upperband"],
dataframe["volume_ratio"] > 1.5,
dataframe["regime"] != "DOWNTREND",
],
)
dataframe.loc[breakout_mask, "enter_tag"] = "breakout"
entry_masks.append(breakout_mask)
if entry_masks:
dataframe.loc[
reduce(lambda x, y: x | y, entry_masks), "enter_long"
] = 1
return dataframe
# ------------------------------------------------------------------
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
exit_masks = []
exit_masks.append(
(dataframe["regime"] == "UPTREND")
& (dataframe["rsi"] > self.exit_rsi_trend.value)
)
exit_masks.append(
(dataframe["regime"] == "SIDEWAYS")
& (dataframe["rsi"] > self.exit_rsi_side.value)
)
exit_masks.append(
(dataframe["regime"] == "VOLATILE")
& (dataframe["macdhist"] < 0)
)
exit_masks.append(
(dataframe["ema_stack"] <= 1)
& (dataframe["close"] < dataframe["ema50"])
)
exit_masks.append(~dataframe["master_bull"])
if exit_masks:
dataframe.loc[
reduce(lambda x, y: x | y, exit_masks), "exit_long"
] = 1
return dataframe
# ------------------------------------------------------------------
def adjust_stoploss(
self,
trade: Trade,
current_profit: float,
**kwargs,
) -> tuple[float, float]:
dataframe, _ = self.dp.get_pair_dataframe(
pair=trade.pair, timeframe=self.timeframe
)
if dataframe is None or len(dataframe) < 20:
return current_profit, 0.0
current_atr = dataframe["atr"].iloc[-1]
current_price = dataframe["close"].iloc[-1]
atr_pct = current_atr / current_price
is_bull = dataframe["master_bull"].iloc[-1]
stop_mult = 1.5 if not is_bull else self.atr_stop_mult.value
new_stoploss = trade.open_rate * (1 - atr_pct * stop_mult)
if current_profit > 0.03:
trail_stop = current_price * (1 - atr_pct * 2)
return trail_stop / trade.open_rate, 0.0
return new_stoploss / trade.open_rate, 0.0