SharkMeatHarmonic

import logging
from datetime import datetime
from typing import Dict, List, Optional, Tuple
import talib.abstract as ta
from freqtrade.strategy import IStrategy, merge_informative_pair
from freqtrade.persistence import Trade
from pandas import DataFrame
import numpy as np

logger = logging.getLogger(__name__)

class SharkMeatHarmonic(IStrategy):
    INTERFACE_VERSION = 3
    
    # --- [ TIMEFRAMES ] ---
    timeframe = '5m'
    inf_1h = '1h'
    startup_candle_count: int = 1000
    process_only_new_candles = True

    # --- [ RISK MANAGEMENT ] ---
    can_short = True
    stoploss = -0.10  # Hard stop at -10% price change
    minimal_roi = {"0": 100.0} 
    use_custom_stoploss = True

    # --- [ STRATEGY PARAMETERS ] ---
    error_rate = 0.05  # 5% tolerance for Fibonacci ratios
    _last_btc_log_time = None
    _pattern_logged = {} 
    _developing_patterns = {} 

    def informative_pairs(self) -> List[Tuple[str, str]]:
        pairs = self.dp.current_whitelist()
        informative = [(pair, self.inf_1h) for pair in pairs]
        btc_pair = "BTC/USDT:USDT"
        if btc_pair not in pairs:
            informative.append((btc_pair, self.inf_1h))
        return informative

    def get_pivots(self, df: DataFrame, window: int = 5) -> List[Tuple[int, float]]:
        """ Identifies XABCD pivot points using local price extrema. """
        pivots = []
        for i in range(window, len(df) - window):
            if df['high'].iloc[i] == df['high'].iloc[i-window:i+window].max():
                pivots.append((i, float(df['high'].iloc[i])))
            elif df['low'].iloc[i] == df['low'].iloc[i-window:i+window].min():
                pivots.append((i, float(df['low'].iloc[i])))
        return pivots[-5:]

    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:
        return 5.0

    def validate_harmonics(self, pivots: List[Tuple[int, float]]) -> Optional[Tuple[str, str, Dict, Dict]]:
        if len(pivots) < 5: return None
        
        _, x = pivots[0]
        _, a = pivots[1]
        _, b = pivots[2]
        _, c = pivots[3]
        _, d = pivots[4]
        
        xa, ab, bc = abs(a - x), abs(b - a), abs(c - b)
        if xa == 0 or ab == 0: return None
        
        ab_xa, bc_ab, d_xa = ab / xa, bc / ab, abs(d - x) / xa
        direction = "bullish" if d < c else "bearish"
        
        ratios = {'ab_xa': ab_xa, 'bc_ab': bc_ab, 'd_xa': d_xa}
        prices = {'X': x, 'A': a, 'B': b, 'C': c, 'D': d}

        # Bat (0.886 D)
        if (0.382 <= ab_xa <= 0.50) and (0.382 <= bc_ab <= 0.886) and \
           (0.886 * (1-self.error_rate) <= d_xa <= 0.886 * (1+self.error_rate)):
            return (direction, "Bat", ratios, prices)
        
        # Gartley (0.786 D)
        if (0.618 * (1-self.error_rate) <= ab_xa <= 0.618 * (1+self.error_rate)) and \
           (0.382 <= bc_ab <= 0.886) and \
           (0.786 * (1-self.error_rate) <= d_xa <= 0.786 * (1+self.error_rate)):
            return (direction, "Gartley", ratios, prices)

        # Crab (1.618 D)
        if (0.382 <= ab_xa <= 0.618) and (0.382 <= bc_ab <= 0.886) and \
           (1.618 * (1-self.error_rate) <= d_xa <= 1.618 * (1+self.error_rate)):
            return (direction, "Crab", ratios, prices)

        return None

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe['engulfing'] = ta.CDLENGULFING(dataframe)
        inf = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1h)
        if inf is None or inf.empty: return dataframe

        pivots = self.get_pivots(inf)
        h_result = self.validate_harmonics(pivots)
        
        if h_result:
            direction, pattern, ratios, prices = h_result
            inf['harmonic_bull'] = 1 if direction == "bullish" else 0
            inf['harmonic_bear'] = 1 if direction == "bearish" else 0
            inf['harmonic_pattern'] = pattern
        else:
            inf['harmonic_bull'] = 0
            inf['harmonic_bear'] = 0
            inf['harmonic_pattern'] = ""

        return merge_informative_pair(dataframe, inf, self.timeframe, self.inf_1h, ffill=True)

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Bullish Entry
        harmonic_long = dataframe[f'harmonic_bull_{self.inf_1h}'] == 1
        long_entries = harmonic_long & (dataframe['engulfing'] == 100)
        dataframe.loc[long_entries, 'enter_long'] = 1
        dataframe.loc[long_entries, 'enter_tag'] = "Harmonic_Bull"

        # Bearish Entry
        harmonic_short = dataframe[f'harmonic_bear_{self.inf_1h}'] == 1
        short_entries = harmonic_short & (dataframe['engulfing'] == -100)
        dataframe.loc[short_entries, 'enter_short'] = 1
        dataframe.loc[short_entries, 'enter_tag'] = "Harmonic_Bear"

        return dataframe

    def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
                        current_rate: float, current_profit: float, **kwargs) -> float:
        if 0.03 <= current_profit < 0.10:
            return -0.03
        if current_profit >= 0.10:
            return -0.01
        return self.stoploss