from __future__ import annotations from datetime import datetime from pandas import DataFrame import talib.abstract as ta from freqtrade.strategy import DecimalParameter, IStrategy, IntParameter, stoploss_from_absolute class SolLargeRedRebound1hV1(IStrategy): """Research-only SOL rebound after large red 1h candle. Hypothesis from edge lab: - On recent SOL data, large red 1h candles often mean-reverted over the next few hours. - This is a tactical long-only module, not a cycle strategy. Do not use live without a separate dry-run and kill-switch review. """ INTERFACE_VERSION = 3 can_short = False timeframe = "1h" startup_candle_count = 240 process_only_new_candles = True position_adjustment_enable = False max_entry_position_adjustment = 0 minimal_roi = {"0": 0.99} stoploss = -0.06 trailing_stop = False use_custom_stoploss = False use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False take_profit_pct: float | None = None body_recovery_target: float | None = None range_recovery_target: float | None = None dynamic_tp_min_pct = 0.015 dynamic_tp_max_pct = 0.08 break_even_after_pct: float | None = None break_even_stop_pct = 0.001 trail_after_pct: float | None = None trail_distance_pct: float | None = None leverage_value = 1.0 min_range_atr = DecimalParameter(1.60, 2.60, default=2.00, decimals=2, space="buy", optimize=False) max_range_atr = DecimalParameter(2.60, 4.00, default=3.80, decimals=2, space="buy", optimize=False) min_volume_ratio = DecimalParameter(0.50, 1.50, default=0.80, decimals=2, space="buy", optimize=False) max_btc_drop_6h = DecimalParameter(-0.12, -0.03, default=-0.07, decimals=3, space="buy", optimize=False) time_exit_candles = IntParameter(4, 24, default=8, space="sell", optimize=False) panic_exit_btc_drop_4h = DecimalParameter(-0.12, -0.03, default=-0.08, decimals=3, space="sell", optimize=False) @property def protections(self) -> list[dict]: return [ {"method": "CooldownPeriod", "stop_duration_candles": 2}, { "method": "StoplossGuard", "lookback_period_candles": 48, "trade_limit": 2, "stop_duration_candles": 24, "required_profit": 0.0, "only_per_pair": True, "only_per_side": False, }, { "method": "MaxDrawdown", "calculation_mode": "equity", "lookback_period_candles": 168, "trade_limit": 10, "stop_duration_candles": 24, "max_allowed_drawdown": 0.08, }, ] @staticmethod def _is_sol_pair(pair: str) -> bool: return pair.upper().startswith("SOL/") def _btc_reference_pair(self) -> str: stake_currency = str(self.config.get("stake_currency", "USDT")).upper() exchange_name = str(self.config.get("exchange", {}).get("name", "")).lower() if stake_currency == "USDC" or exchange_name == "hyperliquid": return "BTC/USDC:USDC" return "BTC/USDT:USDT" def informative_pairs(self) -> list[tuple[str, str]]: return [(self._btc_reference_pair(), self.timeframe)] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20) dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50) dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200) dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["volume_mean_24"] = dataframe["volume"].rolling(24, min_periods=24).mean() dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_mean_24"] dataframe["range_atr"] = (dataframe["high"] - dataframe["low"]) / dataframe["atr"] dataframe["red_candle"] = dataframe["close"] < dataframe["open"] dataframe["ret_6"] = dataframe["close"].pct_change(6) dataframe["ret_24"] = dataframe["close"].pct_change(24) if self.dp: btc = self.dp.get_pair_dataframe(pair=self._btc_reference_pair(), timeframe=self.timeframe) if not btc.empty: btc = btc[["date", "close"]].copy() btc["btc_ret_6"] = btc["close"].pct_change(6) btc = btc[["date", "btc_ret_6"]] dataframe = dataframe.merge(btc, on="date", how="left") if "btc_ret_6" not in dataframe.columns: dataframe["btc_ret_6"] = 0.0 return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["enter_long"] = 0 if not self._is_sol_pair(metadata["pair"]): return dataframe valid = ( (dataframe["volume"] > 0) & dataframe["atr"].notna() & (dataframe["volume_mean_24"] > 0) & dataframe["btc_ret_6"].notna() ) large_red_rebound = ( valid & dataframe["red_candle"] & (dataframe["range_atr"] >= self.min_range_atr.value) & (dataframe["range_atr"] <= self.max_range_atr.value) & (dataframe["volume_ratio"] >= self.min_volume_ratio.value) & (dataframe["btc_ret_6"] > self.max_btc_drop_6h.value) ) dataframe.loc[large_red_rebound, ["enter_long", "enter_tag"]] = (1, "large_red_rebound") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["exit_long"] = 0 return dataframe @staticmethod def _signal_candle(dataframe: DataFrame, trade): frame = dataframe.loc[dataframe["date"] < trade.open_date_utc] if frame.empty: frame = dataframe.loc[dataframe["date"] <= trade.open_date_utc] if frame.empty: return None return frame.iloc[-1] def _body_recovery_profit_target(self, dataframe: DataFrame, trade) -> float | None: if self.body_recovery_target is None: return None signal = self._signal_candle(dataframe, trade) if signal is None: return None signal_open = float(signal.get("open", 0.0)) signal_close = float(signal.get("close", 0.0)) if signal_open <= signal_close or signal_close <= 0: return None recovery_price = signal_close + ((signal_open - signal_close) * self.body_recovery_target) raw_profit_target = (recovery_price / float(trade.open_rate)) - 1.0 return min(max(raw_profit_target, self.dynamic_tp_min_pct), self.dynamic_tp_max_pct) def _range_recovery_profit_target(self, dataframe: DataFrame, trade) -> float | None: if self.range_recovery_target is None: return None signal = self._signal_candle(dataframe, trade) if signal is None: return None signal_high = float(signal.get("high", 0.0)) signal_low = float(signal.get("low", 0.0)) signal_close = float(signal.get("close", 0.0)) if signal_high <= signal_low or signal_close <= 0: return None candle_range_pct = (signal_high - signal_low) / signal_close raw_profit_target = candle_range_pct * self.range_recovery_target return min(max(raw_profit_target, self.dynamic_tp_min_pct), self.dynamic_tp_max_pct) def custom_exit( self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs, ) -> str | bool | None: elapsed_candles = int((current_time - trade.open_date_utc).total_seconds() // 3600) if self.take_profit_pct is not None and current_profit >= self.take_profit_pct: return f"take_profit_{self.take_profit_pct:.0%}" if self.dp: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if not dataframe.empty: dynamic_target = self._range_recovery_profit_target(dataframe, trade) if dynamic_target is not None and current_profit >= dynamic_target: return f"range_recovery_{self.range_recovery_target:.0%}" dynamic_target = self._body_recovery_profit_target(dataframe, trade) if dynamic_target is not None and current_profit >= dynamic_target: return f"body_recovery_{self.body_recovery_target:.0%}" candle = dataframe.iloc[-1] if elapsed_candles >= 2 and float(candle.get("btc_ret_6", 0.0)) <= self.panic_exit_btc_drop_4h.value: return "btc_panic_exit" if elapsed_candles >= self.time_exit_candles.value: return f"time_exit_{self.time_exit_candles.value}h" return None def custom_stoploss( self, pair: str, trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool = False, **kwargs, ) -> float | None: if trade.is_short: return None stop_level: float | None = None if self.break_even_after_pct is not None and current_profit >= self.break_even_after_pct: stop_level = float(trade.open_rate) * (1 + self.break_even_stop_pct) if ( self.trail_after_pct is not None and self.trail_distance_pct is not None and current_profit >= self.trail_after_pct ): trailing_level = current_rate * (1 - self.trail_distance_pct) stop_level = max(stop_level or 0.0, trailing_level) if stop_level is None: return None if stop_level >= current_rate: return 0.001 return stoploss_from_absolute( stop_level, current_rate, is_short=False, leverage=getattr(trade, "leverage", 1.0) or 1.0, ) def leverage( self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str, **kwargs, ) -> float: return min(self.leverage_value, max_leverage) class SolLargeRedRebound1hV1Exit8h(SolLargeRedRebound1hV1): time_exit_candles = IntParameter(4, 24, default=8, space="sell", optimize=False) class SolLargeRedRebound1hV1Exit24h(SolLargeRedRebound1hV1): time_exit_candles = IntParameter(4, 24, default=24, space="sell", optimize=False) class SolLargeRedRebound1hV1Exit24hLev2(SolLargeRedRebound1hV1Exit24h): leverage_value = 2.0 class SolLargeRedRebound1hV1Exit24hLev3(SolLargeRedRebound1hV1Exit24h): leverage_value = 3.0 class SolLargeRedRebound1hV1Exit24hTp2(SolLargeRedRebound1hV1Exit24h): take_profit_pct = 0.02 class SolLargeRedRebound1hV1Exit24hTp3(SolLargeRedRebound1hV1Exit24h): take_profit_pct = 0.03 class SolLargeRedRebound1hV1Exit24hTp4(SolLargeRedRebound1hV1Exit24h): take_profit_pct = 0.04 class SolLargeRedRebound1hV1Exit24hBody35(SolLargeRedRebound1hV1Exit24h): body_recovery_target = 0.35 class SolLargeRedRebound1hV1Exit24hBody50(SolLargeRedRebound1hV1Exit24h): body_recovery_target = 0.50 class SolLargeRedRebound1hV1Exit24hBody65(SolLargeRedRebound1hV1Exit24h): body_recovery_target = 0.65 class SolLargeRedRebound1hV1Exit24hRange35(SolLargeRedRebound1hV1Exit24h): range_recovery_target = 0.35 class SolLargeRedRebound1hV1Exit24hRange50(SolLargeRedRebound1hV1Exit24h): range_recovery_target = 0.50 class SolLargeRedRebound1hV1Exit24hRange65(SolLargeRedRebound1hV1Exit24h): range_recovery_target = 0.65 class SolLargeRedRebound1hV1Exit24hBeOnly(SolLargeRedRebound1hV1Exit24h): use_custom_stoploss = True break_even_after_pct = 0.015 break_even_stop_pct = 0.001 class SolLargeRedRebound1hV1Exit24hTp4Be(SolLargeRedRebound1hV1Exit24h): take_profit_pct = 0.04 use_custom_stoploss = True break_even_after_pct = 0.015 break_even_stop_pct = 0.001 class SolLargeRedRebound1hV1Exit24hBeTrail(SolLargeRedRebound1hV1Exit24h): use_custom_stoploss = True break_even_after_pct = 0.015 break_even_stop_pct = 0.001 trail_after_pct = 0.035 trail_distance_pct = 0.025