ClaudeSmartMoney

# 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

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


class ClaudeSmartMoney(IStrategy):
    """
    Professional Smart Money Concepts Strategy with Volume Detection
    
    This strategy combines institutional smart money concepts with advanced volume analysis:
    - Order blocks and fair value gaps identification
    - Liquidity sweeps and stop hunts detection
    - Volume profile analysis
    - Market structure breaks
    - Risk management with proper position sizing
    
    Designed for hyperopt, backtesting, FreqAI and ML integration.
    No future leak or data bias - all indicators use historical data only.
    """

    INTERFACE_VERSION = 3

    # Minimal ROI designed for the strategy
    minimal_roi = {
        "0": 0.20,    # 20% at any time
        "15": 0.10,   # 10% after 15 minutes
        "30": 0.05,   # 5% after 30 minutes
        "60": 0.02,   # 2% after 1 hour
        "120": 0.01   # 1% after 2 hours
    }

    # Optimal stoploss
    stoploss = -0.08  # 8% stoploss

    # Optimal timeframe for the strategy
    timeframe = '15m'
    
    # Informative timeframes for multi-timeframe analysis
    inf_1h = '1h'
    inf_4h = '4h'
    inf_1d = '1d'

    # Run "populate_indicators" only for new candle
    process_only_new_candles = True

    # These values can be overridden in the config
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # Number of candles the strategy requires before producing valid signals
    startup_candle_count: int = 200

    # Strategy parameters - Hyperopt enabled
    
    # === Volume Analysis Parameters ===
    volume_lookback = IntParameter(10, 30, default=20, space='buy', optimize=True)
    volume_threshold = DecimalParameter(1.2, 3.0, default=2.0, space='buy', optimize=True)
    volume_ma_period = IntParameter(10, 50, default=20, space='buy', optimize=True)
    
    # === Smart Money Structure Parameters ===
    structure_lookback = IntParameter(20, 50, default=30, space='buy', optimize=True)
    liquidity_threshold = DecimalParameter(0.5, 2.0, default=1.0, space='buy', optimize=True)
    order_block_strength = IntParameter(3, 10, default=5, space='buy', optimize=True)
    
    # === Fair Value Gap Parameters ===
    fvg_min_size = DecimalParameter(0.1, 1.0, default=0.3, space='buy', optimize=True)
    fvg_lookback = IntParameter(5, 20, default=10, space='buy', optimize=True)
    
    # === Trend Parameters ===
    ema_fast = IntParameter(8, 21, default=12, space='buy', optimize=True)
    ema_slow = IntParameter(21, 55, default=34, space='buy', optimize=True)
    ema_trend = IntParameter(50, 200, default=100, space='buy', optimize=True)
    
    # === RSI Parameters ===
    rsi_period = IntParameter(10, 21, default=14, space='buy', optimize=True)
    rsi_overbought = IntParameter(70, 85, default=75, space='buy', optimize=True)
    rsi_oversold = IntParameter(15, 30, default=25, space='buy', optimize=True)
    
    # === Exit Parameters ===
    exit_rsi_high = IntParameter(75, 90, default=80, space='sell', optimize=True)
    exit_rsi_low = IntParameter(10, 25, default=20, space='sell', optimize=True)
    
    # === Risk Management ===
    max_open_trades = 3
    position_adjustment_enable = True

    def informative_pairs(self):
        """
        Define additional, informative pair/interval combinations to be cached from the exchange.
        These pairs will automatically be available in strategy methods.
        """
        pairs = self.dp.current_whitelist()
        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]
        return informative_pairs

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Populate all indicators used by the strategy
        """
        
        # === 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
        
        # === Moving Averages ===
        dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast.value)
        dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_slow.value)
        dataframe['ema_trend'] = ta.EMA(dataframe, timeperiod=self.ema_trend.value)
        
        # === RSI ===
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)
        
        # === 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
        
        # Volume-weighted indicators
        dataframe['vwap'] = qtpylib.vwap(dataframe)
        
        # === ATR for volatility ===
        dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
        dataframe['atr_percent'] = (dataframe['atr'] / dataframe['close']) * 100
        
        # === Smart Money Concepts ===
        dataframe = self.calculate_order_blocks(dataframe)
        dataframe = self.calculate_fair_value_gaps(dataframe)
        dataframe = self.calculate_market_structure(dataframe)
        dataframe = self.calculate_liquidity_levels(dataframe)
        
        # === Higher Timeframe Analysis ===
        dataframe = self.populate_higher_timeframe_indicators(dataframe, metadata)
        
        return dataframe

    def calculate_order_blocks(self, dataframe: DataFrame) -> DataFrame:
        """
        Identify order blocks - zones where institutional orders are likely placed
        """
        # Bullish order blocks - last down candle before strong up move
        dataframe['bull_ob_high'] = 0.0
        dataframe['bull_ob_low'] = 0.0
        dataframe['bull_ob_active'] = False
        
        # Bearish order blocks - last up candle before strong down move  
        dataframe['bear_ob_high'] = 0.0
        dataframe['bear_ob_low'] = 0.0
        dataframe['bear_ob_active'] = False
        
        for i in range(self.order_block_strength.value, len(dataframe)):
            # Look for bullish order blocks
            if (dataframe.iloc[i]['close'] > dataframe.iloc[i-1]['high'] and  # Break of previous high
                dataframe.iloc[i]['volume'] > dataframe.iloc[i-1]['volume'] * 1.5):  # Strong volume
                
                # Find the last bearish candle before the break
                for j in range(i-1, max(0, i-self.order_block_strength.value), -1):
                    if dataframe.iloc[j]['close'] < dataframe.iloc[j]['open']:  # Bearish candle
                        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
                        break
            
            # Look for bearish order blocks
            if (dataframe.iloc[i]['close'] < dataframe.iloc[i-1]['low'] and  # Break of previous low
                dataframe.iloc[i]['volume'] > dataframe.iloc[i-1]['volume'] * 1.5):  # Strong volume
                
                # Find the last bullish candle before the break
                for j in range(i-1, max(0, i-self.order_block_strength.value), -1):
                    if dataframe.iloc[j]['close'] > dataframe.iloc[j]['open']:  # Bullish candle
                        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
                        break
        
        return dataframe

    def calculate_fair_value_gaps(self, dataframe: DataFrame) -> DataFrame:
        """
        Identify Fair Value Gaps (FVG) - price gaps that indicate inefficiency
        """
        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
        
        for i in range(2, len(dataframe)):
            # Bullish FVG: gap between candle i-2 high and candle i low
            if (dataframe.iloc[i-2]['high'] < dataframe.iloc[i]['low'] and
                dataframe.iloc[i-1]['high'] < dataframe.iloc[i]['low'] and
                (dataframe.iloc[i]['low'] - dataframe.iloc[i-2]['high']) / dataframe.iloc[i]['close'] > self.fvg_min_size.value / 100):
                
                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']
            
            # Bearish FVG: gap between candle i-2 low and candle i high  
            if (dataframe.iloc[i-2]['low'] > dataframe.iloc[i]['high'] and
                dataframe.iloc[i-1]['low'] > dataframe.iloc[i]['high'] and
                (dataframe.iloc[i-2]['low'] - dataframe.iloc[i]['high']) / dataframe.iloc[i]['close'] > self.fvg_min_size.value / 100):
                
                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']
        
        return dataframe

    def calculate_market_structure(self, dataframe: DataFrame) -> DataFrame:
        """
        Identify market structure breaks and trend changes
        """
        # Calculate swing highs and lows
        dataframe['swing_high'] = (
            (dataframe['high'] > dataframe['high'].shift(1)) &
            (dataframe['high'] > dataframe['high'].shift(-1)) &
            (dataframe['high'] > dataframe['high'].shift(2)) &
            (dataframe['high'] > dataframe['high'].shift(-2))
        )
        
        dataframe['swing_low'] = (
            (dataframe['low'] < dataframe['low'].shift(1)) &
            (dataframe['low'] < dataframe['low'].shift(-1)) &
            (dataframe['low'] < dataframe['low'].shift(2)) &
            (dataframe['low'] < dataframe['low'].shift(-2))
        )
        
        # Market structure breaks
        dataframe['bullish_bos'] = False
        dataframe['bearish_bos'] = False
        
        # Simple structure break logic
        for i in range(self.structure_lookback.value, len(dataframe)):
            # Recent swing high broken to upside
            recent_highs = dataframe.iloc[i-self.structure_lookback.value:i][dataframe.iloc[i-self.structure_lookback.value:i]['swing_high']]['high']
            if len(recent_highs) > 0 and dataframe.iloc[i]['close'] > recent_highs.max():
                dataframe.iloc[i, dataframe.columns.get_loc('bullish_bos')] = True
            
            # Recent swing low broken to downside
            recent_lows = dataframe.iloc[i-self.structure_lookback.value:i][dataframe.iloc[i-self.structure_lookback.value:i]['swing_low']]['low']
            if len(recent_lows) > 0 and dataframe.iloc[i]['close'] < recent_lows.min():
                dataframe.iloc[i, dataframe.columns.get_loc('bearish_bos')] = True
        
        return dataframe

    def calculate_liquidity_levels(self, dataframe: DataFrame) -> DataFrame:
        """
        Identify liquidity levels where stops are likely placed
        """
        # Equal highs/lows where liquidity sits
        dataframe['liquidity_high'] = 0.0
        dataframe['liquidity_low'] = 0.0
        dataframe['sweep_high'] = False
        dataframe['sweep_low'] = False
        
        for i in range(20, len(dataframe)):
            # Look for equal highs (liquidity levels)
            recent_highs = dataframe.iloc[i-20:i]['high']
            high_clusters = recent_highs[abs(recent_highs - recent_highs.max()) < (recent_highs.max() * 0.002)]
            
            if len(high_clusters) >= 2:
                dataframe.iloc[i, dataframe.columns.get_loc('liquidity_high')] = recent_highs.max()
                
                # Check for liquidity sweep
                if dataframe.iloc[i]['high'] > recent_highs.max() and dataframe.iloc[i]['close'] < recent_highs.max():
                    dataframe.iloc[i, dataframe.columns.get_loc('sweep_high')] = True
            
            # Look for equal lows (liquidity levels)
            recent_lows = dataframe.iloc[i-20:i]['low']
            low_clusters = recent_lows[abs(recent_lows - recent_lows.min()) < (recent_lows.min() * 0.002)]
            
            if len(low_clusters) >= 2:
                dataframe.iloc[i, dataframe.columns.get_loc('liquidity_low')] = recent_lows.min()
                
                # Check for liquidity sweep
                if dataframe.iloc[i]['low'] < recent_lows.min() and dataframe.iloc[i]['close'] > recent_lows.min():
                    dataframe.iloc[i, dataframe.columns.get_loc('sweep_low')] = True
        
        return dataframe

    def populate_higher_timeframe_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Add higher timeframe context for better decision making
        """
        # 1H timeframe
        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['trend_1h'] = (inf_1h['close'] > inf_1h['ema_trend_1h']).astype(int)
        
        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['ema_trend_4h'] = ta.EMA(inf_4h, timeperiod=21)
        inf_4h['trend_4h'] = (inf_4h['close'] > inf_4h['ema_trend_4h']).astype(int)
        
        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['ema_trend_1d'] = ta.EMA(inf_1d, timeperiod=12)
        inf_1d['trend_1d'] = (inf_1d['close'] > inf_1d['ema_trend_1d']).astype(int)
        
        dataframe = merge_informative_pair(dataframe, inf_1d, self.timeframe, self.inf_1d, ffill=True)
        
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the entry signal for the given dataframe
        """
        conditions = []
        
        # === BULLISH CONDITIONS ===
        
        # 1. Higher timeframe trend alignment
        higher_tf_bullish = (
            (dataframe['trend_1h'] == 1) &
            (dataframe['trend_4h'] == 1) &
            (dataframe['trend_1d'] == 1)
        )
        
        # 2. Price action conditions
        price_bullish = (
            (dataframe['close'] > dataframe['ema_fast']) &
            (dataframe['ema_fast'] > dataframe['ema_slow']) &
            (dataframe['close'] > dataframe['vwap'])
        )
        
        # 3. Smart money structure
        structure_bullish = (
            (dataframe['bullish_bos']) |
            (dataframe['bull_ob_active']) |
            (dataframe['fvg_bullish']) |
            (dataframe['sweep_low'])
        )
        
        # 4. Volume confirmation
        volume_bullish = (
            (dataframe['high_volume']) &
            (dataframe['volume'] > dataframe['volume'].shift(1))
        )
        
        # 5. RSI conditions
        rsi_bullish = (
            (dataframe['rsi'] > self.rsi_oversold.value) &
            (dataframe['rsi'] < self.rsi_overbought.value) &
            (dataframe['rsi'] > dataframe['rsi'].shift(1))
        )
        
        # 6. Volatility filter
        volatility_ok = dataframe['atr_percent'] < 5.0  # Not too volatile
        
        # Combine all bullish conditions
        dataframe.loc[
            (higher_tf_bullish) &
            (price_bullish) &
            (structure_bullish) &
            (volume_bullish) &
            (rsi_bullish) &
            (volatility_ok),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the exit signal for the given dataframe
        """
        # === EXIT LONG CONDITIONS ===
        
        # 1. RSI overbought
        rsi_exit = (
            (dataframe['rsi'] > self.exit_rsi_high.value)
        )
        
        # 2. Price action reversal
        reversal_signals = (
            (dataframe['bearish_bos']) |
            (dataframe['bear_ob_active']) |
            (dataframe['fvg_bearish']) |
            (dataframe['sweep_high'])
        )
        
        # 3. Moving average cross down
        ma_cross_down = (
            (dataframe['ema_fast'] < dataframe['ema_slow']) &
            (dataframe['close'] < dataframe['ema_fast'])
        )
        
        # 4. Volume declining
        volume_declining = (
            (dataframe['volume'] < dataframe['volume_sma']) &
            (dataframe['volume'] < dataframe['volume'].shift(1))
        )
        
        # 5. Higher timeframe turning bearish
        htf_bearish = (
            (dataframe['trend_1h'] == 0) |
            (dataframe['rsi_1h'] > 75)
        )
        
        dataframe.loc[
            (rsi_exit) |
            (reversal_signals) |
            (ma_cross_down & volume_declining) |
            (htf_bearish),
            '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:
        """
        Customize leverage for each new trade. This method is only called in futures mode.
        """
        # Conservative leverage for smart money strategy
        return min(proposed_leverage, 3.0)

    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
                       current_rate: float, current_profit: float, **kwargs) -> float:
        """
        Custom stoploss logic - trailing stop based on ATR
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = dataframe.iloc[-1].squeeze()
        
        # ATR-based trailing stop
        atr_multiplier = 2.0
        atr_stop = atr_multiplier * last_candle['atr'] / current_rate
        
        # Don't allow stop to be closer than original stoploss
        return max(self.stoploss, -atr_stop)

    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:
        """
        Called right before placing a entry order.
        Timing of this method is as close as possible to the actual order placement.
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = dataframe.iloc[-1].squeeze()
        
        # Additional safety checks
        # 1. Ensure we're not buying at resistance levels
        if side == "long":
            # Don't buy if price is at a bearish order block
            if last_candle['bear_ob_active'] and last_candle['bear_ob_low'] <= rate <= last_candle['bear_ob_high']:
                return False
            
            # Don't buy if RSI too high
            if last_candle['rsi'] > 70:
                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 beyond the exit signals
        """
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = dataframe.iloc[-1].squeeze()
        
        # Take profit at major resistance levels
        if trade.is_open and current_profit > 0.05:  # 5% profit
            if last_candle['bear_ob_active']:
                return "resistance_exit"
        
        # Emergency exit on major bearish structure break
        if last_candle['bearish_bos'] and current_profit < -0.03:  # -3% loss with structure break
            return "emergency_exit"
        
        return None

    def bot_loop_start(self, **kwargs) -> None:
        """
        Called at the start of the bot iteration (one loop).
        Might be used to perform pair-independent tasks
        (e.g. gather some remote resource for comparison)
        """
        pass

    def check_entry_timeout(self, pair: str, trade: 'Trade', order: 'Order',
                           current_time: datetime, **kwargs) -> bool:
        """
        Check entry timeout function callback.
        This method can be used to override the entry-timeout.
        """
        return False

    def check_exit_timeout(self, pair: str, trade: 'Trade', order: 'Order',
                          current_time: datetime, **kwargs) -> bool:
        """
        Check exit timeout function callback.
        This method can be used to override the exit-timeout.
        """
        return False