# source: https://raw.githubusercontent.com/markelayan/hhec-dois-bull-manager/09fa8d07121e68880686b2f9b8d4201bca0dd669/1strategies1/ClaudePricePrediction.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these libs --- import numpy as np import pandas as pd from pandas import DataFrame from datetime import datetime, timedelta from typing import Optional, Union import logging from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, IStrategy, merge_informative_pair) # -------------------------------- # Add your lib to import here import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from scipy import stats from scipy.signal import find_peaks, find_peaks_cwt from sklearn.preprocessing import StandardScaler from sklearn.cluster import KMeans import warnings warnings.filterwarnings('ignore') logger = logging.getLogger(__name__) class Github_markelayan_hhec_dois_bull_manager__ClaudePricePrediction__20250810_092830(IStrategy): """ Advanced Price Prediction Strategy with Support/Resistance Detection This strategy combines machine learning with sophisticated technical analysis: - Multi-timeframe price prediction using FreqAI - Dynamic support and resistance level detection - Volume profile analysis for key price zones - Advanced pattern recognition - Statistical analysis of price movements - Risk-adjusted position sizing - Profitable entry/exit timing optimization Features: - Machine learning price targets - Support/resistance breakout detection - Volume-weighted price levels - Fibonacci retracement levels - Market regime detection - Risk parity position sizing - Advanced exit strategies """ INTERFACE_VERSION = 3 # Optimized ROI for ML predictions minimal_roi = { "0": 0.15, # 15% at any time (ML confidence high) "10": 0.08, # 8% after 10 minutes "20": 0.05, # 5% after 20 minutes "40": 0.03, # 3% after 40 minutes "80": 0.015, # 1.5% after 80 minutes "120": 0.01 # 1% after 2 hours } # Dynamic stoploss based on volatility stoploss = -0.06 # 6% base stoploss # Multi-timeframe approach timeframe = '15m' # Informative timeframes for comprehensive analysis inf_5m = '5m' inf_1h = '1h' inf_4h = '4h' inf_1d = '1d' # Process only new candles process_only_new_candles = True # Use ML-enhanced exit signals use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = True # Extended startup for comprehensive analysis startup_candle_count: int = 400 # === Strategy Parameters - Hyperopt Optimized === # === ML Prediction Parameters === prediction_confidence_threshold = DecimalParameter(0.6, 0.9, default=0.75, space='buy', optimize=True) prediction_horizon = IntParameter(12, 48, default=24, space='buy', optimize=True) ml_signal_weight = DecimalParameter(0.3, 0.8, default=0.5, space='buy', optimize=True) # === Support/Resistance Parameters === sr_lookback_period = IntParameter(50, 200, default=100, space='buy', optimize=True) sr_touch_tolerance = DecimalParameter(0.002, 0.01, default=0.005, space='buy', optimize=True) sr_strength_threshold = IntParameter(3, 8, default=5, space='buy', optimize=True) sr_break_threshold = DecimalParameter(0.003, 0.015, default=0.008, space='buy', optimize=True) # === Volume Profile Parameters === volume_profile_periods = IntParameter(100, 300, default=200, space='buy', optimize=True) poc_deviation_threshold = DecimalParameter(0.01, 0.05, default=0.02, space='buy', optimize=True) # === Fibonacci Parameters === fib_lookback = IntParameter(20, 100, default=50, space='buy', optimize=True) fib_deviation = DecimalParameter(0.005, 0.02, default=0.01, space='buy', optimize=True) # === Trend Parameters === trend_ema_fast = IntParameter(12, 26, default=20, space='buy', optimize=True) trend_ema_slow = IntParameter(40, 80, default=60, space='buy', optimize=True) trend_strength_threshold = DecimalParameter(0.02, 0.08, default=0.04, space='buy', optimize=True) # === Risk Parameters === volatility_threshold = DecimalParameter(0.01, 0.05, default=0.03, space='buy', optimize=True) risk_reward_ratio = DecimalParameter(1.5, 4.0, default=2.5, space='buy', optimize=True) # === Exit Parameters === take_profit_multiplier = DecimalParameter(1.5, 3.5, default=2.0, space='sell', optimize=True) trailing_stop_activation = DecimalParameter(0.02, 0.08, default=0.04, space='sell', optimize=True) # Position sizing max_open_trades = 3 position_adjustment_enable = True def informative_pairs(self): """ Define informative pairs for comprehensive market analysis """ pairs = self.dp.current_whitelist() informative_pairs = [] # Add all timeframes for current pairs for pair in pairs: informative_pairs.extend([ (pair, self.inf_5m), (pair, self.inf_1h), (pair, self.inf_4h), (pair, self.inf_1d) ]) # Add major pairs for correlation analysis major_pairs = ['BTC/USDT:USDT', 'ETH/USDT:USDT'] for pair in major_pairs: if pair not in pairs: informative_pairs.extend([ (pair, self.timeframe), (pair, self.inf_1h), (pair, self.inf_4h) ]) return informative_pairs def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Populate all indicators including ML features and S/R levels """ # === Basic Price Data === dataframe['hl2'] = (dataframe['high'] + dataframe['low']) / 2 dataframe['hlc3'] = (dataframe['high'] + dataframe['low'] + dataframe['close']) / 3 dataframe['ohlc4'] = (dataframe['open'] + dataframe['high'] + dataframe['low'] + dataframe['close']) / 4 # === Trend Analysis === dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.trend_ema_fast.value) dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.trend_ema_slow.value) dataframe['ema_trend'] = dataframe['ema_fast'] > dataframe['ema_slow'] dataframe['trend_strength'] = abs((dataframe['ema_fast'] - dataframe['ema_slow']) / dataframe['close']) # === Volatility === dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['volatility'] = dataframe['atr'] / dataframe['close'] # === Volume Analysis === dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20) dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] dataframe['vwap'] = qtpylib.vwap(dataframe) # === Price Action Patterns === dataframe = self.calculate_price_patterns(dataframe) # === Support/Resistance Levels === dataframe = self.calculate_support_resistance(dataframe) # === Volume Profile === dataframe = self.calculate_volume_profile(dataframe) # === Fibonacci Levels === dataframe = self.calculate_fibonacci_levels(dataframe) # === Market Regime Detection === dataframe = self.detect_market_regime(dataframe) # === Advanced Technical Indicators === dataframe = self.calculate_advanced_indicators(dataframe) # === Higher Timeframe Analysis === dataframe = self.populate_higher_timeframe_data(dataframe, metadata) # === FreqAI Features === dataframe = self.populate_freqai_features(dataframe) return dataframe def calculate_support_resistance(self, dataframe: DataFrame) -> DataFrame: """ Advanced support and resistance level detection """ # Calculate pivot points using different methods dataframe = self.calculate_pivot_points(dataframe) # Find significant highs and lows high_peaks = find_peaks(dataframe['high'].values, distance=10, prominence=dataframe['atr'].mean())[0] low_peaks = find_peaks(-dataframe['low'].values, distance=10, prominence=dataframe['atr'].mean())[0] # Initialize S/R columns dataframe['resistance_level'] = np.nan dataframe['support_level'] = np.nan dataframe['resistance_strength'] = 0 dataframe['support_strength'] = 0 dataframe['at_resistance'] = False dataframe['at_support'] = False dataframe['resistance_break'] = False dataframe['support_break'] = False lookback = min(len(dataframe), self.sr_lookback_period.value) for i in range(lookback, len(dataframe)): current_price = dataframe.iloc[i]['close'] tolerance = self.sr_touch_tolerance.value # Get recent peaks for resistance recent_highs = [] for peak in high_peaks: if i - lookback <= peak < i: peak_price = dataframe.iloc[peak]['high'] # Check if this level was tested multiple times touches = 0 for j in range(max(0, peak - 20), min(len(dataframe), peak + 20)): if abs(dataframe.iloc[j]['high'] - peak_price) / peak_price <= tolerance: touches += 1 if touches >= self.sr_strength_threshold.value: recent_highs.append((peak_price, touches)) # Get recent lows for support recent_lows = [] for peak in low_peaks: if i - lookback <= peak < i: peak_price = dataframe.iloc[peak]['low'] # Check if this level was tested multiple times touches = 0 for j in range(max(0, peak - 20), min(len(dataframe), peak + 20)): if abs(dataframe.iloc[j]['low'] - peak_price) / peak_price <= tolerance: touches += 1 if touches >= self.sr_strength_threshold.value: recent_lows.append((peak_price, touches)) # Find closest resistance and support if recent_highs: resistance_candidates = [(level, strength) for level, strength in recent_highs if level > current_price] if resistance_candidates: resistance_level, resistance_strength = min(resistance_candidates, key=lambda x: x[0]) dataframe.iloc[i, dataframe.columns.get_loc('resistance_level')] = resistance_level dataframe.iloc[i, dataframe.columns.get_loc('resistance_strength')] = resistance_strength # Check if at resistance if abs(current_price - resistance_level) / resistance_level <= tolerance: dataframe.iloc[i, dataframe.columns.get_loc('at_resistance')] = True # Check for resistance break if current_price > resistance_level * (1 + self.sr_break_threshold.value): dataframe.iloc[i, dataframe.columns.get_loc('resistance_break')] = True if recent_lows: support_candidates = [(level, strength) for level, strength in recent_lows if level < current_price] if support_candidates: support_level, support_strength = max(support_candidates, key=lambda x: x[0]) dataframe.iloc[i, dataframe.columns.get_loc('support_level')] = support_level dataframe.iloc[i, dataframe.columns.get_loc('support_strength')] = support_strength # Check if at support if abs(current_price - support_level) / support_level <= tolerance: dataframe.iloc[i, dataframe.columns.get_loc('at_support')] = True # Check for support break if current_price < support_level * (1 - self.sr_break_threshold.value): dataframe.iloc[i, dataframe.columns.get_loc('support_break')] = True return dataframe def calculate_pivot_points(self, dataframe: DataFrame) -> DataFrame: """ Calculate various pivot point levels """ # Standard Pivot Points dataframe['pivot'] = (dataframe['high'].shift(1) + dataframe['low'].shift(1) + dataframe['close'].shift(1)) / 3 dataframe['r1'] = 2 * dataframe['pivot'] - dataframe['low'].shift(1) dataframe['s1'] = 2 * dataframe['pivot'] - dataframe['high'].shift(1) dataframe['r2'] = dataframe['pivot'] + (dataframe['high'].shift(1) - dataframe['low'].shift(1)) dataframe['s2'] = dataframe['pivot'] - (dataframe['high'].shift(1) - dataframe['low'].shift(1)) # Fibonacci Pivot Points dataframe['fib_r1'] = dataframe['pivot'] + 0.382 * (dataframe['high'].shift(1) - dataframe['low'].shift(1)) dataframe['fib_s1'] = dataframe['pivot'] - 0.382 * (dataframe['high'].shift(1) - dataframe['low'].shift(1)) dataframe['fib_r2'] = dataframe['pivot'] + 0.618 * (dataframe['high'].shift(1) - dataframe['low'].shift(1)) dataframe['fib_s2'] = dataframe['pivot'] - 0.618 * (dataframe['high'].shift(1) - dataframe['low'].shift(1)) return dataframe def calculate_volume_profile(self, dataframe: DataFrame) -> DataFrame: """ Calculate Volume Profile and Point of Control (POC) """ lookback = min(len(dataframe), self.volume_profile_periods.value) dataframe['poc_level'] = np.nan dataframe['vah_level'] = np.nan # Value Area High dataframe['val_level'] = np.nan # Value Area Low dataframe['near_poc'] = False for i in range(lookback, len(dataframe)): # Get price and volume data for the lookback period price_data = dataframe.iloc[i-lookback:i]['hlc3'].values volume_data = dataframe.iloc[i-lookback:i]['volume'].values # Create price bins price_min, price_max = price_data.min(), price_data.max() price_bins = np.linspace(price_min, price_max, 50) # Calculate volume at each price level volume_profile = [] for j in range(len(price_bins)-1): bin_volume = 0 bin_center = (price_bins[j] + price_bins[j+1]) / 2 for k, price in enumerate(price_data): if price_bins[j] <= price <= price_bins[j+1]: bin_volume += volume_data[k] volume_profile.append((bin_center, bin_volume)) if volume_profile: # Find Point of Control (highest volume level) poc_level = max(volume_profile, key=lambda x: x[1])[0] dataframe.iloc[i, dataframe.columns.get_loc('poc_level')] = poc_level # Calculate Value Area (70% of total volume) total_volume = sum([x[1] for x in volume_profile]) target_volume = total_volume * 0.7 # Sort by volume and find value area sorted_profile = sorted(volume_profile, key=lambda x: x[1], reverse=True) cumulative_volume = 0 value_area_prices = [] for price, volume in sorted_profile: cumulative_volume += volume value_area_prices.append(price) if cumulative_volume >= target_volume: break if value_area_prices: vah = max(value_area_prices) val = min(value_area_prices) dataframe.iloc[i, dataframe.columns.get_loc('vah_level')] = vah dataframe.iloc[i, dataframe.columns.get_loc('val_level')] = val # Check if current price is near POC current_price = dataframe.iloc[i]['close'] if abs(current_price - poc_level) / poc_level <= self.poc_deviation_threshold.value: dataframe.iloc[i, dataframe.columns.get_loc('near_poc')] = True return dataframe def calculate_fibonacci_levels(self, dataframe: DataFrame) -> DataFrame: """ Calculate dynamic Fibonacci retracement levels """ lookback = min(len(dataframe), self.fib_lookback.value) dataframe['fib_0'] = np.nan dataframe['fib_236'] = np.nan dataframe['fib_382'] = np.nan dataframe['fib_500'] = np.nan dataframe['fib_618'] = np.nan dataframe['fib_786'] = np.nan dataframe['fib_1000'] = np.nan dataframe['near_fib_level'] = False for i in range(lookback, len(dataframe)): # Find swing high and low in lookback period high_data = dataframe.iloc[i-lookback:i]['high'] low_data = dataframe.iloc[i-lookback:i]['low'] swing_high = high_data.max() swing_low = low_data.min() # Calculate Fibonacci levels diff = swing_high - swing_low dataframe.iloc[i, dataframe.columns.get_loc('fib_0')] = swing_high dataframe.iloc[i, dataframe.columns.get_loc('fib_236')] = swing_high - 0.236 * diff dataframe.iloc[i, dataframe.columns.get_loc('fib_382')] = swing_high - 0.382 * diff dataframe.iloc[i, dataframe.columns.get_loc('fib_500')] = swing_high - 0.5 * diff dataframe.iloc[i, dataframe.columns.get_loc('fib_618')] = swing_high - 0.618 * diff dataframe.iloc[i, dataframe.columns.get_loc('fib_786')] = swing_high - 0.786 * diff dataframe.iloc[i, dataframe.columns.get_loc('fib_1000')] = swing_low # Check if price is near any Fibonacci level current_price = dataframe.iloc[i]['close'] fib_levels = [ dataframe.iloc[i]['fib_236'], dataframe.iloc[i]['fib_382'], dataframe.iloc[i]['fib_500'], dataframe.iloc[i]['fib_618'], dataframe.iloc[i]['fib_786'] ] for fib_level in fib_levels: if not pd.isna(fib_level): if abs(current_price - fib_level) / fib_level <= self.fib_deviation.value: dataframe.iloc[i, dataframe.columns.get_loc('near_fib_level')] = True break return dataframe def detect_market_regime(self, dataframe: DataFrame) -> DataFrame: """ Detect market regime (trending, ranging, volatile) """ # Calculate regime indicators dataframe['regime_adx'] = ta.ADX(dataframe, timeperiod=14) dataframe['regime_atr_ratio'] = dataframe['atr'] / dataframe['close'] # Price efficiency (how much price moved vs total movement) dataframe['price_efficiency'] = abs(dataframe['close'] - dataframe['close'].shift(20)) / ( dataframe['atr'].rolling(20).sum() ) # Market regime classification dataframe['market_regime'] = 'ranging' # Default for i in range(20, len(dataframe)): adx = dataframe.iloc[i]['regime_adx'] efficiency = dataframe.iloc[i]['price_efficiency'] volatility = dataframe.iloc[i]['regime_atr_ratio'] if adx > 25 and efficiency > 0.3: dataframe.iloc[i, dataframe.columns.get_loc('market_regime')] = 'trending' elif volatility > self.volatility_threshold.value: dataframe.iloc[i, dataframe.columns.get_loc('market_regime')] = 'volatile' else: dataframe.iloc[i, dataframe.columns.get_loc('market_regime')] = 'ranging' return dataframe def calculate_price_patterns(self, dataframe: DataFrame) -> DataFrame: """ Detect important price patterns """ # Higher highs and higher lows (uptrend) dataframe['higher_high'] = ( (dataframe['high'] > dataframe['high'].shift(1)) & (dataframe['high'].shift(1) > dataframe['high'].shift(2)) ) dataframe['higher_low'] = ( (dataframe['low'] > dataframe['low'].shift(1)) & (dataframe['low'].shift(1) > dataframe['low'].shift(2)) ) # Lower highs and lower lows (downtrend) dataframe['lower_high'] = ( (dataframe['high'] < dataframe['high'].shift(1)) & (dataframe['high'].shift(1) < dataframe['high'].shift(2)) ) dataframe['lower_low'] = ( (dataframe['low'] < dataframe['low'].shift(1)) & (dataframe['low'].shift(1) < dataframe['low'].shift(2)) ) # Engulfing patterns dataframe['bullish_engulfing'] = ( (dataframe['close'].shift(1) < dataframe['open'].shift(1)) & # Previous red (dataframe['close'] > dataframe['open']) & # Current green (dataframe['open'] < dataframe['close'].shift(1)) & # Opens below previous close (dataframe['close'] > dataframe['open'].shift(1)) # Closes above previous open ) dataframe['bearish_engulfing'] = ( (dataframe['close'].shift(1) > dataframe['open'].shift(1)) & # Previous green (dataframe['close'] < dataframe['open']) & # Current red (dataframe['open'] > dataframe['close'].shift(1)) & # Opens above previous close (dataframe['close'] < dataframe['open'].shift(1)) # Closes below previous open ) return dataframe def calculate_advanced_indicators(self, dataframe: DataFrame) -> DataFrame: """ Calculate advanced technical indicators """ # RSI with multiple periods dataframe['rsi_14'] = ta.RSI(dataframe, timeperiod=14) dataframe['rsi_21'] = ta.RSI(dataframe, timeperiod=21) # MACD macd = ta.MACD(dataframe) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] dataframe['macdhist'] = macd['macdhist'] # Stochastic stoch = ta.STOCH(dataframe) dataframe['stoch_k'] = stoch['slowk'] dataframe['stoch_d'] = stoch['slowd'] # Williams %R dataframe['willr'] = ta.WILLR(dataframe, timeperiod=14) # Commodity Channel Index dataframe['cci'] = ta.CCI(dataframe, timeperiod=20) # Money Flow Index dataframe['mfi'] = ta.MFI(dataframe, timeperiod=14) # Bollinger Bands bollinger = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=2) dataframe['bb_lower'] = bollinger['lower'] dataframe['bb_middle'] = bollinger['mid'] dataframe['bb_upper'] = bollinger['upper'] dataframe['bb_percent'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower']) return dataframe def populate_higher_timeframe_data(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Add higher timeframe analysis """ # 1H timeframe inf_1h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1h) inf_1h['trend_1h'] = ta.EMA(inf_1h, timeperiod=20) < inf_1h['close'] inf_1h['rsi_1h'] = ta.RSI(inf_1h, timeperiod=14) inf_1h['volume_1h'] = inf_1h['volume'] dataframe = merge_informative_pair(dataframe, inf_1h, self.timeframe, self.inf_1h, ffill=True) # 4H timeframe inf_4h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_4h) inf_4h['trend_4h'] = ta.EMA(inf_4h, timeperiod=20) < inf_4h['close'] inf_4h['rsi_4h'] = ta.RSI(inf_4h, timeperiod=14) inf_4h['atr_4h'] = ta.ATR(inf_4h, timeperiod=14) dataframe = merge_informative_pair(dataframe, inf_4h, self.timeframe, self.inf_4h, ffill=True) # Daily timeframe inf_1d = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1d) inf_1d['trend_1d'] = ta.EMA(inf_1d, timeperiod=20) < inf_1d['close'] inf_1d['volatility_1d'] = ta.ATR(inf_1d, timeperiod=14) / inf_1d['close'] dataframe = merge_informative_pair(dataframe, inf_1d, self.timeframe, self.inf_1d, ffill=True) return dataframe def populate_freqai_features(self, dataframe: DataFrame) -> DataFrame: """ Create features specifically for FreqAI ML model """ # === Price-based features === for period in [5, 10, 20, 50]: dataframe[f'price_change_{period}'] = dataframe['close'].pct_change(period) dataframe[f'high_low_ratio_{period}'] = ( dataframe['high'].rolling(period).max() / dataframe['low'].rolling(period).min() - 1 ) # === Volume-based features === for period in [5, 10, 20]: dataframe[f'volume_sma_{period}'] = dataframe['volume'].rolling(period).mean() dataframe[f'volume_ratio_{period}'] = dataframe['volume'] / dataframe[f'volume_sma_{period}'] # === Momentum features === for period in [7, 14, 21]: dataframe[f'rsi_{period}'] = ta.RSI(dataframe, timeperiod=period) dataframe[f'roc_{period}'] = ta.ROC(dataframe, timeperiod=period) # === Volatility features === for period in [10, 20, 30]: dataframe[f'atr_{period}'] = ta.ATR(dataframe, timeperiod=period) dataframe[f'volatility_{period}'] = dataframe[f'atr_{period}'] / dataframe['close'] # === Pattern features === dataframe['doji'] = ( abs(dataframe['close'] - dataframe['open']) / (dataframe['high'] - dataframe['low']) < 0.1 ).astype(int) dataframe['hammer'] = ( ((dataframe['high'] - dataframe['low']) > 3 * (dataframe['open'] - dataframe['close'])) & ((dataframe['close'] - dataframe['low']) / (0.001 + dataframe['high'] - dataframe['low']) > 0.6) & ((dataframe['open'] - dataframe['low']) / (0.001 + dataframe['high'] - dataframe['low']) > 0.6) ).astype(int) # === Support/Resistance features === dataframe['distance_to_support'] = ( (dataframe['close'] - dataframe['support_level']) / dataframe['close'] ).fillna(0) dataframe['distance_to_resistance'] = ( (dataframe['resistance_level'] - dataframe['close']) / dataframe['close'] ).fillna(0) # === Fibonacci features === dataframe['near_fib'] = dataframe['near_fib_level'].astype(int) # === Market regime features === dataframe['regime_trending'] = (dataframe['market_regime'] == 'trending').astype(int) dataframe['regime_ranging'] = (dataframe['market_regime'] == 'ranging').astype(int) dataframe['regime_volatile'] = (dataframe['market_regime'] == 'volatile').astype(int) # === Target variable for ML === # Predict future price movement future_periods = self.prediction_horizon.value dataframe['future_price'] = dataframe['close'].shift(-future_periods) dataframe['price_change_future'] = ( (dataframe['future_price'] - dataframe['close']) / dataframe['close'] ) # Classification target (up/down/sideways) dataframe['future_direction'] = 0 # sideways dataframe.loc[dataframe['price_change_future'] > 0.02, 'future_direction'] = 1 # up dataframe.loc[dataframe['price_change_future'] < -0.02, 'future_direction'] = -1 # down # Regression target (actual price change) dataframe['target_price_change'] = dataframe['price_change_future'] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ ML-enhanced entry signals with support/resistance confirmation """ # === ML Prediction Conditions === ml_bullish_prediction = ( (dataframe['do_predict'] == 1) & (dataframe['DI_values'] > self.prediction_confidence_threshold.value) & (dataframe['prediction'] > 0.02) # Predicting >2% price increase ) ml_bearish_prediction = ( (dataframe['do_predict'] == 1) & (dataframe['DI_values'] > self.prediction_confidence_threshold.value) & (dataframe['prediction'] < -0.02) # Predicting >2% price decrease ) # === Technical Confirmation === technical_bullish = ( (dataframe['trend_1h'] == 1) & (dataframe['trend_4h'] == 1) & (dataframe['ema_trend'] == 1) & (dataframe['trend_strength'] > self.trend_strength_threshold.value) & (dataframe['rsi_14'] > 30) & (dataframe['rsi_14'] < 70) & (dataframe['volume_ratio'] > 1.2) ) technical_bearish = ( (dataframe['trend_1h'] == 0) & (dataframe['trend_4h'] == 0) & (dataframe['ema_trend'] == 0) & (dataframe['trend_strength'] > self.trend_strength_threshold.value) & (dataframe['rsi_14'] > 30) & (dataframe['rsi_14'] < 70) & (dataframe['volume_ratio'] > 1.2) ) # === Support/Resistance Confirmation === sr_bullish = ( (dataframe['at_support']) | (dataframe['support_break'] == False) | (dataframe['resistance_break'] == True) | (dataframe['near_poc'] & (dataframe['close'] > dataframe['vwap'])) ) sr_bearish = ( (dataframe['at_resistance']) | (dataframe['resistance_break'] == False) | (dataframe['support_break'] == True) | (dataframe['near_poc'] & (dataframe['close'] < dataframe['vwap'])) ) # === Pattern Confirmation === pattern_bullish = ( (dataframe['bullish_engulfing']) | (dataframe['hammer'] == 1) | (dataframe['higher_low'] & dataframe['higher_high']) ) pattern_bearish = ( (dataframe['bearish_engulfing']) | (dataframe['lower_high'] & dataframe['lower_low']) ) # === Risk Management Filters === risk_acceptable = ( (dataframe['volatility'] < self.volatility_threshold.value) & (dataframe['market_regime'] != 'volatile') ) # === Entry Conditions === # Long entries dataframe.loc[ (ml_bullish_prediction) & (technical_bullish) & (sr_bullish) & (pattern_bullish) & (risk_acceptable) & (dataframe['near_fib_level'] | dataframe['at_support']), 'enter_long'] = 1 # Short entries dataframe.loc[ (ml_bearish_prediction) & (technical_bearish) & (sr_bearish) & (pattern_bearish) & (risk_acceptable) & (dataframe['near_fib_level'] | dataframe['at_resistance']), 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ ML-enhanced exit signals """ # === ML-based exits === ml_exit_long = ( (dataframe['do_predict'] == 1) & (dataframe['DI_values'] > 0.6) & (dataframe['prediction'] < -0.01) # Predicting price decline ) ml_exit_short = ( (dataframe['do_predict'] == 1) & (dataframe['DI_values'] > 0.6) & (dataframe['prediction'] > 0.01) # Predicting price increase ) # === Technical exits === technical_exit_long = ( (dataframe['rsi_14'] > 75) | (dataframe['at_resistance']) | (dataframe['bearish_engulfing']) | (dataframe['ema_trend'] == 0) | (dataframe['macd'] < dataframe['macdsignal']) ) technical_exit_short = ( (dataframe['rsi_14'] < 25) | (dataframe['at_support']) | (dataframe['bullish_engulfing']) | (dataframe['ema_trend'] == 1) | (dataframe['macd'] > dataframe['macdsignal']) ) # === Support/Resistance exits === sr_exit_long = ( (dataframe['resistance_break'] == False) | (dataframe['support_break'] == True) ) sr_exit_short = ( (dataframe['support_break'] == False) | (dataframe['resistance_break'] == True) ) # === Exit Conditions === dataframe.loc[ (ml_exit_long) | (technical_exit_long) | (sr_exit_long), 'exit_long'] = 1 dataframe.loc[ (ml_exit_short) | (technical_exit_short) | (sr_exit_short), 'exit_short'] = 1 return dataframe def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: """ Dynamic leverage based on ML confidence and volatility """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Base leverage on ML confidence base_leverage = 5.0 if 'DI_values' in last_candle and not pd.isna(last_candle['DI_values']): confidence = last_candle['DI_values'] # Higher confidence = higher leverage leverage_multiplier = 1 + (confidence - 0.5) * 2 base_leverage *= leverage_multiplier # Adjust for volatility if 'volatility' in last_candle and not pd.isna(last_candle['volatility']): volatility = last_candle['volatility'] if volatility > 0.03: # High volatility base_leverage *= 0.7 elif volatility < 0.01: # Low volatility base_leverage *= 1.3 return min(base_leverage, max_leverage, 10.0) # Max 10x leverage def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Dynamic stoploss based on support/resistance levels """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Default stoploss stop_loss = self.stoploss # Adjust based on support/resistance if not trade.is_short: # Long position if 'support_level' in last_candle and not pd.isna(last_candle['support_level']): support_distance = (current_rate - last_candle['support_level']) / current_rate # Set stop just below support stop_loss = max(self.stoploss, -support_distance * 1.1) else: # Short position if 'resistance_level' in last_candle and not pd.isna(last_candle['resistance_level']): resistance_distance = (last_candle['resistance_level'] - current_rate) / current_rate # Set stop just above resistance stop_loss = max(self.stoploss, -resistance_distance * 1.1) # Trailing stop based on ML predictions if current_profit > self.trailing_stop_activation.value: if 'prediction' in last_candle and not pd.isna(last_candle['prediction']): # If ML predicts continued favorable movement, trail closer if (not trade.is_short and last_candle['prediction'] > 0) or \ (trade.is_short and last_candle['prediction'] < 0): trail_distance = self.trailing_stop_activation.value * 0.5 else: trail_distance = self.trailing_stop_activation.value stop_loss = max(stop_loss, trail_distance - current_profit) return stop_loss def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: """ Final confirmation before entering trade """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Ensure ML model is providing predictions if 'do_predict' not in last_candle or last_candle['do_predict'] != 1: return False # Ensure confidence is above threshold if 'DI_values' not in last_candle or last_candle['DI_values'] < self.prediction_confidence_threshold.value: return False # Check risk/reward ratio if side == "long": if 'resistance_level' in last_candle and not pd.isna(last_candle['resistance_level']): potential_profit = (last_candle['resistance_level'] - rate) / rate potential_loss = abs(self.stoploss) if potential_profit / potential_loss < self.risk_reward_ratio.value: return False return True def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]: """ Custom exit logic based on ML predictions and S/R levels """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # ML-based exit if 'prediction' in last_candle and 'DI_values' in last_candle: if last_candle['DI_values'] > 0.7: # High confidence prediction if not trade.is_short and last_candle['prediction'] < -0.015: # Predicting 1.5% decline return "ml_reversal_prediction" elif trade.is_short and last_candle['prediction'] > 0.015: # Predicting 1.5% increase return "ml_reversal_prediction" # Profit taking at resistance/support if current_profit > 0.03: # 3% profit if not trade.is_short and last_candle.get('at_resistance', False): return "resistance_profit_taking" elif trade.is_short and last_candle.get('at_support', False): return "support_profit_taking" # Risk management exit if last_candle.get('market_regime') == 'volatile' and current_profit < -0.02: return "volatile_market_exit" return None