# source: https://raw.githubusercontent.com/wangc-joker/ft_userdata/39b98171b9a29ada7a0af26814682d68ee48bd81/user_data/strategies/myStrage/Top9RegimeStructureStrategy.py from __future__ import annotations from datetime import timedelta from typing import Optional import numpy as np import pandas as pd import talib.abstract as ta from pandas import DataFrame from freqtrade.persistence import Trade from freqtrade.strategy import IStrategy, informative, stoploss_from_absolute class Github_wangc_joker_ft_userdata__Top9RegimeStructureStrategy__20260515_060339(IStrategy): """ Clean V1 redesign of the old Top9 regime/reversal family. V1 intentionally excludes reversal entries. The goal is to validate a shorter, more transparent 1h/1d regime-structure strategy before adding reversal and live-trading capital caps in later versions. """ INTERFACE_VERSION = 3 can_short = True timeframe = "1h" process_only_new_candles = True startup_candle_count = 240 use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = True use_custom_stoploss = True minimal_roi = {"0": 0.10} stoploss = -0.03 allowed_pairs = { "BTC/USDT:USDT", "ETH/USDT:USDT", "BNB/USDT:USDT", "SOL/USDT:USDT", "TRX/USDT:USDT", "ADA/USDT:USDT", "ZEC/USDT:USDT", "XRP/USDT:USDT", "DOGE/USDT:USDT", } pair_multipliers = { "BTC/USDT:USDT": 1.00, "ETH/USDT:USDT": 1.00, "SOL/USDT:USDT": 1.00, "BNB/USDT:USDT": 0.90, "XRP/USDT:USDT": 0.80, "ADA/USDT:USDT": 0.75, "DOGE/USDT:USDT": 0.60, "TRX/USDT:USDT": 0.70, "ZEC/USDT:USDT": 0.80, } signal_multipliers = { "long_trend_breakout": 1.00, "long_pullback_restart": 0.90, "short_trend_breakdown": 0.80, "short_pullback_fail": 0.60, } enable_long_trend_breakout = True enable_long_pullback_restart = True enable_short_trend_breakdown = True enable_short_pullback_fail = True allow_range_short_trend = True require_4h_short_trend = False require_4h_short_breakdown = False require_major_4h_short_breakdown = False major_pairs = {"BTC/USDT:USDT", "ETH/USDT:USDT"} use_signal_quality_stake_scale = False require_4h_long_trend = False require_strong_bull_long = False use_v2_long_breakout = False @property def protections(self): return [ {"method": "CooldownPeriod", "stop_duration_candles": 3}, { "method": "StoplossGuard", "lookback_period_candles": 60, "trade_limit": 2, "stop_duration_candles": 14, "only_per_pair": False, }, { "method": "MaxDrawdown", "lookback_period_candles": 96, "trade_limit": 10, "stop_duration_candles": 24, "max_allowed_drawdown": 0.10, }, ] @staticmethod def _safe_div(a, b): return a / b.replace(0, np.nan) @staticmethod def _crossed_above(series: pd.Series, level: pd.Series) -> pd.Series: return (series > level) & (series.shift(1) <= level.shift(1)) @staticmethod def _crossed_below(series: pd.Series, level: pd.Series) -> pd.Series: return (series < level) & (series.shift(1) >= level.shift(1)) @staticmethod def _trend_indicators(dataframe: DataFrame) -> DataFrame: dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=20) dataframe["ema_slow"] = 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["volume_mean"] = dataframe["volume"].shift(1).rolling(20).mean() dataframe["recent_high"] = dataframe["high"].shift(1).rolling(24).max() dataframe["recent_low"] = dataframe["low"].shift(1).rolling(24).min() dataframe["range_high"] = dataframe["high"].shift(1).rolling(12).max() dataframe["range_low"] = dataframe["low"].shift(1).rolling(12).min() dataframe["range_width"] = (dataframe["range_high"] - dataframe["range_low"]) / dataframe["close"] dataframe["range_tight"] = dataframe["range_width"] < 0.045 dataframe["body_pct"] = (dataframe["close"] - dataframe["open"]).abs() / dataframe["close"] dataframe["ema_slow_slope"] = dataframe["ema_slow"] - dataframe["ema_slow"].shift(6) dataframe["ema_slow_slope_up"] = dataframe["ema_slow_slope"] > 0 dataframe["ema_slow_slope_down"] = dataframe["ema_slow_slope"] < 0 dataframe["center"] = (dataframe["high"] + dataframe["low"] + dataframe["close"]) / 3.0 dataframe["center_ma"] = dataframe["center"].rolling(5).mean() dataframe["center_up"] = dataframe["center_ma"] > dataframe["center_ma"].shift(3) dataframe["center_down"] = dataframe["center_ma"] < dataframe["center_ma"].shift(3) dataframe["structure_stop_long"] = dataframe["low"].shift(1).rolling(8).min() dataframe["structure_stop_short"] = dataframe["high"].shift(1).rolling(8).max() return dataframe @informative("1d") def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe = self._trend_indicators(dataframe) bull_strong = ( (dataframe["close"] > dataframe["ema_fast"]) & (dataframe["ema_fast"] > dataframe["ema_slow"]) & (dataframe["rsi"] > 58) & dataframe["ema_slow_slope_up"] ) bull_weak = ( ~bull_strong & (dataframe["close"] > dataframe["ema_slow"]) & (dataframe["rsi"] > 48) ) bear = ( (dataframe["close"] < dataframe["ema_fast"]) & (dataframe["ema_fast"] < dataframe["ema_slow"]) & (dataframe["rsi"] < 45) & dataframe["ema_slow_slope_down"] ) dataframe["regime_id"] = 0 dataframe.loc[bull_strong, "regime_id"] = 3 dataframe.loc[bull_weak, "regime_id"] = 2 dataframe.loc[bear, "regime_id"] = -1 dataframe["regime_long_multiplier"] = np.select( [dataframe["regime_id"].eq(3), dataframe["regime_id"].eq(2), dataframe["regime_id"].eq(-1)], [1.10, 0.90, 0.40], default=0.60, ) dataframe["regime_short_multiplier"] = np.select( [dataframe["regime_id"].eq(3), dataframe["regime_id"].eq(2), dataframe["regime_id"].eq(-1)], [0.20, 0.50, 1.00], default=0.60, ) return dataframe @informative("4h") def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe = self._trend_indicators(dataframe) dataframe["trend_short_ok"] = ( (dataframe["close"] < dataframe["ema_fast"]) & (dataframe["ema_fast"] < dataframe["ema_slow"]) & dataframe["ema_slow_slope_down"] & (dataframe["rsi"] < 48) ) dataframe["breakdown_short_ok"] = ( dataframe["trend_short_ok"] & (dataframe["close"] < dataframe["recent_low"] * 0.997) & (dataframe["body_pct"] > 0.006) ) dataframe["trend_long_ok"] = ( (dataframe["close"] > dataframe["ema_fast"]) & (dataframe["ema_fast"] > dataframe["ema_slow"]) & dataframe["ema_slow_slope_up"] & (dataframe["rsi"] > 52) ) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe = self._trend_indicators(dataframe) dataframe["vol_ok"] = dataframe["volume"] > dataframe["volume_mean"] * 1.20 dataframe["breakout_vol_ok"] = dataframe["volume"] > dataframe["volume_mean"] * 1.70 dataframe["volatility_ok"] = dataframe["atr_pct"] < 0.060 dataframe["pullback_seen_long"] = ( (dataframe["low"].shift(1).rolling(8).min() < dataframe["ema_fast"].shift(1) * 1.006) | (dataframe["low"].shift(1).rolling(8).min() < dataframe["ema_slow"].shift(1) * 1.010) ) dataframe["pullback_seen_short"] = ( (dataframe["high"].shift(1).rolling(8).max() > dataframe["ema_fast"].shift(1) * 0.994) | (dataframe["high"].shift(1).rolling(8).max() > dataframe["ema_slow"].shift(1) * 0.990) ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: if metadata["pair"] not in self.allowed_pairs: return dataframe regime = dataframe.get("regime_id_1d", pd.Series(0, index=dataframe.index)).fillna(0) bull_strong = regime.eq(3) bull_weak = regime.eq(2) bear = regime.eq(-1) range_regime = regime.eq(0) base_filter = dataframe["vol_ok"] & dataframe["volatility_ok"] & (dataframe["volume"] > 0) trend_4h_short_ok = dataframe.get("trend_short_ok_4h", pd.Series(True, index=dataframe.index)).fillna(False) breakdown_4h_short_ok = dataframe.get("breakdown_short_ok_4h", pd.Series(True, index=dataframe.index)).fillna(False) trend_4h_long_ok = dataframe.get("trend_long_ok_4h", pd.Series(True, index=dataframe.index)).fillna(False) long_4h_trend_filter = trend_4h_long_ok if self.require_4h_long_trend else True short_4h_trend_filter = trend_4h_short_ok if self.require_4h_short_trend else True pair_needs_major_breakdown = ( self.require_major_4h_short_breakdown and metadata["pair"] in self.major_pairs ) short_4h_breakdown_filter = ( breakdown_4h_short_ok if self.require_4h_short_breakdown or pair_needs_major_breakdown else True ) if self.use_v2_long_breakout: long_regime = bull_strong if self.require_strong_bull_long else (bull_strong | bull_weak) long_trend_breakout = ( long_regime & base_filter & long_4h_trend_filter & dataframe["breakout_vol_ok"] & dataframe["range_tight"] & dataframe["ema_slow_slope_up"] & (dataframe["rsi"] > 55) & (dataframe["rsi"] < 72) & (dataframe["close"] > dataframe["recent_high"] * 1.012) & (dataframe["body_pct"] > 0.010) ) else: long_trend_breakout = ( (bull_strong | bull_weak) & base_filter & dataframe["breakout_vol_ok"] & dataframe["range_tight"] & dataframe["ema_slow_slope_up"] & (dataframe["rsi"] > 54) & (dataframe["close"] > dataframe["recent_high"] * 1.010) & (dataframe["body_pct"] > 0.010) ) long_pullback_restart = ( (bull_strong | bull_weak | range_regime) & base_filter & ~bear & dataframe["pullback_seen_long"] & self._crossed_above(dataframe["close"], dataframe["ema_fast"]) & dataframe["center_up"] & (dataframe["rsi"] > 52) & (dataframe["rsi"] < 68) ) short_trend_regime = bear | range_regime if self.allow_range_short_trend else bear short_trend_breakdown = ( short_trend_regime & base_filter & short_4h_trend_filter & short_4h_breakdown_filter & dataframe["breakout_vol_ok"] & dataframe["ema_slow_slope_down"] & (dataframe["rsi"] < 46) & (dataframe["close"] < dataframe["recent_low"] * 0.990) & (dataframe["body_pct"] > 0.010) ) short_pullback_fail = ( bear & base_filter & dataframe["pullback_seen_short"] & self._crossed_below(dataframe["close"], dataframe["ema_fast"]) & dataframe["center_down"] & (dataframe["rsi"] < 48) ) if self.enable_long_trend_breakout: dataframe.loc[long_trend_breakout, ["enter_long", "enter_tag"]] = (1, "long_trend_breakout") if self.enable_long_pullback_restart: dataframe.loc[long_pullback_restart, ["enter_long", "enter_tag"]] = (1, "long_pullback_restart") if self.enable_short_trend_breakdown: dataframe.loc[short_trend_breakdown, ["enter_short", "enter_tag"]] = (1, "short_trend_breakdown") if self.enable_short_pullback_fail: dataframe.loc[short_pullback_fail, ["enter_short", "enter_tag"]] = (1, "short_pullback_fail") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe def _current_candle(self, pair: str): if not self.dp: return None dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe.empty: return None return dataframe.iloc[-1] def _signal_quality_multiplier( self, pair: str, candle, entry_tag: Optional[str], side: str, ) -> float: if not self.use_signal_quality_stake_scale or candle is None: return 1.0 if side == "long" and entry_tag == "long_trend_breakout": quality = 0.78 if bool(candle.get("trend_long_ok_4h", False)): quality *= 1.08 else: quality *= 0.70 regime = int(candle.get("regime_id_1d", 0) or 0) if regime == 3: quality *= 1.10 elif regime == 2: quality *= 0.86 else: quality *= 0.60 rsi = float(candle.get("rsi", 50.0) or 50.0) body_pct = float(candle.get("body_pct", 0.0) or 0.0) volume_mean = float(candle.get("volume_mean", 0.0) or 0.0) volume = float(candle.get("volume", 0.0) or 0.0) if 58 <= rsi <= 68: quality *= 1.08 elif rsi > 72: quality *= 0.76 if body_pct > 0.018: quality *= 1.06 elif body_pct < 0.012: quality *= 0.92 if volume_mean > 0: volume_ratio = volume / volume_mean if volume_ratio > 2.4: quality *= 1.06 elif volume_ratio < 1.8: quality *= 0.90 return max(0.40, min(quality, 1.05)) if side != "short" or entry_tag != "short_trend_breakdown": return 1.0 quality = 1.0 if bool(candle.get("breakdown_short_ok_4h", False)): quality *= 1.08 elif bool(candle.get("trend_short_ok_4h", False)): quality *= 0.92 else: quality *= 0.75 if pair in self.major_pairs: quality *= 0.82 rsi = float(candle.get("rsi", 50.0) or 50.0) body_pct = float(candle.get("body_pct", 0.0) or 0.0) volume_mean = float(candle.get("volume_mean", 0.0) or 0.0) volume = float(candle.get("volume", 0.0) or 0.0) if rsi < 38: quality *= 1.08 elif rsi > 44: quality *= 0.88 if body_pct > 0.018: quality *= 1.06 elif body_pct < 0.012: quality *= 0.92 if volume_mean > 0: volume_ratio = volume / volume_mean if volume_ratio > 2.4: quality *= 1.06 elif volume_ratio < 1.8: quality *= 0.92 return max(0.55, min(quality, 1.18)) def custom_stake_amount( self, pair: str, current_time, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs, ) -> float: candle = self._current_candle(pair) signal_mult = self.signal_multipliers.get(entry_tag or "", 0.70) pair_mult = self.pair_multipliers.get(pair, 0.70) if candle is None: regime_mult = 1.0 elif side == "short": regime_mult = float(candle.get("regime_short_multiplier_1d", 0.6) or 0.6) else: regime_mult = float(candle.get("regime_long_multiplier_1d", 0.6) or 0.6) quality_mult = self._signal_quality_multiplier(pair, candle, entry_tag, side) final_mult = max(0.35, min(signal_mult * regime_mult * pair_mult * quality_mult, 1.3)) stake = proposed_stake * final_mult if min_stake is not None and stake < min_stake: return 0.0 return min(stake, max_stake) def custom_stoploss( self, pair: str, trade: Trade, current_time, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> Optional[float]: candle = self._current_candle(pair) if candle is None: return self.stoploss if trade.is_short: structure_stop = candle.get("structure_stop_short") capped_stop = trade.open_rate * 1.03 stop_price = capped_stop if pd.isna(structure_stop) else min(float(structure_stop), capped_stop) else: structure_stop = candle.get("structure_stop_long") capped_stop = trade.open_rate * 0.97 stop_price = capped_stop if pd.isna(structure_stop) else max(float(structure_stop), capped_stop) return stoploss_from_absolute( stop_price, current_rate, is_short=trade.is_short, leverage=trade.leverage, ) def custom_exit( self, pair: str, trade: Trade, current_time, current_rate: float, current_profit: float, **kwargs, ) -> Optional[str]: candle = self._current_candle(pair) if candle is None: return None age = current_time - trade.open_date_utc if age > timedelta(hours=72) and current_profit < 0: return "stale_loss_72h" if age > timedelta(hours=120) and current_profit < 0.01: return "stale_flat_120h" if age > timedelta(hours=240) and current_profit < 0.03: return "stale_low_profit_240h" regime = int(candle.get("regime_id_1d", 0) or 0) if trade.is_short: if regime in (2, 3) and bool(candle.get("center_up", False)) and current_profit > -0.01: return "trend_flip_short" if current_profit > 0.005 and bool(candle.get("center_up", False)) and candle["close"] > candle.get("ema_fast", candle["close"]): return "structure_exit_short" stop_short = candle.get("structure_stop_short") if pd.notna(stop_short) and candle["close"] > float(stop_short): return "swing_exit_short" return None if regime == -1 and bool(candle.get("center_down", False)) and current_profit > -0.01: return "trend_flip_long" if current_profit > 0.005 and bool(candle.get("center_down", False)) and candle["close"] < candle.get("ema_fast", candle["close"]): return "structure_exit_long" stop_long = candle.get("structure_stop_long") if pd.notna(stop_long) and candle["close"] < float(stop_long): return "swing_exit_long" return None class Top9RegimeStructureNoLongPullbackStrategy(Github_wangc_joker_ft_userdata__Top9RegimeStructureStrategy__20260515_060339): """V1 experiment: remove the currently weakest long pullback entry.""" enable_long_pullback_restart = False class Top9RegimeStructureTrendOnlyStrategy(Github_wangc_joker_ft_userdata__Top9RegimeStructureStrategy__20260515_060339): """V1 experiment: keep only trend breakout/breakdown entries.""" enable_long_pullback_restart = False enable_short_pullback_fail = False class Top9RegimeStructureCorePairsStrategy(Github_wangc_joker_ft_userdata__Top9RegimeStructureStrategy__20260515_060339): """V1 experiment: remove DOGE/BNB/TRX while keeping the same signal set.""" allowed_pairs = { "BTC/USDT:USDT", "ETH/USDT:USDT", "SOL/USDT:USDT", "ADA/USDT:USDT", "ZEC/USDT:USDT", "XRP/USDT:USDT", } class Top9RegimeStructureCoreTrendOnlyStrategy(Top9RegimeStructureCorePairsStrategy): """V1 experiment: core pairs plus only trend breakout/breakdown entries.""" enable_long_pullback_restart = False enable_short_pullback_fail = False class Top9RegimeStructureShortOnlyStrategy(Github_wangc_joker_ft_userdata__Top9RegimeStructureStrategy__20260515_060339): """V1 experiment: disable longs and keep both short entries.""" enable_long_trend_breakout = False enable_long_pullback_restart = False class Top9RegimeStructureShortTrendOnlyStrategy(Github_wangc_joker_ft_userdata__Top9RegimeStructureStrategy__20260515_060339): """V1 experiment: keep only the strongest short trend breakdown entry.""" enable_long_trend_breakout = False enable_long_pullback_restart = False enable_short_pullback_fail = False class Top9RegimeStructureCoreShortOnlyStrategy(Top9RegimeStructureCorePairsStrategy): """V1 experiment: core pairs, short entries only.""" enable_long_trend_breakout = False enable_long_pullback_restart = False class Top9RegimeStructureCoreShortOnlyTightStopStrategy(Top9RegimeStructureCoreShortOnlyStrategy): """V1 experiment: core short-only with a tighter 2% structural stop cap.""" stoploss = -0.02 def custom_stoploss( self, pair: str, trade: Trade, current_time, current_rate: float, current_profit: float, after_fill: bool, **kwargs, ) -> Optional[float]: candle = self._current_candle(pair) if candle is None: return self.stoploss if trade.is_short: structure_stop = candle.get("structure_stop_short") capped_stop = trade.open_rate * 1.02 stop_price = capped_stop if pd.isna(structure_stop) else min(float(structure_stop), capped_stop) else: structure_stop = candle.get("structure_stop_long") capped_stop = trade.open_rate * 0.98 stop_price = capped_stop if pd.isna(structure_stop) else max(float(structure_stop), capped_stop) return stoploss_from_absolute( stop_price, current_rate, is_short=trade.is_short, leverage=trade.leverage, ) class Top9RegimeStructureCoreShortTrendTightStopStrategy( Top9RegimeStructureCoreShortOnlyTightStopStrategy ): """V1 experiment: core pairs, tight stop, only short trend breakdown.""" enable_short_pullback_fail = False class Top9RegimeStructureCoreShortPullbackTightStopStrategy( Top9RegimeStructureCoreShortOnlyTightStopStrategy ): """V1 experiment: core pairs, tight stop, only short pullback fail.""" enable_short_trend_breakdown = False class Top9RegimeStructureAltShortOnlyTightStopStrategy( Top9RegimeStructureCoreShortOnlyTightStopStrategy ): """V1 experiment: remove currently weak BTC/ETH from the short-only basket.""" allowed_pairs = { "SOL/USDT:USDT", "ADA/USDT:USDT", "ZEC/USDT:USDT", "XRP/USDT:USDT", } class Top9RegimeStructureAltShortTrendTightStopStrategy( Top9RegimeStructureAltShortOnlyTightStopStrategy ): """V1 experiment: alt basket, tight stop, only short trend breakdown.""" enable_short_pullback_fail = False class Top9RegimeStructureBearOnlyShortTrendTightStopStrategy( Top9RegimeStructureCoreShortTrendTightStopStrategy ): """V1 experiment: core trend shorts, but reject range-regime breakdowns.""" allow_range_short_trend = False class Top9RegimeStructureAltBearOnlyShortTrendTightStopStrategy( Top9RegimeStructureAltShortTrendTightStopStrategy ): """V1 experiment: alt trend shorts, but reject range-regime breakdowns.""" allow_range_short_trend = False class Top9RegimeStructureBear4hShortTrendTightStopStrategy( Top9RegimeStructureBearOnlyShortTrendTightStopStrategy ): """V1 experiment: 1d bear trend shorts need 4h bearish confirmation.""" require_4h_short_trend = True class Top9RegimeStructureAltBear4hShortTrendTightStopStrategy( Top9RegimeStructureAltBearOnlyShortTrendTightStopStrategy ): """V1 experiment: alt bear trend shorts need 4h bearish confirmation.""" require_4h_short_trend = True class Top9RegimeStructureBear4hBreakdownShortTrendTightStopStrategy( Top9RegimeStructureBearOnlyShortTrendTightStopStrategy ): """V1 experiment: 1d bear trend shorts need a 4h structural breakdown.""" require_4h_short_trend = True require_4h_short_breakdown = True class Top9RegimeStructureMajorStrict4hShortTrendTightStopStrategy( Top9RegimeStructureBear4hShortTrendTightStopStrategy ): """V1 experiment: 4h trend for all shorts, stricter 4h breakdown for BTC/ETH.""" require_major_4h_short_breakdown = True class Top9RegimeStructureQualityStake4hShortTrendTightStopStrategy( Top9RegimeStructureBear4hShortTrendTightStopStrategy ): """V1 experiment: keep V1.16 entries, but scale stake by signal quality.""" use_signal_quality_stake_scale = True class Top9RegimeStructureV2LongBreakoutQualityStrategy( Top9RegimeStructureQualityStake4hShortTrendTightStopStrategy ): """V2.1/V2.2: add conservative bull-regime long breakouts to V1.20.""" enable_long_trend_breakout = True enable_long_pullback_restart = False require_4h_long_trend = True require_strong_bull_long = True use_v2_long_breakout = True signal_multipliers = { **Top9RegimeStructureQualityStake4hShortTrendTightStopStrategy.signal_multipliers, "long_trend_breakout": 0.72, }