# technical_analysis.py

import asyncio
import logging
import time
from typing import Dict, List, Tuple
from datetime import datetime, timedelta

import pandas as pd
import numpy as np
from ta.momentum import (
    RSIIndicator,
    MoneyFlowIndexIndicator,
)
from ta.trend import (
    ADXIndicator,
    MACD,
    SMAIndicator,
    KeltnerChannel,
)
from ta.volatility import BollingerBands, AverageTrueRange
from functools import wraps

# Django setup
import os
import django
from django.utils import timezone

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "binancebot.settings")
django.setup()

from .models import Kline, Signal, TradingPair  # Adjust the import according to your project structure

# Logger setup with enhanced formatting
logger = logging.getLogger(__name__)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

# Global settings for retry mechanism and circuit breaker
MAX_RETRIES = 5
CIRCUIT_BREAKER_THRESHOLD = 3  # Number of consecutive failures before opening the circuit
CIRCUIT_BREAKER_TIMEOUT = 60  # Time in seconds to keep the circuit open

def retry_with_exponential_backoff(coroutine):
    """
    Decorator for retrying a coroutine with exponential backoff.
    """
    @wraps(coroutine)
    async def wrapper(*args, **kwargs):
        retries = 0
        delay = 1
        while retries < MAX_RETRIES:
            try:
                result = await coroutine(*args, **kwargs)
                return result
            except Exception as e:
                retries += 1
                logger.warning(f"Retry {retries}/{MAX_RETRIES} for {coroutine.__name__}: {e}")
                await asyncio.sleep(delay)
                delay *= 2  # Exponential backoff
        logger.error(f"Max retries exceeded for {coroutine.__name__}")
        raise Exception(f"Max retries exceeded for {coroutine.__name__}")
    return wrapper

