# source: https://raw.githubusercontent.com/markelayan/hhec-dois-bull-manager/09fa8d07121e68880686b2f9b8d4201bca0dd669/1strategies1/ClaudeSmartMoneyEnhanced.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 logger = logging.getLogger(__name__) class Github_markelayan_hhec_dois_bull_manager__ClaudeSmartMoneyEnhanced__20250810_092830(IStrategy): """ Enhanced Professional Smart Money Concepts Strategy with AI Integration New Features: - Advanced order block detection with ML-ready features - Enhanced fair value gap analysis with timeframe correlation - Liquidity sweep patterns with volume confirmation - Market structure break analysis with momentum divergence - FreqAI integration support - Enhanced hyperopt parameters - Advanced risk management with dynamic position sizing - Multi-timeframe smart money flow analysis - Institutional footprint detection - Enhanced volume profile analysis - Smart money accumulation/distribution patterns - Wyckoff methodology integration - Options flow influence detection - Real-time sentiment integration """ INTERFACE_VERSION = 3 # Enhanced ROI with smart money targets minimal_roi = { "0": 0.25, # 25% at any time (smart money targets) "30": 0.15, # 15% after 30 minutes "60": 0.10, # 10% after 1 hour "120": 0.05, # 5% after 2 hours "240": 0.025, # 2.5% after 4 hours "480": 0.01 # 1% after 8 hours } # Dynamic stoploss with smart money levels stoploss = -0.10 # 10% base stoploss # Primary timeframe optimized for smart money timeframe = '15m' # Enhanced informative timeframes inf_5m = '5m' inf_1h = '1h' inf_4h = '4h' inf_1d = '1d' inf_1w = '1w' # Performance optimizations process_only_new_candles = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # Enhanced startup for proper smart money analysis startup_candle_count: int = 300 # === Enhanced Strategy Parameters - Hyperopt Enabled === # === Advanced Volume Analysis === volume_lookback = IntParameter(15, 50, default=30, space='buy', optimize=True) volume_threshold = DecimalParameter(1.5, 4.0, default=2.5, space='buy', optimize=True) volume_ma_period = IntParameter(15, 60, default=30, space='buy', optimize=True) volume_spike_threshold = DecimalParameter(3.0, 8.0, default=5.0, space='buy', optimize=True) institutional_volume_threshold = DecimalParameter(5.0, 15.0, default=10.0, space='buy', optimize=True) # === Enhanced Smart Money Structure === structure_lookback = IntParameter(30, 100, default=50, space='buy', optimize=True) liquidity_threshold = DecimalParameter(0.8, 3.0, default=1.5, space='buy', optimize=True) order_block_strength = IntParameter(5, 20, default=10, space='buy', optimize=True) order_block_validation_period = IntParameter(10, 50, default=25, space='buy', optimize=True) # === Advanced Fair Value Gap === fvg_min_size = DecimalParameter(0.2, 2.0, default=0.5, space='buy', optimize=True) fvg_lookback = IntParameter(10, 40, default=20, space='buy', optimize=True) fvg_retest_threshold = DecimalParameter(0.1, 1.0, default=0.3, space='buy', optimize=True) fvg_invalidation_threshold = DecimalParameter(0.5, 2.0, default=1.0, space='buy', optimize=True) # === Enhanced Trend Analysis === ema_fast = IntParameter(8, 25, default=12, space='buy', optimize=True) ema_medium = IntParameter(21, 55, default=34, space='buy', optimize=True) ema_slow = IntParameter(50, 200, default=100, space='buy', optimize=True) ema_trend = IntParameter(100, 400, default=200, space='buy', optimize=True) # === Multi-Timeframe RSI === rsi_period = IntParameter(10, 25, default=14, space='buy', optimize=True) rsi_overbought = IntParameter(65, 85, default=75, space='buy', optimize=True) rsi_oversold = IntParameter(15, 35, default=25, space='buy', optimize=True) rsi_divergence_lookback = IntParameter(10, 30, default=20, space='buy', optimize=True) # === Wyckoff Analysis === wyckoff_accumulation_threshold = DecimalParameter(0.5, 2.0, default=1.0, space='buy', optimize=True) wyckoff_distribution_threshold = DecimalParameter(0.5, 2.0, default=1.0, space='sell', optimize=True) wyckoff_volume_confirmation = DecimalParameter(1.5, 4.0, default=2.5, space='buy', optimize=True) # === Market Structure Breaks === msb_confirmation_candles = IntParameter(2, 8, default=3, space='buy', optimize=True) msb_volume_confirmation = DecimalParameter(1.2, 3.0, default=2.0, space='buy', optimize=True) msb_strength_threshold = DecimalParameter(0.3, 1.5, default=0.8, space='buy', optimize=True) # === Liquidity Analysis === liquidity_cluster_size = IntParameter(3, 10, default=5, space='buy', optimize=True) liquidity_sweep_confirmation = IntParameter(1, 5, default=2, space='buy', optimize=True) equal_highs_lows_threshold = DecimalParameter(0.1, 0.8, default=0.3, space='buy', optimize=True) # === Advanced Exit Parameters === exit_rsi_high = IntParameter(70, 95, default=80, space='sell', optimize=True) exit_rsi_low = IntParameter(5, 30, default=20, space='sell', optimize=True) take_profit_ob_retest = BooleanParameter(default=True, space='sell', optimize=True) exit_on_fvg_fill = BooleanParameter(default=True, space='sell', optimize=True) # === Risk Management Enhancement === use_dynamic_position_sizing = BooleanParameter(default=True, space='buy', optimize=True) risk_per_trade = DecimalParameter(0.5, 3.0, default=1.5, space='buy', optimize=True) max_correlation_exposure = DecimalParameter(0.3, 0.8, default=0.6, space='buy', optimize=True) # === FreqAI Integration === use_freqai_signals = BooleanParameter(default=False, space='buy', optimize=False) freqai_signal_weight = DecimalParameter(0.2, 0.8, default=0.5, space='buy', optimize=True) freqai_confidence_threshold = DecimalParameter(0.6, 0.9, default=0.75, space='buy', optimize=True) # === Signal Quality Filters === min_signal_quality = DecimalParameter(0.4, 0.8, default=0.6, space='buy', optimize=True) confluence_requirement = IntParameter(2, 6, default=3, space='buy', optimize=True) # === Risk Management === max_open_trades = 2 # Conservative for smart money position_adjustment_enable = True def informative_pairs(self): """Enhanced informative pairs for comprehensive smart money analysis""" pairs = self.dp.current_whitelist() informative_pairs = [(pair, self.inf_5m) for pair in pairs] informative_pairs += [(pair, self.inf_1h) for pair in pairs] informative_pairs += [(pair, self.inf_4h) for pair in pairs] informative_pairs += [(pair, self.inf_1d) for pair in pairs] informative_pairs += [(pair, self.inf_1w) for pair in pairs] # Add major correlation pairs for risk management major_pairs = ['BTC/USDT', 'ETH/USDT'] for major_pair in major_pairs: if major_pair not in [pair[0] for pair in informative_pairs]: informative_pairs += [(major_pair, self.inf_1h)] informative_pairs += [(major_pair, self.inf_4h)] return informative_pairs def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Enhanced indicators with comprehensive smart money analysis""" # === Basic Price Action === 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 dataframe['typical_price'] = dataframe['hlc3'] # === Enhanced Moving Averages === dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast.value) dataframe['ema_medium'] = ta.EMA(dataframe, timeperiod=self.ema_medium.value) dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_slow.value) dataframe['ema_trend'] = ta.EMA(dataframe, timeperiod=self.ema_trend.value) # EMA relationships for trend analysis dataframe['emas_aligned_bullish'] = ( (dataframe['ema_fast'] > dataframe['ema_medium']) & (dataframe['ema_medium'] > dataframe['ema_slow']) & (dataframe['ema_slow'] > dataframe['ema_trend']) ) # === Enhanced RSI with Divergence === dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value) dataframe['rsi_ma'] = ta.SMA(dataframe['rsi'], timeperiod=5) dataframe = self.calculate_rsi_divergence(dataframe) # === Enhanced Volume Analysis === dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=self.volume_ma_period.value) dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma'] dataframe['high_volume'] = dataframe['volume_ratio'] > self.volume_threshold.value dataframe['volume_spike'] = dataframe['volume_ratio'] > self.volume_spike_threshold.value dataframe['institutional_volume'] = dataframe['volume_ratio'] > self.institutional_volume_threshold.value # Advanced volume indicators dataframe['volume_momentum'] = dataframe['volume'].rolling(5).mean() / dataframe['volume'].rolling(20).mean() dataframe['volume_acceleration'] = dataframe['volume_momentum'] - dataframe['volume_momentum'].shift(1) # Volume-weighted indicators dataframe['vwap'] = qtpylib.vwap(dataframe) dataframe['vwap_distance'] = (dataframe['close'] - dataframe['vwap']) / dataframe['vwap'] # Money Flow Index dataframe['mfi'] = ta.MFI(dataframe, timeperiod=14) # On Balance Volume dataframe['obv'] = ta.OBV(dataframe) dataframe['obv_ma'] = ta.SMA(dataframe['obv'], timeperiod=20) # === Enhanced ATR and Volatility === dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['atr_percent'] = (dataframe['atr'] / dataframe['close']) * 100 dataframe['volatility_regime'] = np.where( dataframe['atr_percent'] > dataframe['atr_percent'].rolling(50).quantile(0.8), 'high', 'normal' ) # === Advanced Smart Money Concepts === dataframe = self.calculate_enhanced_order_blocks(dataframe) dataframe = self.calculate_enhanced_fair_value_gaps(dataframe) dataframe = self.calculate_enhanced_market_structure(dataframe) dataframe = self.calculate_enhanced_liquidity_levels(dataframe) dataframe = self.calculate_wyckoff_patterns(dataframe) dataframe = self.calculate_institutional_footprint(dataframe) # === Higher Timeframe Analysis === dataframe = self.populate_higher_timeframe_indicators(dataframe, metadata) # === Signal Quality Scoring === dataframe = self.calculate_signal_quality_scores(dataframe) return dataframe def calculate_rsi_divergence(self, dataframe: DataFrame) -> DataFrame: """Calculate RSI divergence patterns""" lookback = self.rsi_divergence_lookback.value # Find swing highs and lows dataframe['swing_high'] = ( (dataframe['high'] == dataframe['high'].rolling(5, center=True).max()) ) dataframe['swing_low'] = ( (dataframe['low'] == dataframe['low'].rolling(5, center=True).min()) ) # Initialize divergence columns dataframe['bullish_divergence'] = False dataframe['bearish_divergence'] = False # Calculate divergences for i in range(lookback, len(dataframe)): # Bearish divergence: price higher high, RSI lower high if dataframe.iloc[i]['swing_high']: recent_highs = dataframe.iloc[i-lookback:i][dataframe.iloc[i-lookback:i]['swing_high']] if len(recent_highs) > 0: prev_price_high = recent_highs['high'].iloc[-1] prev_rsi_high = recent_highs['rsi'].iloc[-1] if (dataframe.iloc[i]['high'] > prev_price_high and dataframe.iloc[i]['rsi'] < prev_rsi_high): dataframe.iloc[i, dataframe.columns.get_loc('bearish_divergence')] = True # Bullish divergence: price lower low, RSI higher low if dataframe.iloc[i]['swing_low']: recent_lows = dataframe.iloc[i-lookback:i][dataframe.iloc[i-lookback:i]['swing_low']] if len(recent_lows) > 0: prev_price_low = recent_lows['low'].iloc[-1] prev_rsi_low = recent_lows['rsi'].iloc[-1] if (dataframe.iloc[i]['low'] < prev_price_low and dataframe.iloc[i]['rsi'] > prev_rsi_low): dataframe.iloc[i, dataframe.columns.get_loc('bullish_divergence')] = True return dataframe def calculate_enhanced_order_blocks(self, dataframe: DataFrame) -> DataFrame: """Enhanced order block detection with validation and strength scoring""" # Initialize order block columns dataframe['bull_ob_high'] = 0.0 dataframe['bull_ob_low'] = 0.0 dataframe['bull_ob_active'] = False dataframe['bull_ob_strength'] = 0.0 dataframe['bull_ob_volume_confirmation'] = False dataframe['bear_ob_high'] = 0.0 dataframe['bear_ob_low'] = 0.0 dataframe['bear_ob_active'] = False dataframe['bear_ob_strength'] = 0.0 dataframe['bear_ob_volume_confirmation'] = False strength = self.order_block_strength.value validation_period = self.order_block_validation_period.value for i in range(strength, len(dataframe)): # Enhanced bullish order block detection if (dataframe.iloc[i]['close'] > dataframe.iloc[i-1]['high'] and # Break of structure dataframe.iloc[i]['volume'] > dataframe.iloc[i-1]['volume'] * 1.5 and # Volume confirmation dataframe.iloc[i]['close'] > dataframe.iloc[i]['open']): # Bullish candle # Find the last bearish candle before the break ob_found = False for j in range(i-1, max(0, i-strength), -1): if (dataframe.iloc[j]['close'] < dataframe.iloc[j]['open'] and # Bearish candle dataframe.iloc[j]['volume'] > dataframe.iloc[j-5:j]['volume'].mean()): # Above average volume # Calculate order block strength volume_strength = dataframe.iloc[i]['volume'] / dataframe.iloc[j-10:j]['volume'].mean() price_strength = (dataframe.iloc[i]['close'] - dataframe.iloc[j]['high']) / dataframe.iloc[j]['high'] ob_strength = (volume_strength * 0.6) + (price_strength * 100 * 0.4) dataframe.iloc[i, dataframe.columns.get_loc('bull_ob_high')] = dataframe.iloc[j]['high'] dataframe.iloc[i, dataframe.columns.get_loc('bull_ob_low')] = dataframe.iloc[j]['low'] dataframe.iloc[i, dataframe.columns.get_loc('bull_ob_active')] = True dataframe.iloc[i, dataframe.columns.get_loc('bull_ob_strength')] = ob_strength dataframe.iloc[i, dataframe.columns.get_loc('bull_ob_volume_confirmation')] = volume_strength > 2.0 ob_found = True break # Validate order block over validation period if ob_found: for k in range(i+1, min(len(dataframe), i+validation_period)): if dataframe.iloc[k]['low'] < dataframe.iloc[i]['bull_ob_low']: # Order block invalidated dataframe.iloc[i, dataframe.columns.get_loc('bull_ob_active')] = False break # Enhanced bearish order block detection if (dataframe.iloc[i]['close'] < dataframe.iloc[i-1]['low'] and # Break of structure dataframe.iloc[i]['volume'] > dataframe.iloc[i-1]['volume'] * 1.5 and # Volume confirmation dataframe.iloc[i]['close'] < dataframe.iloc[i]['open']): # Bearish candle # Find the last bullish candle before the break ob_found = False for j in range(i-1, max(0, i-strength), -1): if (dataframe.iloc[j]['close'] > dataframe.iloc[j]['open'] and # Bullish candle dataframe.iloc[j]['volume'] > dataframe.iloc[j-5:j]['volume'].mean()): # Above average volume # Calculate order block strength volume_strength = dataframe.iloc[i]['volume'] / dataframe.iloc[j-10:j]['volume'].mean() price_strength = (dataframe.iloc[j]['low'] - dataframe.iloc[i]['close']) / dataframe.iloc[j]['low'] ob_strength = (volume_strength * 0.6) + (price_strength * 100 * 0.4) dataframe.iloc[i, dataframe.columns.get_loc('bear_ob_high')] = dataframe.iloc[j]['high'] dataframe.iloc[i, dataframe.columns.get_loc('bear_ob_low')] = dataframe.iloc[j]['low'] dataframe.iloc[i, dataframe.columns.get_loc('bear_ob_active')] = True dataframe.iloc[i, dataframe.columns.get_loc('bear_ob_strength')] = ob_strength dataframe.iloc[i, dataframe.columns.get_loc('bear_ob_volume_confirmation')] = volume_strength > 2.0 ob_found = True break # Validate order block over validation period if ob_found: for k in range(i+1, min(len(dataframe), i+validation_period)): if dataframe.iloc[k]['high'] > dataframe.iloc[i]['bear_ob_high']: # Order block invalidated dataframe.iloc[i, dataframe.columns.get_loc('bear_ob_active')] = False break return dataframe def calculate_enhanced_fair_value_gaps(self, dataframe: DataFrame) -> DataFrame: """Enhanced Fair Value Gap detection with retest and invalidation logic""" # Initialize FVG columns dataframe['fvg_bullish'] = False dataframe['fvg_bearish'] = False dataframe['fvg_bull_high'] = 0.0 dataframe['fvg_bull_low'] = 0.0 dataframe['fvg_bear_high'] = 0.0 dataframe['fvg_bear_low'] = 0.0 dataframe['fvg_bull_strength'] = 0.0 dataframe['fvg_bear_strength'] = 0.0 dataframe['fvg_bull_retested'] = False dataframe['fvg_bear_retested'] = False min_size = self.fvg_min_size.value / 100 for i in range(2, len(dataframe)): # Enhanced Bullish FVG detection gap_size = dataframe.iloc[i]['low'] - dataframe.iloc[i-2]['high'] gap_percent = gap_size / dataframe.iloc[i]['close'] if (dataframe.iloc[i-2]['high'] < dataframe.iloc[i]['low'] and dataframe.iloc[i-1]['high'] < dataframe.iloc[i]['low'] and gap_percent > min_size): # Calculate FVG strength volume_strength = dataframe.iloc[i]['volume'] / dataframe.iloc[i-10:i]['volume'].mean() size_strength = gap_percent / min_size fvg_strength = (volume_strength * 0.7) + (size_strength * 0.3) dataframe.iloc[i, dataframe.columns.get_loc('fvg_bullish')] = True dataframe.iloc[i, dataframe.columns.get_loc('fvg_bull_low')] = dataframe.iloc[i-2]['high'] dataframe.iloc[i, dataframe.columns.get_loc('fvg_bull_high')] = dataframe.iloc[i]['low'] dataframe.iloc[i, dataframe.columns.get_loc('fvg_bull_strength')] = fvg_strength # Check for retest in subsequent candles for j in range(i+1, min(len(dataframe), i+self.fvg_lookback.value)): if (dataframe.iloc[j]['low'] <= dataframe.iloc[i]['fvg_bull_high'] and dataframe.iloc[j]['low'] >= dataframe.iloc[i]['fvg_bull_low']): dataframe.iloc[i, dataframe.columns.get_loc('fvg_bull_retested')] = True break elif dataframe.iloc[j]['low'] < dataframe.iloc[i]['fvg_bull_low']: # FVG filled/invalidated break # Enhanced Bearish FVG detection gap_size = dataframe.iloc[i-2]['low'] - dataframe.iloc[i]['high'] gap_percent = gap_size / dataframe.iloc[i]['close'] if (dataframe.iloc[i-2]['low'] > dataframe.iloc[i]['high'] and dataframe.iloc[i-1]['low'] > dataframe.iloc[i]['high'] and gap_percent > min_size): # Calculate FVG strength volume_strength = dataframe.iloc[i]['volume'] / dataframe.iloc[i-10:i]['volume'].mean() size_strength = gap_percent / min_size fvg_strength = (volume_strength * 0.7) + (size_strength * 0.3) dataframe.iloc[i, dataframe.columns.get_loc('fvg_bearish')] = True dataframe.iloc[i, dataframe.columns.get_loc('fvg_bear_high')] = dataframe.iloc[i-2]['low'] dataframe.iloc[i, dataframe.columns.get_loc('fvg_bear_low')] = dataframe.iloc[i]['high'] dataframe.iloc[i, dataframe.columns.get_loc('fvg_bear_strength')] = fvg_strength # Check for retest in subsequent candles for j in range(i+1, min(len(dataframe), i+self.fvg_lookback.value)): if (dataframe.iloc[j]['high'] >= dataframe.iloc[i]['fvg_bear_low'] and dataframe.iloc[j]['high'] <= dataframe.iloc[i]['fvg_bear_high']): dataframe.iloc[i, dataframe.columns.get_loc('fvg_bear_retested')] = True break elif dataframe.iloc[j]['high'] > dataframe.iloc[i]['fvg_bear_high']: # FVG filled/invalidated break return dataframe def calculate_enhanced_market_structure(self, dataframe: DataFrame) -> DataFrame: """Enhanced market structure analysis with confirmation""" # Calculate swing highs and lows with more precision dataframe['swing_high'] = ( (dataframe['high'] > dataframe['high'].shift(2)) & (dataframe['high'] > dataframe['high'].shift(1)) & (dataframe['high'] > dataframe['high'].shift(-1)) & (dataframe['high'] > dataframe['high'].shift(-2)) ) dataframe['swing_low'] = ( (dataframe['low'] < dataframe['low'].shift(2)) & (dataframe['low'] < dataframe['low'].shift(1)) & (dataframe['low'] < dataframe['low'].shift(-1)) & (dataframe['low'] < dataframe['low'].shift(-2)) ) # Enhanced market structure breaks dataframe['bullish_bos'] = False dataframe['bearish_bos'] = False dataframe['bos_strength'] = 0.0 dataframe['bos_volume_confirmation'] = False lookback = self.structure_lookback.value confirmation_candles = self.msb_confirmation_candles.value volume_confirmation = self.msb_volume_confirmation.value for i in range(lookback, len(dataframe) - confirmation_candles): # Enhanced bullish break of structure recent_highs = dataframe.iloc[i-lookback:i][dataframe.iloc[i-lookback:i]['swing_high']]['high'] if len(recent_highs) > 0: highest_high = recent_highs.max() # Check for break with confirmation if (dataframe.iloc[i]['close'] > highest_high and dataframe.iloc[i]['volume'] > dataframe.iloc[i-5:i]['volume'].mean() * volume_confirmation): # Confirm with subsequent candles confirmed = True for j in range(i+1, i+confirmation_candles+1): if j < len(dataframe) and dataframe.iloc[j]['close'] < highest_high: confirmed = False break if confirmed: # Calculate BOS strength price_strength = (dataframe.iloc[i]['close'] - highest_high) / highest_high volume_strength = dataframe.iloc[i]['volume'] / dataframe.iloc[i-10:i]['volume'].mean() bos_strength = (price_strength * 100 * 0.6) + (volume_strength * 0.4) dataframe.iloc[i, dataframe.columns.get_loc('bullish_bos')] = True dataframe.iloc[i, dataframe.columns.get_loc('bos_strength')] = bos_strength dataframe.iloc[i, dataframe.columns.get_loc('bos_volume_confirmation')] = volume_strength > 2.0 # Enhanced bearish break of structure recent_lows = dataframe.iloc[i-lookback:i][dataframe.iloc[i-lookback:i]['swing_low']]['low'] if len(recent_lows) > 0: lowest_low = recent_lows.min() # Check for break with confirmation if (dataframe.iloc[i]['close'] < lowest_low and dataframe.iloc[i]['volume'] > dataframe.iloc[i-5:i]['volume'].mean() * volume_confirmation): # Confirm with subsequent candles confirmed = True for j in range(i+1, i+confirmation_candles+1): if j < len(dataframe) and dataframe.iloc[j]['close'] > lowest_low: confirmed = False break if confirmed: # Calculate BOS strength price_strength = (lowest_low - dataframe.iloc[i]['close']) / lowest_low volume_strength = dataframe.iloc[i]['volume'] / dataframe.iloc[i-10:i]['volume'].mean() bos_strength = (price_strength * 100 * 0.6) + (volume_strength * 0.4) dataframe.iloc[i, dataframe.columns.get_loc('bearish_bos')] = True dataframe.iloc[i, dataframe.columns.get_loc('bos_strength')] = bos_strength dataframe.iloc[i, dataframe.columns.get_loc('bos_volume_confirmation')] = volume_strength > 2.0 return dataframe def calculate_enhanced_liquidity_levels(self, dataframe: DataFrame) -> DataFrame: """Enhanced liquidity analysis with cluster detection""" # Initialize liquidity columns dataframe['liquidity_high'] = 0.0 dataframe['liquidity_low'] = 0.0 dataframe['sweep_high'] = False dataframe['sweep_low'] = False dataframe['liquidity_cluster_strength'] = 0.0 dataframe['equal_highs'] = False dataframe['equal_lows'] = False cluster_size = self.liquidity_cluster_size.value threshold = self.equal_highs_lows_threshold.value / 100 sweep_confirmation = self.liquidity_sweep_confirmation.value for i in range(30, len(dataframe)): # Enhanced equal highs detection recent_highs = dataframe.iloc[i-30:i]['high'] max_high = recent_highs.max() # Find equal highs within threshold equal_highs_mask = abs(recent_highs - max_high) <= (max_high * threshold) equal_highs_count = equal_highs_mask.sum() if equal_highs_count >= cluster_size: cluster_strength = equal_highs_count / 30 # Normalize by lookback period dataframe.iloc[i, dataframe.columns.get_loc('liquidity_high')] = max_high dataframe.iloc[i, dataframe.columns.get_loc('equal_highs')] = True dataframe.iloc[i, dataframe.columns.get_loc('liquidity_cluster_strength')] = cluster_strength # Check for liquidity sweep if (dataframe.iloc[i]['high'] > max_high and dataframe.iloc[i]['close'] < max_high * 0.998): # Wick above with close below # Confirm sweep with volume if dataframe.iloc[i]['volume'] > dataframe.iloc[i-5:i]['volume'].mean() * 1.5: dataframe.iloc[i, dataframe.columns.get_loc('sweep_high')] = True # Enhanced equal lows detection recent_lows = dataframe.iloc[i-30:i]['low'] min_low = recent_lows.min() # Find equal lows within threshold equal_lows_mask = abs(recent_lows - min_low) <= (min_low * threshold) equal_lows_count = equal_lows_mask.sum() if equal_lows_count >= cluster_size: cluster_strength = equal_lows_count / 30 dataframe.iloc[i, dataframe.columns.get_loc('liquidity_low')] = min_low dataframe.iloc[i, dataframe.columns.get_loc('equal_lows')] = True dataframe.iloc[i, dataframe.columns.get_loc('liquidity_cluster_strength')] = cluster_strength # Check for liquidity sweep if (dataframe.iloc[i]['low'] < min_low and dataframe.iloc[i]['close'] > min_low * 1.002): # Wick below with close above # Confirm sweep with volume if dataframe.iloc[i]['volume'] > dataframe.iloc[i-5:i]['volume'].mean() * 1.5: dataframe.iloc[i, dataframe.columns.get_loc('sweep_low')] = True return dataframe def calculate_wyckoff_patterns(self, dataframe: DataFrame) -> DataFrame: """Calculate Wyckoff accumulation and distribution patterns""" # Initialize Wyckoff columns dataframe['wyckoff_accumulation'] = False dataframe['wyckoff_distribution'] = False dataframe['wyckoff_spring'] = False dataframe['wyckoff_upthrust'] = False dataframe['wyckoff_strength'] = 0.0 accumulation_threshold = self.wyckoff_accumulation_threshold.value distribution_threshold = self.wyckoff_distribution_threshold.value volume_confirmation = self.wyckoff_volume_confirmation.value # Calculate price and volume relationships dataframe['price_range'] = dataframe['high'] - dataframe['low'] dataframe['price_range_ma'] = dataframe['price_range'].rolling(20).mean() dataframe['volume_ma'] = dataframe['volume'].rolling(20).mean() for i in range(50, len(dataframe)): # Wyckoff Accumulation (Spring pattern) if (dataframe.iloc[i]['low'] < dataframe.iloc[i-20:i]['low'].min() and # New low dataframe.iloc[i]['close'] > dataframe.iloc[i]['low'] + (dataframe.iloc[i]['high'] - dataframe.iloc[i]['low']) * 0.7 and # Strong close dataframe.iloc[i]['volume'] > dataframe.iloc[i]['volume_ma'] * volume_confirmation): # High volume # Check for subsequent strength strength_confirmed = False for j in range(i+1, min(len(dataframe), i+5)): if dataframe.iloc[j]['close'] > dataframe.iloc[i]['high']: strength_confirmed = True break if strength_confirmed: wyckoff_strength = (dataframe.iloc[i]['volume'] / dataframe.iloc[i]['volume_ma']) * \ (dataframe.iloc[i]['close'] - dataframe.iloc[i]['low']) / dataframe.iloc[i]['price_range'] dataframe.iloc[i, dataframe.columns.get_loc('wyckoff_accumulation')] = True dataframe.iloc[i, dataframe.columns.get_loc('wyckoff_spring')] = True dataframe.iloc[i, dataframe.columns.get_loc('wyckoff_strength')] = wyckoff_strength # Wyckoff Distribution (Upthrust pattern) if (dataframe.iloc[i]['high'] > dataframe.iloc[i-20:i]['high'].max() and # New high dataframe.iloc[i]['close'] < dataframe.iloc[i]['high'] - (dataframe.iloc[i]['high'] - dataframe.iloc[i]['low']) * 0.7 and # Weak close dataframe.iloc[i]['volume'] > dataframe.iloc[i]['volume_ma'] * volume_confirmation): # High volume # Check for subsequent weakness weakness_confirmed = False for j in range(i+1, min(len(dataframe), i+5)): if dataframe.iloc[j]['close'] < dataframe.iloc[i]['low']: weakness_confirmed = True break if weakness_confirmed: wyckoff_strength = (dataframe.iloc[i]['volume'] / dataframe.iloc[i]['volume_ma']) * \ (dataframe.iloc[i]['high'] - dataframe.iloc[i]['close']) / dataframe.iloc[i]['price_range'] dataframe.iloc[i, dataframe.columns.get_loc('wyckoff_distribution')] = True dataframe.iloc[i, dataframe.columns.get_loc('wyckoff_upthrust')] = True dataframe.iloc[i, dataframe.columns.get_loc('wyckoff_strength')] = wyckoff_strength return dataframe def calculate_institutional_footprint(self, dataframe: DataFrame) -> DataFrame: """Calculate institutional trading footprint indicators""" # Initialize institutional footprint columns dataframe['institutional_buying'] = False dataframe['institutional_selling'] = False dataframe['smart_money_accumulation'] = 0.0 dataframe['smart_money_distribution'] = 0.0 dataframe['institutional_confidence'] = 0.0 # Calculate institutional volume patterns dataframe['large_volume'] = dataframe['volume'] > dataframe['volume'].rolling(50).quantile(0.9) dataframe['volume_price_trend'] = (dataframe['close'] - dataframe['open']) * dataframe['volume'] dataframe['vpt_ma'] = dataframe['volume_price_trend'].rolling(20).mean() # Money flow calculations dataframe['typical_price'] = (dataframe['high'] + dataframe['low'] + dataframe['close']) / 3 dataframe['money_flow'] = dataframe['typical_price'] * dataframe['volume'] dataframe['money_flow_ratio'] = dataframe['money_flow'] / dataframe['money_flow'].rolling(20).mean() for i in range(20, len(dataframe)): # Institutional buying detection if (dataframe.iloc[i]['large_volume'] and dataframe.iloc[i]['close'] > dataframe.iloc[i]['open'] and dataframe.iloc[i]['volume_price_trend'] > dataframe.iloc[i]['vpt_ma'] * 2 and dataframe.iloc[i]['money_flow_ratio'] > 1.5): # Calculate institutional confidence volume_strength = dataframe.iloc[i]['volume'] / dataframe.iloc[i-20:i]['volume'].mean() price_strength = (dataframe.iloc[i]['close'] - dataframe.iloc[i]['open']) / dataframe.iloc[i]['open'] institutional_confidence = (volume_strength * 0.6) + (price_strength * 100 * 0.4) dataframe.iloc[i, dataframe.columns.get_loc('institutional_buying')] = True dataframe.iloc[i, dataframe.columns.get_loc('institutional_confidence')] = institutional_confidence # Accumulation scoring accumulation_score = min(institutional_confidence / 5, 1.0) # Normalize to 0-1 dataframe.iloc[i, dataframe.columns.get_loc('smart_money_accumulation')] = accumulation_score # Institutional selling detection elif (dataframe.iloc[i]['large_volume'] and dataframe.iloc[i]['close'] < dataframe.iloc[i]['open'] and dataframe.iloc[i]['volume_price_trend'] < dataframe.iloc[i]['vpt_ma'] * -2 and dataframe.iloc[i]['money_flow_ratio'] > 1.5): # Calculate institutional confidence (for selling) volume_strength = dataframe.iloc[i]['volume'] / dataframe.iloc[i-20:i]['volume'].mean() price_strength = (dataframe.iloc[i]['open'] - dataframe.iloc[i]['close']) / dataframe.iloc[i]['open'] institutional_confidence = (volume_strength * 0.6) + (price_strength * 100 * 0.4) dataframe.iloc[i, dataframe.columns.get_loc('institutional_selling')] = True dataframe.iloc[i, dataframe.columns.get_loc('institutional_confidence')] = institutional_confidence # Distribution scoring distribution_score = min(institutional_confidence / 5, 1.0) # Normalize to 0-1 dataframe.iloc[i, dataframe.columns.get_loc('smart_money_distribution')] = distribution_score return dataframe def populate_higher_timeframe_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Enhanced higher timeframe analysis with smart money context""" # 5-minute context inf_5m = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_5m) inf_5m['ema_trend_5m'] = ta.EMA(inf_5m, timeperiod=50) inf_5m['rsi_5m'] = ta.RSI(inf_5m, timeperiod=14) inf_5m['volume_5m'] = inf_5m['volume'].rolling(20).mean() inf_5m['trend_5m'] = (inf_5m['close'] > inf_5m['ema_trend_5m']).astype(int) inf_5m['momentum_5m'] = inf_5m['close'].pct_change(5) dataframe = merge_informative_pair(dataframe, inf_5m, self.timeframe, self.inf_5m, ffill=True) # 1-hour context inf_1h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1h) inf_1h['ema_trend_1h'] = ta.EMA(inf_1h, timeperiod=50) inf_1h['rsi_1h'] = ta.RSI(inf_1h, timeperiod=14) inf_1h['adx_1h'] = ta.ADX(inf_1h, timeperiod=14) inf_1h['trend_1h'] = (inf_1h['close'] > inf_1h['ema_trend_1h']).astype(int) inf_1h['volume_strength_1h'] = inf_1h['volume'] / inf_1h['volume'].rolling(30).mean() dataframe = merge_informative_pair(dataframe, inf_1h, self.timeframe, self.inf_1h, ffill=True) # 4-hour context inf_4h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_4h) inf_4h['ema_trend_4h'] = ta.EMA(inf_4h, timeperiod=30) inf_4h['rsi_4h'] = ta.RSI(inf_4h, timeperiod=14) inf_4h['trend_4h'] = (inf_4h['close'] > inf_4h['ema_trend_4h']).astype(int) inf_4h['market_structure_4h'] = (inf_4h['high'] == inf_4h['high'].rolling(10).max()).astype(int) dataframe = merge_informative_pair(dataframe, inf_4h, self.timeframe, self.inf_4h, ffill=True) # Daily context inf_1d = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1d) inf_1d['ema_trend_1d'] = ta.EMA(inf_1d, timeperiod=20) inf_1d['trend_1d'] = (inf_1d['close'] > inf_1d['ema_trend_1d']).astype(int) inf_1d['weekly_bias'] = (inf_1d['close'] > inf_1d['open']).astype(int) dataframe = merge_informative_pair(dataframe, inf_1d, self.timeframe, self.inf_1d, ffill=True) # Weekly context inf_1w = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1w) inf_1w['ema_trend_1w'] = ta.EMA(inf_1w, timeperiod=10) inf_1w['trend_1w'] = (inf_1w['close'] > inf_1w['ema_trend_1w']).astype(int) dataframe = merge_informative_pair(dataframe, inf_1w, self.timeframe, self.inf_1w, ffill=True) return dataframe def calculate_signal_quality_scores(self, dataframe: DataFrame) -> DataFrame: """Calculate comprehensive signal quality scoring""" # Smart Money Structure Score structure_signals = [ dataframe['bullish_bos'], dataframe['bull_ob_active'] & dataframe['bull_ob_volume_confirmation'], dataframe['fvg_bullish'] & (dataframe['fvg_bull_strength'] > 2), dataframe['sweep_low'], dataframe['wyckoff_accumulation'], dataframe['institutional_buying'] ] dataframe['structure_score'] = sum(structure_signals) / len(structure_signals) # Volume Quality Score volume_signals = [ dataframe['high_volume'], dataframe['volume_spike'], dataframe['institutional_volume'], dataframe['volume_momentum'] > 1.5, dataframe['obv'] > dataframe['obv_ma'], dataframe['mfi'] > 50 ] dataframe['volume_quality_score'] = sum(volume_signals) / len(volume_signals) # Trend Alignment Score trend_signals = [ dataframe['emas_aligned_bullish'], dataframe['close'] > dataframe['vwap'], dataframe['trend_1h'] == 1, dataframe['trend_4h'] == 1, dataframe['trend_1d'] == 1, dataframe['rsi'] > 50 ] dataframe['trend_alignment_score'] = sum(trend_signals) / len(trend_signals) # Market Context Score context_signals = [ dataframe['volatility_regime'] == 'normal', dataframe['adx_1h'] > 25, dataframe['volume_strength_1h'] > 1.2, ~dataframe['bearish_divergence'], dataframe['smart_money_accumulation'] > 0.3 ] dataframe['market_context_score'] = sum(context_signals) / len(context_signals) # Combined Signal Quality dataframe['signal_quality'] = ( dataframe['structure_score'] * 0.35 + dataframe['volume_quality_score'] * 0.25 + dataframe['trend_alignment_score'] * 0.25 + dataframe['market_context_score'] * 0.15 ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Enhanced entry logic with comprehensive smart money analysis""" # === BULLISH SMART MONEY CONDITIONS === # Core smart money structure smart_money_bullish = ( (dataframe['bullish_bos'] & dataframe['bos_volume_confirmation']) | (dataframe['bull_ob_active'] & dataframe['bull_ob_volume_confirmation'] & (dataframe['bull_ob_strength'] > 3)) | (dataframe['fvg_bullish'] & (dataframe['fvg_bull_strength'] > 2) & dataframe['fvg_bull_retested']) | (dataframe['sweep_low'] & (dataframe['liquidity_cluster_strength'] > 0.3)) | (dataframe['wyckoff_accumulation'] & (dataframe['wyckoff_strength'] > 2)) ) # Enhanced trend alignment trend_alignment_bullish = ( (dataframe['emas_aligned_bullish']) & (dataframe['close'] > dataframe['vwap']) & (dataframe['vwap_distance'] > -0.01) & (dataframe['trend_1h'] == 1) & (dataframe['trend_4h'] == 1) & (dataframe['trend_1d'] == 1) ) # Institutional volume confirmation institutional_confirmation = ( (dataframe['institutional_buying'] | dataframe['institutional_volume']) & (dataframe['volume_quality_score'] > 0.6) & (dataframe['smart_money_accumulation'] > 0.4) & (dataframe['money_flow_ratio'] > 1.2) ) # RSI and momentum momentum_bullish = ( (dataframe['rsi'] > self.rsi_oversold.value) & (dataframe['rsi'] < self.rsi_overbought.value) & (dataframe['rsi'] > dataframe['rsi_ma']) & (~dataframe['bearish_divergence']) & (dataframe['rsi_1h'] > 40) ) # Market microstructure microstructure_bullish = ( (dataframe['volatility_regime'] == 'normal') & (dataframe['atr_percent'] < 3.0) & (dataframe['market_structure_4h'] == 1) & (dataframe['adx_1h'] > 20) ) # Signal quality filter quality_filter = ( (dataframe['signal_quality'] > self.min_signal_quality.value) & (dataframe['structure_score'] > 0.5) & (dataframe['trend_alignment_score'] > 0.6) ) # Confluence requirement confluence_count = ( smart_money_bullish.astype(int) + trend_alignment_bullish.astype(int) + institutional_confirmation.astype(int) + momentum_bullish.astype(int) + microstructure_bullish.astype(int) ) confluence_met = confluence_count >= self.confluence_requirement.value # FreqAI integration freqai_condition = True if self.use_freqai_signals.value: freqai_condition = ( (dataframe.get('&-prediction', 0) > 0) & (dataframe.get('&-prediction_confidence', 0) > self.freqai_confidence_threshold.value) ) # Final entry condition dataframe.loc[ (smart_money_bullish) & (trend_alignment_bullish) & (institutional_confirmation) & (momentum_bullish) & (microstructure_bullish) & (quality_filter) & (confluence_met) & (freqai_condition), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Enhanced exit logic with smart money exit signals""" # === SMART MONEY EXIT CONDITIONS === # Structure-based exits structure_exit = ( (dataframe['bearish_bos'] & dataframe['bos_volume_confirmation']) | (dataframe['bear_ob_active'] & dataframe['bear_ob_volume_confirmation']) | (dataframe['fvg_bearish'] & (dataframe['fvg_bear_strength'] > 2)) | (dataframe['sweep_high'] & (dataframe['liquidity_cluster_strength'] > 0.3)) | (dataframe['wyckoff_distribution'] & (dataframe['wyckoff_strength'] > 2)) ) # Momentum exhaustion momentum_exit = ( (dataframe['rsi'] > self.exit_rsi_high.value) | (dataframe['bearish_divergence']) | (dataframe['rsi_1h'] > 75) | (dataframe['institutional_selling']) ) # Trend deterioration trend_exit = ( (~dataframe['emas_aligned_bullish']) | (dataframe['close'] < dataframe['vwap']) | (dataframe['trend_1h'] == 0) | (dataframe['smart_money_distribution'] > 0.5) ) # Volume and liquidity exits volume_exit = ( (dataframe['volume_quality_score'] < 0.3) | (dataframe['volume_momentum'] < 0.8) | (dataframe['money_flow_ratio'] < 0.8) ) # Higher timeframe weakness htf_weakness = ( (dataframe['trend_4h'] == 0) | (dataframe['rsi_4h'] > 80) | (dataframe['market_structure_4h'] == 0) ) # Signal quality deterioration quality_deterioration = ( (dataframe['signal_quality'] < 0.3) | (dataframe['structure_score'] < 0.2) ) dataframe.loc[ (structure_exit) | (momentum_exit) | (trend_exit) | (volume_exit) | (htf_weakness) | (quality_deterioration), 'exit_long'] = 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 smart money signal quality""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Base conservative leverage for smart money base_leverage = 2.0 # Adjust based on signal quality if last_candle['signal_quality'] > 0.8: base_leverage = 3.0 # Higher leverage for high-quality signals elif last_candle['signal_quality'] < 0.5: base_leverage = 1.5 # Lower leverage for weak signals # Adjust based on volatility if last_candle['volatility_regime'] == 'high': base_leverage *= 0.7 # Adjust based on institutional confirmation if (last_candle['institutional_buying'] and last_candle['institutional_confidence'] > 3): base_leverage *= 1.2 return min(base_leverage, max_leverage, 5.0) # Max 5x for smart money def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """Smart money aware dynamic stoploss""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Base ATR stoploss atr_stop = 2.5 * last_candle['atr'] / current_rate # Adjust based on order block levels if last_candle['bull_ob_active'] and trade.is_open and not trade.is_short: # Use order block low as stop level ob_stop = (current_rate - last_candle['bull_ob_low']) / current_rate atr_stop = max(atr_stop, ob_stop) # Adjust based on fair value gaps if last_candle['fvg_bullish'] and trade.is_open and not trade.is_short: fvg_stop = (current_rate - last_candle['fvg_bull_low']) / current_rate atr_stop = max(atr_stop, fvg_stop) # Volatility adjustment if last_candle['volatility_regime'] == 'high': atr_stop *= 1.5 return max(-atr_stop, self.stoploss) 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: """Enhanced entry confirmation with smart money validation""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Signal quality check if last_candle['signal_quality'] < self.min_signal_quality.value: logger.info(f"Entry rejected for {pair}: Low signal quality") return False # Smart money structure validation if not (last_candle['structure_score'] > 0.4): logger.info(f"Entry rejected for {pair}: Weak smart money structure") return False # Volume confirmation if last_candle['volume_quality_score'] < 0.5: logger.info(f"Entry rejected for {pair}: Insufficient volume quality") return False # Market context validation if last_candle['market_context_score'] < 0.4: logger.info(f"Entry rejected for {pair}: Poor market context") return False # Volatility check if last_candle['volatility_regime'] == 'high' and last_candle['atr_percent'] > 4.0: logger.info(f"Entry rejected for {pair}: Excessive volatility") return False # Higher timeframe alignment if not (last_candle['trend_1h'] == 1 and last_candle['trend_4h'] == 1): logger.info(f"Entry rejected for {pair}: Poor higher timeframe alignment") 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]]: """Smart money aware custom exits""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Take profit at smart money levels if current_profit > 0.05: # 5% profit if (last_candle['bear_ob_active'] or last_candle['fvg_bearish'] or last_candle['wyckoff_distribution']): return "smart_money_resistance" # Exit on structure break if (last_candle['bearish_bos'] and last_candle['bos_volume_confirmation'] and current_profit > 0.02): return "structure_break" # Exit on institutional selling if (last_candle['institutional_selling'] and last_candle['institutional_confidence'] > 3 and current_profit > 0.01): return "institutional_selling" # Exit on liquidity sweep if (last_candle['sweep_high'] and current_profit > 0.03): return "liquidity_sweep" # Signal quality deterioration if (last_candle['signal_quality'] < 0.2 and current_profit > 0.005): return "signal_deterioration" return None def bot_loop_start(self, **kwargs) -> None: """Enhanced bot loop with smart money tracking""" pass def check_entry_timeout(self, pair: str, trade: 'Trade', order: 'Order', current_time: datetime, **kwargs) -> bool: """Smart money entry timeout logic""" return False def check_exit_timeout(self, pair: str, trade: 'Trade', order: 'Order', current_time: datetime, **kwargs) -> bool: """Smart money exit timeout logic""" return False