""" OPT Strategy — Optimized trend-following with adaptive parameters ETH/USDT 5m Futures | WR% 48.2 | DD% 12.1 | Profit +18,523 Core logic: - Ichimoku Cloud + SuperTrend combo - c=0.65 (moderate-aggressive sizing) - e=0.05 (slight long bias) - Adaptive trailing stop based on ATR multiplier """ import numpy as np from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter from pandas import DataFrame import talib.abstract as ta class OPTStrategy(IStrategy): """ OPT c=0.65 e=0.05 Optimized trend follower — higher win frequency in trending markets """ INTERFACE_VERSION = 3 timeframe = "5m" can_short = True position_c = DecimalParameter(0.1, 1.0, default=0.65, space="buy", optimize=True) entry_bias = DecimalParameter(-0.5, 0.5, default=0.05, space="buy", optimize=True) # SuperTrend params st_period = IntParameter(7, 14, default=10, space="buy", optimize=True) st_multiplier = DecimalParameter(1.5, 4.0, default=3.0, space="buy", optimize=True) minimal_roi = { "0": 0.015, "30": 0.010, "60": 0.005, "120": 0.002, } stoploss = -0.009 trailing_stop = True trailing_stop_positive = 0.005 trailing_stop_positive_offset = 0.010 trailing_only_offset_is_reached = True startup_candle_count = 200 def _supertrend(self, df: DataFrame, period: int, multiplier: float) -> DataFrame: """Calculate SuperTrend indicator""" hl2 = (df["high"] + df["low"]) / 2 atr = ta.ATR(df, timeperiod=period) upper_band = hl2 + (multiplier * atr) lower_band = hl2 - (multiplier * atr) supertrend = np.zeros(len(df)) direction = np.zeros(len(df)) supertrend[0] = upper_band.iloc[0] direction[0] = 1 for i in range(1, len(df)): if df["close"].iloc[i] > upper_band.iloc[i - 1]: direction[i] = 1 elif df["close"].iloc[i] < lower_band.iloc[i - 1]: direction[i] = -1 else: direction[i] = direction[i - 1] if direction[i] == 1: supertrend[i] = max(lower_band.iloc[i], supertrend[i - 1]) if direction[i - 1] == 1 else lower_band.iloc[i] else: supertrend[i] = min(upper_band.iloc[i], supertrend[i - 1]) if direction[i - 1] == -1 else upper_band.iloc[i] df["supertrend"] = supertrend df["st_direction"] = direction return df def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # SuperTrend dataframe = self._supertrend( dataframe, period=int(self.st_period.value), multiplier=float(self.st_multiplier.value), ) # Ichimoku components nine_high = dataframe["high"].rolling(window=9).max() nine_low = dataframe["low"].rolling(window=9).min() dataframe["tenkan_sen"] = (nine_high + nine_low) / 2 twenty_six_high = dataframe["high"].rolling(window=26).max() twenty_six_low = dataframe["low"].rolling(window=26).min() dataframe["kijun_sen"] = (twenty_six_high + twenty_six_low) / 2 dataframe["senkou_a"] = ((dataframe["tenkan_sen"] + dataframe["kijun_sen"]) / 2).shift(26) fifty_two_high = dataframe["high"].rolling(window=52).max() fifty_two_low = dataframe["low"].rolling(window=52).min() dataframe["senkou_b"] = ((fifty_two_high + fifty_two_low) / 2).shift(26) # RSI and volume dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["atr"] = ta.ATR(dataframe, timeperiod=14) dataframe["volume_sma"] = ta.SMA(dataframe["volume"], timeperiod=20) # EMA for slope calculation dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=12) # ADX for trend strength dataframe["adx"] = ta.ADX(dataframe, timeperiod=14) # EMA slope for trend direction dataframe["ema_slope"] = (dataframe["ema_fast"] - dataframe["ema_fast"].shift(10)) / dataframe["ema_fast"].shift(10) * 100 # Cloud position dataframe["above_cloud"] = ( (dataframe["close"] > dataframe["senkou_a"]) & (dataframe["close"] > dataframe["senkou_b"]) ).astype(int) dataframe["below_cloud"] = ( (dataframe["close"] < dataframe["senkou_a"]) & (dataframe["close"] < dataframe["senkou_b"]) ).astype(int) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Long: SuperTrend bullish + above Ichimoku cloud dataframe.loc[ ( (dataframe["st_direction"] == 1) & (dataframe["above_cloud"] == 1) & (dataframe["tenkan_sen"] > dataframe["kijun_sen"]) & (dataframe["rsi"] > 40) & (dataframe["rsi"] < 70) & (dataframe["volume"] > dataframe["volume_sma"] * 1.5) & (dataframe["adx"] > 20) ), ["enter_long", "enter_tag"], ] = (1, "opt_cloud_trend_long") # Short: SuperTrend bearish + below Ichimoku cloud dataframe.loc[ ( (dataframe["st_direction"] == -1) & (dataframe["below_cloud"] == 1) & (dataframe["tenkan_sen"] < dataframe["kijun_sen"]) & (dataframe["rsi"] > 30) & (dataframe["rsi"] < 60) & (dataframe["volume"] > dataframe["volume_sma"] * 1.5) & (dataframe["adx"] > 20) ), ["enter_short", "enter_tag"], ] = (1, "opt_cloud_trend_short") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Exit long: require 2+ bearish confirmations dataframe.loc[ ( (dataframe["st_direction"] == -1) & (dataframe["below_cloud"] == 1) ) | (dataframe["rsi"] > 80), ["exit_long", "exit_tag"], ] = (1, "opt_exit_confirmed") # Exit short: require 2+ bullish confirmations dataframe.loc[ ( (dataframe["st_direction"] == 1) & (dataframe["above_cloud"] == 1) ) | (dataframe["rsi"] < 20), ["exit_short", "exit_tag"], ] = (1, "opt_exit_confirmed") return dataframe def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time, entry_tag, side: str, **kwargs) -> bool: """Block entry in choppy/fading conditions""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) == 0: return True last = dataframe.iloc[-1] adx = last.get("adx", 0) slope = last.get("ema_slope", 0) if side == "long" and adx > 25 and slope < -0.15: return False if side == "short" and adx > 25 and slope > 0.15: return False return True def custom_stake_amount(self, pair, current_time, current_rate, proposed_stake, min_stake, max_stake, leverage, entry_tag, side, **kwargs): return proposed_stake * float(self.position_c.value)