class CircuitBreaker:
    """
    A simple circuit breaker implementation.
    """
    failure_count = 0
    circuit_open = False
    last_failure_time = None

    @staticmethod
    def call(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            if CircuitBreaker.circuit_open:
                elapsed_time = time.time() - CircuitBreaker.last_failure_time
                if elapsed_time > CIRCUIT_BREAKER_TIMEOUT:
                    CircuitBreaker.circuit_open = False
                    CircuitBreaker.failure_count = 0
                    logger.info("Circuit breaker closed, resuming operation.")
                else:
                    logger.error("Circuit breaker is open. Call rejected.")
                    raise Exception("Circuit breaker is open.")
            try:
                result = await func(*args, **kwargs)
                CircuitBreaker.failure_count = 0  # Reset on success
                return result
            except Exception as e:
                CircuitBreaker.failure_count += 1
                CircuitBreaker.last_failure_time = time.time()
                if CircuitBreaker.failure_count >= CIRCUIT_BREAKER_THRESHOLD:
                    CircuitBreaker.circuit_open = True
                    logger.error("Circuit breaker opened due to repeated failures.")
                raise e
        return wrapper

class CryptoBot:
    def __init__(
        self,
        timeframes: List[str] = ["1m", "5m", "15m", "1h"],
        num_candles: int = 500,
    ):
        self.timeframes = timeframes
        self.num_candles = num_candles
        logger.info(
            "CryptoBot initialized with timeframes: %s and num_candles: %d",
            timeframes,
            num_candles,
        )

    @retry_with_exponential_backoff
    async def get_candles_df(self, pair: str, timeframe: str) -> pd.DataFrame:
        """
        Fetches the most recent `num_candles` records for the specified trading pair and timeframe.
        Includes data preprocessing to handle missing values and outliers.
        """
        try:
            klines = (
                Kline.objects.filter(symbol=pair, timeframe=timeframe)
                .order_by("-timestamp")[: self.num_candles]
                .values()
            )
            df = pd.DataFrame.from_records(klines)
            df.set_index("timestamp", inplace=True)
            df.sort_index(inplace=True)  # Ensure chronological order for analysis

            # Data preprocessing
            df.replace([np.inf, -np.inf], np.nan, inplace=True)
            df.fillna(method="ffill", inplace=True)  # Forward-fill missing values
            df.dropna(inplace=True)  # Drop any remaining NaNs

            # Outlier detection and handling using Z-score
            z_scores = ((df - df.mean()) / df.std()).abs()
            df = df[(z_scores < 3).all(axis=1)]

            return df
        except Exception as e:
            logger.error(f"Error fetching candles for {pair} ({timeframe}): {e}")
            raise

    async def calculate_technical_indicators_set1(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Calculates the first set of technical indicators and adds them to the DataFrame.
        """
        try:
            # First set of indicators
            df["rsi"] = RSIIndicator(close=df["close"], window=14).rsi()
            macd = MACD(
                close=df["close"], window_slow=26, window_fast=12, window_sign=9
            )
            df["macd"] = macd.macd()
            df["macd_signal"] = macd.macd_signal()
            df["sma_fast"] = SMAIndicator(close=df["close"], window=14).sma_indicator()
            df["sma_slow"] = SMAIndicator(close=df["close"], window=28).sma_indicator()
            bollinger = BollingerBands(
                close=df["close"], window=20, window_dev=2
            )
            df["bollinger_mavg"] = bollinger.bollinger_mavg()
            df["bollinger_hband"] = bollinger.bollinger_hband()
            df["bollinger_lband"] = bollinger.bollinger_lband()
            # Additional preprocessing
            df.replace([np.inf, -np.inf], np.nan, inplace=True)
            df.fillna(method="ffill", inplace=True)
            df.dropna(inplace=True)
            return df
        except Exception as e:
            logger.error(f"Error calculating indicators set 1: {e}")
            raise

    async def calculate_technical_indicators_set2(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Calculates the second set of technical indicators and adds them to the DataFrame.
        """
        try:
            # Second set of indicators
            df["mfi"] = MoneyFlowIndexIndicator(
                high=df["high"], low=df["low"], close=df["close"], volume=df["volume"], window=14
            ).money_flow_index()
            df["rvi"] = self.calculate_rvi(df)
            keltner = KeltnerChannel(
                high=df["high"], low=df["low"], close=df["close"], window=20
            )
            df["keltner_mavg"] = keltner.keltner_channel_mband()
            df["keltner_hband"] = keltner.keltner_channel_hband()
            df["keltner_lband"] = keltner.keltner_channel_lband()
            df["adx"] = ADXIndicator(
                high=df["high"], low=df["low"], close=df["close"], window=14
            ).adx()
            # Additional preprocessing
            df.replace([np.inf, -np.inf], np.nan, inplace=True)
            df.fillna(method="ffill", inplace=True)
            df.dropna(inplace=True)
            return df
        except Exception as e:
            logger.error(f"Error calculating indicators set 2: {e}")
            raise

    def calculate_rvi(self, df: pd.DataFrame) -> pd.Series:
        """
        Calculates the Relative Vigor Index (RVI).
        """
        close_open = df["close"] - df["open"]
        high_low = df["high"] - df["low"]
        numerator = close_open.rolling(window=10).mean()
        denominator = high_low.rolling(window=10).mean()
        rvi = numerator / denominator
        return rvi

    @CircuitBreaker.call
    async def generate_signal_set(
        self,
        pair: str,
        timeframe: str,
        higher_timeframe: str = "1h",
        lower_timeframe: str = "15m",
        set_number: int = 1
    ) -> Tuple[str, Dict[str, float]]:
        """
        Generates signal for a given set of indicators.
        """
        try:
            df = await self.get_candles_df(pair, timeframe)
            if df.empty:
                return "No Signal", {}

            if set_number == 1:
                df_set = await self.calculate_technical_indicators_set1(df.copy())
            else:
                df_set = await self.calculate_technical_indicators_set2(df.copy())

            # Get higher timeframe data for trend confirmation
            higher_df = await self.get_candles_df(pair, higher_timeframe)
            if higher_df.empty:
                return "No Signal", {}

            if set_number == 1:
                higher_df = await self.calculate_technical_indicators_set1(higher_df)
            else:
                higher_df = await self.calculate_technical_indicators_set2(higher_df)

            # Get lower timeframe data for additional confirmation
            lower_df = await self.get_candles_df(pair, lower_timeframe)
            if lower_df.empty:
                return "No Signal", {}

            if set_number == 1:
                lower_df = await self.calculate_technical_indicators_set1(lower_df)
            else:
                lower_df = await self.calculate_technical_indicators_set2(lower_df)

            current_price = df["close"].iloc[-1]
            current_time = pd.Timestamp.now()

            # Calculate 24h change
            change_24h = self.calculate_24h_change(df)

            # Generate signals
            if set_number == 1:
                signal, reasons = self.generate_signal_from_set1(df_set, higher_df, lower_df)
                # Determine market regime
                market_regime = self.determine_market_regime(higher_df)
            else:
                signal, reasons = self.generate_signal_from_set2(df_set, higher_df, lower_df)
                # Determine market regime
                market_regime = self.determine_market_regime(higher_df)

            # Calculate support and resistance
            support_level, resistance_level = self.calculate_support_and_resistance(df)

            # Calculate stop-loss and take-profit
            stop_loss, take_profit = self.calculate_stop_loss_take_profit(
                signal, current_price, df
            )

            # Determine risk-reward ratio
            risk_reward_ratio = self.calculate_risk_reward_ratio(
                signal, current_price, stop_loss, take_profit
            )

            # Determine the result (Won, Lost, Pending)
            result = "Pending"  # Since real-time price movement is needed

            # Initial success rate and win probability
            success_rate = 50  # Base success rate
            win_probability = success_rate  # For now, set equal to success rate

            # Prepare additional info
            additional_info = {
                "reasons": reasons,
                "support_level": support_level,
                "resistance_level": resistance_level,
            }

            metrics = {
                "timestamp": df.index[-1],
                "symbol": pair,
                "price": current_price,
                "price_change_24h": change_24h,
                "signal": signal,
                "stop_loss": stop_loss,
                "take_profit": take_profit,
                "success_rate": success_rate,
                "win_probability": win_probability,
                "risk_reward_ratio": risk_reward_ratio,
                "market_regime": market_regime,
                "result": result,
                "additional_info": additional_info,
            }

            logger.info(
                f"Generated signal for {pair} ({timeframe}) Set {set_number}: {signal}"
            )

            return signal, metrics
        except Exception as e:
            logger.error(f"Error generating signal for {pair} ({timeframe}): {str(e)}")
            raise

    def calculate_24h_change(self, df: pd.DataFrame) -> float:
        """
        Calculates the percentage change over the last 24 hours.
        """
        try:
            current_time = df.index[-1]
            past_time = current_time - timedelta(hours=24)
            past_price = df[df.index >= past_time]["close"].iloc[0]
            current_price = df["close"].iloc[-1]
            change = ((current_price - past_price) / past_price) * 100
            return round(change, 2)
        except Exception as e:
            logger.error(f"Error calculating 24h change: {e}")
            return 0.0

    def generate_signal_from_set1(
        self, df: pd.DataFrame, higher_df: pd.DataFrame, lower_df: pd.DataFrame
    ) -> Tuple[str, List[str]]:
        """
        Generates signal based on the first set of indicators.
        """
        signal_strength = 0
        reasons = []

        # RSI
        rsi = df["rsi"].iloc[-1]
        if rsi < 30:
            signal_strength += 1
            reasons.append("RSI indicates oversold conditions.")
        elif rsi > 70:
            signal_strength -= 1
            reasons.append("RSI indicates overbought conditions.")

        # RSI Divergence
        if self.detect_rsi_divergence(df):
            signal_strength += 1
            reasons.append("Bullish RSI divergence detected.")

        # MACD crossover
        if df["macd"].iloc[-1] > df["macd_signal"].iloc[-1]:
            signal_strength += 1
            reasons.append("MACD bullish crossover.")
        elif df["macd"].iloc[-1] < df["macd_signal"].iloc[-1]:
            signal_strength -= 1
            reasons.append("MACD bearish crossover.")

        # SMA crossover
        if df["sma_fast"].iloc[-1] > df["sma_slow"].iloc[-1]:
            signal_strength += 1
            reasons.append("SMA bullish crossover.")
        elif df["sma_fast"].iloc[-1] < df["sma_slow"].iloc[-1]:
            signal_strength -= 1
            reasons.append("SMA bearish crossover.")

        # Bollinger Bands
        if df["close"].iloc[-1] < df["bollinger_lband"].iloc[-1]:
            signal_strength += 1
            reasons.append("Price below Bollinger lower band.")
        elif df["close"].iloc[-1] > df["bollinger_hband"].iloc[-1]:
            signal_strength -= 1
            reasons.append("Price above Bollinger upper band.")

        # Higher timeframe trend confirmation
        if higher_df["sma_fast"].iloc[-1] > higher_df["sma_slow"].iloc[-1]:
            signal_strength += 1
            reasons.append("Higher timeframe trend is bullish.")
        else:
            signal_strength -= 1
            reasons.append("Higher timeframe trend is bearish.")

        # Lower timeframe momentum
        if lower_df["rsi"].iloc[-1] > 50:
            signal_strength += 1
            reasons.append("Lower timeframe momentum is bullish.")
        else:
            signal_strength -= 1
            reasons.append("Lower timeframe momentum is bearish.")

        signal = self.determine_signal(signal_strength)

        return signal, reasons

    def generate_signal_from_set2(
        self, df: pd.DataFrame, higher_df: pd.DataFrame, lower_df: pd.DataFrame
    ) -> Tuple[str, List[str]]:
        """
        Generates signal based on the second set of indicators.
        """
        signal_strength = 0
        reasons = []

        # MFI
        mfi = df["mfi"].iloc[-1]
        if mfi < 20:
            signal_strength += 1
            reasons.append("MFI indicates oversold conditions.")
        elif mfi > 80:
            signal_strength -= 1
            reasons.append("MFI indicates overbought conditions.")

        # RVI
        if df["rvi"].iloc[-1] > 0:
            signal_strength += 1
            reasons.append("RVI indicates bullish momentum.")
        else:
            signal_strength -= 1
            reasons.append("RVI indicates bearish momentum.")

        # Keltner Channels
        if df["close"].iloc[-1] < df["keltner_lband"].iloc[-1]:
            signal_strength += 1
            reasons.append("Price below Keltner lower band.")
        elif df["close"].iloc[-1] > df["keltner_hband"].iloc[-1]:
            signal_strength -= 1
            reasons.append("Price above Keltner upper band.")

        # ADX
        if df["adx"].iloc[-1] > 25:
            signal_strength += 1
            reasons.append("ADX indicates a strong trend.")
        else:
            signal_strength -= 1
            reasons.append("ADX indicates a weak trend.")

        # Higher timeframe trend confirmation
        if higher_df["rvi"].iloc[-1] > 0:
            signal_strength += 1
            reasons.append("Higher timeframe momentum is bullish.")
        else:
            signal_strength -= 1
            reasons.append("Higher timeframe momentum is bearish.")

        # Lower timeframe momentum
        if lower_df["mfi"].iloc[-1] > 50:
            signal_strength += 1
            reasons.append("Lower timeframe MFI indicates buying pressure.")
        else:
            signal_strength -= 1
            reasons.append("Lower timeframe MFI indicates selling pressure.")

        signal = self.determine_signal(signal_strength)

        return signal, reasons

    def detect_rsi_divergence(self, df: pd.DataFrame) -> bool:
        """
        Detects bullish RSI divergence.
        """
        try:
            # Price makes lower lows
            price_diff = df["close"].iloc[-2] - df["close"].iloc[-1]
            # RSI makes higher lows
            rsi_diff = df["rsi"].iloc[-1] - df["rsi"].iloc[-2]
            if price_diff > 0 and rsi_diff > 0:
                return True
            else:
                return False
        except Exception as e:
            logger.error(f"Error detecting RSI divergence: {e}")
            return False

    def determine_signal(self, signal_strength: int) -> str:
        """
        Determines the signal based on signal strength.
        """
        if signal_strength >= 4:
            signal = "STRONG BUY"
        elif signal_strength == 3:
            signal = "BUY"
        elif signal_strength == 2:
            signal = "WEAK BUY"
        elif signal_strength <= -4:
            signal = "STRONG SELL"
        elif signal_strength == -3:
            signal = "SELL"
        elif signal_strength == -2:
            signal = "WEAK SELL"
        else:
            signal = "NO_SIGNAL"
        return signal

    def determine_market_regime(self, higher_df: pd.DataFrame) -> str:
        """
        Determines the market regime based on higher timeframe trends.
        """
        if higher_df["sma_fast"].iloc[-1] > higher_df["sma_slow"].iloc[-1]:
            return "Bullish"
        elif higher_df["sma_fast"].iloc[-1] < higher_df["sma_slow"].iloc[-1]:
            return "Bearish"
        else:
            return "Neutral"

    def calculate_support_and_resistance(self, df: pd.DataFrame) -> Tuple[float, float]:
        """
        Calculates support and resistance levels as the recent lows and highs.
        """
        try:
            recent_lows = df["low"].rolling(window=20, min_periods=1).min()
            recent_highs = df["high"].rolling(window=20, min_periods=1).max()
            support_level = recent_lows.iloc[-1]
            resistance_level = recent_highs.iloc[-1]
            return support_level, resistance_level
        except Exception as e:
            logger.error(f"Error calculating support and resistance: {str(e)}")
            return (
                df["low"].iloc[-1],
                df["high"].iloc[-1],
            )  # Fallback to last row values in case of error.

    def calculate_stop_loss_take_profit(
        self, signal: str, current_price: float, df: pd.DataFrame
    ) -> Tuple[float, float]:
        """
        Uses ATR to calculate dynamic stop-loss and take-profit levels based on volatility.
        """
        try:
            atr = AverageTrueRange(
                high=df["high"], low=df["low"], close=df["close"], window=14
            ).average_true_range().iloc[-1]
            risk_to_reward_ratio = 1.5  # Desired risk-to-reward ratio

            if signal in ["BUY", "STRONG BUY", "WEAK BUY"]:
                stop_loss = current_price - (atr * 1.5)
                take_profit = current_price + (atr * 1.5 * risk_to_reward_ratio)
            elif signal in ["SELL", "STRONG SELL", "WEAK SELL"]:
                stop_loss = current_price + (atr * 1.5)
                take_profit = current_price - (atr * 1.5 * risk_to_reward_ratio)
            else:
                stop_loss = current_price - (atr * 2)
                take_profit = current_price + (atr * 2)

            return round(stop_loss, 8), round(take_profit, 8)
        except Exception as e:
            logger.error(f"Error calculating stop-loss and take-profit: {e}")
            return current_price, current_price

    def calculate_risk_reward_ratio(
        self, signal: str, price: float, stop_loss: float, take_profit: float
    ) -> float:
        """
        Calculates the risk-reward ratio.
        """
        try:
            if signal in ["BUY", "STRONG BUY", "WEAK BUY"]:
                risk = price - stop_loss
                reward = take_profit - price
            elif signal in ["SELL", "STRONG SELL", "WEAK SELL"]:
                risk = stop_loss - price
                reward = price - take_profit
            else:
                return None
            if risk == 0:
                return None
            ratio = reward / risk
            return round(ratio, 2)
        except Exception as e:
            logger.error(f"Error calculating risk-reward ratio: {e}")
            return None

    @retry_with_exponential_backoff
    async def save_signal(
        self, pair: str, timeframe: str, signal: str, metrics: Dict[str, float], set_number: int
    ):
        """
        Saves the generated signal to the database with detailed information.
        """
        try:
            Signal.objects.create(
                symbol=pair,
                timestamp=metrics.get("timestamp"),
                timeframe=timeframe,
                signal_type=signal,
                price=metrics.get("price"),
                stop_loss=metrics.get("stop_loss"),
                take_profit=metrics.get("take_profit"),
                result=metrics.get("result"),
                win_probability=metrics.get("win_probability"),
                success_rate=metrics.get("success_rate"),
                price_change_24h=metrics.get("price_change_24h"),
                risk_reward_ratio=metrics.get("risk_reward_ratio"),
                market_regime=metrics.get("market_regime"),
                additional_info=metrics.get("additional_info"),
            )
            logger.info(f"Signal for {pair} {timeframe} (Set {set_number}) saved to the database.")
        except Exception as e:
            logger.error(f"Error saving signal for {pair} {timeframe}: {e}")
            raise

    async def run_bot(self):
        """
        Main loop of the bot. It runs continuously, processing symbols asynchronously.
        """
        while True:
            logger.info("Starting bot cycle...")
            start_time = time.time()

            tasks = []
            symbols = TradingPair.objects.values_list("symbol", flat=True)

            for symbol in symbols:
                for timeframe in self.timeframes:
                    task = self.process_symbol(symbol, timeframe)
                    tasks.append(task)

            await asyncio.gather(*tasks)

            end_time = time.time()
            execution_time = end_time - start_time
            sleep_time = max(60 - execution_time, 0)
            logger.info(f"Bot cycle complete. Sleeping for {sleep_time} seconds.")
            await asyncio.sleep(sleep_time)

    async def process_symbol(self, symbol: str, timeframe: str):
        """
        Processes a single symbol and timeframe, generating and saving signals from both sets.
        """
        try:
            # Generate signals for both sets
            signal1, metrics1 = await self.generate_signal_set(
                symbol, timeframe, set_number=1
            )
            signal2, metrics2 = await self.generate_signal_set(
                symbol, timeframe, set_number=2
            )

            tasks = []
            # Save signals from each set independently
            if signal1 != "NO_SIGNAL":
                tasks.append(
                    self.save_signal(symbol, timeframe, signal1, metrics1, set_number=1)
                )
            if signal2 != "NO_SIGNAL":
                tasks.append(
                    self.save_signal(symbol, timeframe, signal2, metrics2, set_number=2)
                )

            # If both sets have the same signal (excluding "NO_SIGNAL"), mark it as Strong and increase success rate
            if signal1 == signal2 and signal1 != "NO_SIGNAL":
                combined_signal = f"STRONG {signal1}" if "STRONG" not in signal1 else signal1
                # Merge metrics for combined signal
                combined_metrics = metrics1.copy()
                combined_metrics["signal"] = combined_signal
                combined_metrics["success_rate"] = 80  # Increase success rate
                combined_metrics["win_probability"] = 80  # Increase win probability
                # Combine reasons from both sets
                combined_metrics["additional_info"]["reasons"] += metrics2["additional_info"]["reasons"]
                tasks.append(
                    self.save_signal(symbol, timeframe, combined_signal, combined_metrics, set_number=0)
                )
                logger.info(
                    f"Confirmed signal for {symbol} ({timeframe}): {combined_signal}"
                )
            else:
                logger.info(
                    f"No confirmed signal for {symbol} ({timeframe}). Signals: Set1={signal1}, Set2={signal2}"
                )

            if tasks:
                await asyncio.gather(*tasks)
            else:
                logger.info(
                    f"No significant signals to save for {symbol} ({timeframe})."
                )
        except Exception as e:
            logger.error(f"Error processing symbol {symbol} ({timeframe}): {e}")

if __name__ == "__main__":
    bot = CryptoBot()
    try:
        asyncio.run(bot.run_bot())
    except KeyboardInterrupt:
        logger.info("Stopping bot...")
Explanation of Changes:

Analyze the code and tell me what would be the accuracy of signals from this bot in percentage based on the features and engineering. What functions are missing in this code and what can we do to improve the accuracy of signals