import asyncio
import logging
from logging.handlers import RotatingFileHandler
from datetime import datetime, timedelta
import json
from typing import List, Dict, Any
import pandas as pd
from decimal import Decimal
import time, os, django
from django.db import transaction
from django.utils import timezone
from zoneinfo import ZoneInfo
from asgiref.sync import sync_to_async
from binance.client import Client
from binance.exceptions import BinanceAPIException
import aiohttp
from concurrent.futures import ThreadPoolExecutor
import backoff
import sys

# Initialize Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'comber.settings')
django.setup()

# Configure logging
def setup_logging():
    """
    Set up comprehensive logging with both console and file handlers.
    """
    # Create logs directory if it doesn't exist
    log_dir = 'logs'
    os.makedirs(log_dir, exist_ok=True)

    # Create a logger
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    # Console Handler
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.INFO)
    console_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    console_handler.setFormatter(console_formatter)

    # File Handler with log rotation
    file_handler = RotatingFileHandler(
        os.path.join(log_dir, 'fetch_data.log'), 
        maxBytes=10*1024*1024,  # 10 MB
        backupCount=5
    )
    file_handler.setLevel(logging.INFO)
    file_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(file_formatter)

    # Add handlers to logger
    logger.addHandler(console_handler)
    logger.addHandler(file_handler)

    return logger

# Setup logging
logger = setup_logging()

# Constants
BASE_URL = "https://fapi.binance.com"
WS_URL = "wss://fstream.binance.com"
NAIROBI_TIMEZONE = ZoneInfo('Africa/Nairobi')
SYMBOLS_PER_BATCH = 10  # Limit to 10 symbols per batch as requested
TIMEFRAMES = ['1m', '5m', '15m', '1h']
MAX_HISTORICAL_FETCH_LIMIT = 1000  # Binance API limit

class BinanceDataFetcher:
    def __init__(self):
        self.client = Client()
        self.session = None
        self.all_symbols = []
        self.websocket_connections = {}

    @staticmethod
    def make_nairobi_time(timestamp):
        """Convert timestamp to timezone-aware Nairobi time."""
        if timestamp > 1e12:  # Check if timestamp is in milliseconds
            timestamp /= 1000
        return datetime.fromtimestamp(timestamp, tz=NAIROBI_TIMEZONE)

    async def initialize(self):
        """Initialize the data fetcher and load all USDT futures pairs."""
        self.session = aiohttp.ClientSession()
        try:
            async with self.session.get(f"{BASE_URL}/fapi/v1/exchangeInfo") as response:
                if response.status != 200:
                    raise Exception(f"HTTP {response.status}: {await response.text()}")
                exchange_info = await response.json()
                self.all_symbols = [
                    s['symbol'] for s in exchange_info['symbols'] 
                    if s['symbol'].endswith('USDT') and s['status'] == 'TRADING'
                ]
                logger.info(f"Total loaded USDT futures pairs: {len(self.all_symbols)}")
        except Exception as e:
            logger.error(f"Error initializing Binance connection: {e}")
            raise

    async def fetch_historical_data(self):
        """Fetch historical data for all symbols concurrently with enhanced logging."""
        # Create batches of symbols
        symbol_batches = [
            self.all_symbols[i:i+SYMBOLS_PER_BATCH] 
            for i in range(0, len(self.all_symbols), SYMBOLS_PER_BATCH)
        ]

        logger.info(f"Starting historical data fetch for {len(self.all_symbols)} symbols")
        logger.info(f"Will process in {len(symbol_batches)} batches of {SYMBOLS_PER_BATCH} symbols each")

        # Create tasks for each batch
        historical_tasks = []
        for batch_num, batch in enumerate(symbol_batches, 1):
            logger.info(f"Preparing batch {batch_num}: Symbols {batch}")
            historical_tasks.append(self.fetch_batch_historical_klines(batch, batch_num))

        # Execute all historical fetch tasks
        await asyncio.gather(*historical_tasks)
        
        logger.info("Completed historical data fetching for all symbols")

    async def fetch_batch_historical_klines(self, symbols: List[str], batch_num: int):
        """Fetch historical klines for a batch of symbols with enhanced logging."""
        from rajj.models import Kline

        batch_tasks = []
        for symbol in symbols:
            for interval in TIMEFRAMES:
                task = self.fetch_symbol_interval_klines(symbol, interval, batch_num)
                batch_tasks.append(task)

        logger.info(f"Batch {batch_num}: Initiating historical kline fetch for {len(symbols)} symbols across {len(TIMEFRAMES)} timeframes")
        
        results = await asyncio.gather(*batch_tasks, return_exceptions=True)
        
        # Log any exceptions
        for result in results:
            if isinstance(result, Exception):
                logger.error(f"Exception in batch processing: {result}")

        logger.info(f"Batch {batch_num}: Completed historical kline fetch")

    async def fetch_symbol_interval_klines(self, symbol: str, interval: str, batch_num: int):
        """Fetch klines for a specific symbol and interval with advanced logging."""
        from rajj.models import Kline

        logger.info(f"Batch {batch_num}: Starting kline fetch for {symbol} at {interval} interval")

        try:
            # Get the last timestamp from the database
            last_timestamp = await self.get_last_kline_timestamp(symbol, interval)
            
            # Determine start time for fetching data
            if last_timestamp:
                start_time = int(last_timestamp.timestamp() * 1000) + 1
                logger.info(f"{symbol} {interval}: Fetching data since last recorded timestamp {last_timestamp}")
            else:
                # If no data exists, fetch data for the last 7 days
                end_time = int(time.time() * 1000)
                start_time = end_time - (7 * 24 * 60 * 60 * 1000)
                logger.info(f"{symbol} {interval}: No previous data found. Fetching last 7 days")

            # Track number of klines saved
            klines_saved = 0

            # Fetch klines
            async with self.session.get(f"{BASE_URL}/fapi/v1/klines", params={
                'symbol': symbol,
                'interval': interval,
                'startTime': start_time,
                'limit': MAX_HISTORICAL_FETCH_LIMIT
            }) as response:
                if response.status != 200:
                    error_text = await response.text()
                    logger.error(f"{symbol} {interval}: HTTP {response.status} - {error_text}")
                    raise Exception(f"HTTP {response.status}: {error_text}")
                
                klines = await response.json()

            # Process and save klines
            for kline in klines:
                kline_data = {
                    'symbol': symbol,
                    'timeframe': interval,
                    'timestamp': self.make_nairobi_time(kline[0]),
                    'open': Decimal(str(kline[1])),
                    'high': Decimal(str(kline[2])),
                    'low': Decimal(str(kline[3])),
                    'close': Decimal(str(kline[4])),
                    'volume': Decimal(str(kline[5]))
                }
                
                # Use get_or_create to avoid duplicates
                _, created = await sync_to_async(Kline.objects.get_or_create)(
                    symbol=symbol,
                    timeframe=interval,
                    timestamp=kline_data['timestamp'],
                    defaults=kline_data
                )
                
                if created:
                    klines_saved += 1

            logger.info(f"Batch {batch_num}: {symbol} {interval} - Processed {len(klines)} klines, saved {klines_saved} new entries")

        except Exception as e:
            logger.error(f"Batch {batch_num}: Error fetching/saving klines for {symbol} {interval}: {e}")

    async def fetch_and_save_initial_orderbooks(self):
        """Fetch and save initial orderbooks for all symbols."""
        from rajj.models import OrderBook

        # Create batches of symbols
        symbol_batches = [
            self.all_symbols[i:i+SYMBOLS_PER_BATCH] 
            for i in range(0, len(self.all_symbols), SYMBOLS_PER_BATCH)
        ]

        # Create tasks for each batch
        orderbook_tasks = []
        for batch in symbol_batches:
            orderbook_tasks.append(self.fetch_batch_orderbooks(batch))

        # Execute all orderbook fetch tasks
        await asyncio.gather(*orderbook_tasks)

    async def fetch_batch_orderbooks(self, symbols: List[str]):
        """Fetch orderbooks for a batch of symbols."""
        from rajj.models import OrderBook

        for symbol in symbols:
            try:
                # Check if orderbook already exists for this symbol
                latest_orderbook = await sync_to_async(OrderBook.objects.filter(
                    symbol=symbol
                ).order_by('-timestamp').first)()

                # Fetch current orderbook
                async with self.session.get(f"{BASE_URL}/fapi/v1/depth", params={
                    'symbol': symbol,
                    'limit': 20  # Adjust limit as needed
                }) as response:
                    if response.status != 200:
                        logger.error(f"Failed to fetch orderbook for {symbol}")
                        continue

                    data = await response.json()
                    bids = data.get('bids', [])
                    asks = data.get('asks', [])

                    # Calculate spread
                    if bids and asks:
                        first_bid_price = Decimal(str(bids[0][0]))
                        first_ask_price = Decimal(str(asks[0][0]))
                        spread = first_ask_price - first_bid_price
                        spread_percentage = (spread / first_ask_price) * 100

                        # Use Nairobi time
                        current_time = timezone.now().astimezone(NAIROBI_TIMEZONE)

                        orderbook_data = {
                            'symbol': symbol,
                            'timestamp': current_time,
                            'last_updated': current_time,
                            'bids': json.dumps(bids),
                            'asks': json.dumps(asks),
                            'spread': spread,
                            'spread_percentage': spread_percentage,
                            'total_bid_volume': sum(Decimal(str(bid[1])) for bid in bids),
                            'total_ask_volume': sum(Decimal(str(ask[1])) for ask in asks)
                        }

                        # Create orderbook entry if it doesn't exist or is significantly different
                        await sync_to_async(OrderBook.objects.create)(**orderbook_data)

            except Exception as e:
                logger.error(f"Error processing orderbook for {symbol}: {e}")

    async def connect_websockets(self):
        """
        Establish WebSocket connections in batches for all symbols.
        """
        # Create batches of symbols
        symbol_batches = [
            self.all_symbols[i:i+SYMBOLS_PER_BATCH] 
            for i in range(0, len(self.all_symbols), SYMBOLS_PER_BATCH)
        ]

        # Create WebSocket connection tasks for each batch
        websocket_tasks = []
        for batch in symbol_batches:
            websocket_tasks.append(self.create_websocket_connection(batch))

        # Run WebSocket connections concurrently
        await asyncio.gather(*websocket_tasks)

    async def create_websocket_connection(self, symbols: List[str]):
        """Create a WebSocket connection for a batch of symbols."""
        # Prepare streams for the batch of symbols
        streams = []
        for symbol in symbols:
            # Add kline streams for each timeframe
            streams.extend([
                f"{symbol.lower()}@kline_{interval}" for interval in TIMEFRAMES
            ])
            # Add depth stream
            streams.append(f"{symbol.lower()}@depth")
        
        # Combine streams into a single WebSocket connection
        ws_url = f"{WS_URL}/stream?streams={'/'.join(streams)}"
        
        try:
            async with aiohttp.ClientSession().ws_connect(ws_url) as ws:
                logger.info(f"WebSocket connected for {len(symbols)} symbols")
                
                async for msg in ws:
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        await self.handle_websocket_message(json.loads(msg.data))
                    elif msg.type == aiohttp.WSMsgType.CLOSED:
                        logger.warning(f"WebSocket closed for batch {symbols}")
                        break
                    elif msg.type == aiohttp.WSMsgType.ERROR:
                        logger.error(f"WebSocket error for batch {symbols}")
                        break
        except Exception as e:
            logger.error(f"WebSocket connection error for batch {symbols}: {e}")

    async def handle_websocket_message(self, msg):
        """Handle incoming websocket messages."""
        try:
            # Extract the actual data from the WebSocket message
            data = msg.get('data', {})
            
            # Determine message type
            event_type = data.get('e')
            
            if event_type == 'kline':
                await self.process_kline_message(data)
            elif event_type == 'depthUpdate':
                await self.process_depth_message(data)
        except Exception as e:
            logger.error(f"Error processing websocket message: {e}")

    async def process_kline_message(self, msg):
        """Process kline websocket message."""
        from rajj.models import Kline
        
        try:
            kline = msg.get('k', {})
            symbol = msg.get('s')
            
            if not kline or not symbol:
                return
            
            kline_data = {
                'symbol': symbol,
                'timeframe': kline.get('i'),
                'timestamp': self.make_nairobi_time(kline.get('t')),
                'open': Decimal(str(kline.get('o', 0))),
                'high': Decimal(str(kline.get('h', 0))),
                'low': Decimal(str(kline.get('l', 0))),
                'close': Decimal(str(kline.get('c', 0))),
                'volume': Decimal(str(kline.get('v', 0)))
            }
            
            # Ensure all required data is present
            if all(kline_data.values()):
                await sync_to_async(Kline.objects.get_or_create)(
                    symbol=kline_data['symbol'],
                    timeframe=kline_data['timeframe'],
                    timestamp=kline_data['timestamp'],
                    defaults=kline_data
                )
            
        except Exception as e:
            logger.error(f"Error processing kline message: {e}")

    async def process_depth_message(self, msg):
        """Process depth (orderbook) websocket message."""
        from rajj.models import OrderBook
        
        try:
            symbol = msg.get('s')
            bids = msg.get('b', [])
            asks = msg.get('a', [])
            
            if not bids or not asks:
                return
            
            # Convert bid and ask prices to decimals
            first_bid_price = Decimal(str(bids[0][0]))
            first_ask_price = Decimal(str(asks[0][0]))
            spread = first_ask_price - first_bid_price
            spread_percentage = (spread / first_ask_price) * 100
            
            # Use Nairobi time
            current_time = timezone.now().astimezone(NAIROBI_TIMEZONE)
            
            orderbook_data = {
                'symbol': symbol,
                'timestamp': current_time,
                'last_updated': current_time,
                'bids': json.dumps(bids),
                'asks': json.dumps(asks),
                'spread': spread,
                'spread_percentage': spread_percentage,
                'total_bid_volume': sum(Decimal(str(bid[1])) for bid in bids),
                'total_ask_volume': sum(Decimal(str(ask[1])) for ask in asks)
            }
            
            await sync_to_async(OrderBook.objects.create)(**orderbook_data)
            
        except Exception as e:
            logger.error(f"Error processing depth message: {e}")

    async def get_last_kline_timestamp(self, symbol: str, interval: str):
        """
        Retrieve the last kline timestamp for a specific symbol and interval.
        Returns None if no klines exist.
        """
        from rajj.models import Kline
        last_kline = await sync_to_async(Kline.objects.filter(
            symbol=symbol, 
            timeframe=interval
        ).order_by('-timestamp').first)()
        
        return last_kline.timestamp if last_kline else None

    async def run(self):
        """Main execution method."""
        try:
            # Initialize and load all symbols
            await self.initialize()
            
            # Fetch historical klines concurrently
            logger.info("Fetching historical klines...")
            await self.fetch_historical_data()
            
            # Fetch and save initial orderbooks
            logger.info("Fetching initial orderbooks...")
            await self.fetch_and_save_initial_orderbooks()
            
            # Connect to WebSockets
            logger.info("Connecting to WebSockets...")
            await self.connect_websockets()
        
        except Exception as e:
            logger.error(f"Error in main execution: {e}")

    async def cleanup(self):
        """Cleanup resources."""
        if self.session:
            await self.session.close()

if __name__ == "__main__":
    fetcher = BinanceDataFetcher()
    try:
        asyncio.run(fetcher.run())
    except KeyboardInterrupt:
        logger.info("Shutting down...")
        asyncio.run(fetcher.cleanup())
    except Exception as e:
        logger.error(f"Fatal error: {e}")
        asyncio.run(fetcher.cleanup())

The code above is supposed to fetch and store data as well as connecting to websockets. When I run it it runs successfully and fetch over 300 symbols. The code below is supposed to use the symbols fetched and use realtime data to perform technical analysis and generate signals. however, when I run the code it says 0 symbols fetched. It only show 
(myenv) PS D:\binbot\comber> python -m rajj.tean
2024-12-03 14:49:18 [INFO] CryptoBot initialized with timeframes: ['1m', '5m', '15m', '1h'] and num_candles: 500
2024-12-03 14:49:18 [INFO] Starting bot cycle...
2024-12-03 14:49:18 [INFO] Fetched 0 symbols.
2024-12-03 14:49:18 [INFO] Bot cycle complete. Sleeping for 59.90007185935974 seconds.

Please analyze the above code for fetching data and analyze the code below for performing technical analysis, and tell me why it is not working and assist in debugging. 

# technical_analysis.py

import asyncio
import logging
import time
from typing import Dict, List, Tuple
from datetime import timedelta
from asgiref.sync import sync_to_async
import pandas as pd
import numpy as np
from ta.momentum import (
    RSIIndicator,
    StochasticOscillator,
)
from ta.trend import (
    ADXIndicator,
    MACD,
    SMAIndicator,
)
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", "comber.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_symbols(self) -> List[str]:
        """
        Asynchronously fetch trading pair symbols.
        """
        symbols = await sync_to_async(list)(
            TradingPair.objects.values_list("symbol", flat=True)
        )
        return [symbol for symbol in symbols if symbol]  # Filter out empty strings

    @retry_with_exponential_backoff
    async def get_candles_df(self, symbol: 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:
            # Use sync_to_async to perform synchronous database query in an async context
            klines = await sync_to_async(list)(
                Kline.objects.filter(symbol=symbol, timeframe=timeframe)
                .order_by("-timestamp")[: self.num_candles]
                .values()
            )
            
            if not klines:
                logger.warning(f"No candles found for {symbol} ({timeframe})")
                return pd.DataFrame()

            df = pd.DataFrame.from_records(klines)
            df["timestamp"] = pd.to_datetime(df["timestamp"])
            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 {symbol} ({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 with advanced volatility and trend indicators.
        """
        try:
        # Stochastic Oscillator
            stoch = StochasticOscillator(
                high=df["high"], low=df["low"], close=df["close"], window=14, smooth_window=3
            )
            df["stoch_k"] = stoch.stoch()
            df["stoch_d"] = stoch.stoch_signal()
        
            # RVI
            df["rvi"] = self.calculate_rvi(df)
        
            # Advanced Trend and Volatility Indicators
            # ADX for trend strength
            adx_indicator = ADXIndicator(
                high=df["high"], low=df["low"], close=df["close"], window=14
            )
            df["adx"] = adx_indicator.adx()
            df["adx_pos"] = adx_indicator.adx_pos()
            df["adx_neg"] = adx_indicator.adx_neg()
        
            # ATR for volatility
            atr = AverageTrueRange(
                high=df["high"], low=df["low"], close=df["close"], window=14
            )
            df["atr"] = atr.average_true_range()
        
            # Volatility Bands (more advanced than Keltner Channels)
            df["upper_volatility_band"] = df["close"] + (df["atr"] * 2)
            df["lower_volatility_band"] = df["close"] - (df["atr"] * 2)
        
            # Directional Movement Index (DMI)
            df["plus_di"] = adx_indicator.adx_pos()
            df["minus_di"] = adx_indicator.adx_neg()
        
        # 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,
        symbol: 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(symbol, 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(symbol, 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(symbol, 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": symbol,
                "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 {symbol} ({timeframe}) Set {set_number}: {signal}"
            )

            return signal, metrics
        except Exception as e:
            logger.error(f"Error generating signal for {symbol} ({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 advanced second set of indicators.
        """
        signal_strength = 0
        reasons = []

        # Stochastic Oscillator
        stoch_k = df["stoch_k"].iloc[-1]
        stoch_d = df["stoch_d"].iloc[-1]
    
    # Advanced trend strength analysis
        adx = df["adx"].iloc[-1]
        plus_di = df["plus_di"].iloc[-1]
        minus_di = df["minus_di"].iloc[-1]
        atr = df["atr"].iloc[-1]

    # Stochastic analysis with more nuanced conditions
        if stoch_k < 20 and stoch_k > stoch_d:
            signal_strength += 2
            reasons.append("Strong bullish Stochastic Oscillator signal in oversold territory.")
        elif stoch_k > 80 and stoch_k < stoch_d:
            signal_strength -= 2
            reasons.append("Strong bearish Stochastic Oscillator signal in overbought territory.")

    # Advanced trend strength evaluation
        if adx > 25:  # Strong trend
            if plus_di > minus_di:
                signal_strength += 2
                reasons.append("Strong bullish trend confirmed by ADX and Directional Indicators.")
            else:
                signal_strength -= 2
                reasons.append("Strong bearish trend confirmed by ADX and Directional Indicators.")
        else:
            reasons.append("Weak trend strength, market might be ranging.")

    # Volatility-based analysis
        if df["close"].iloc[-1] < df["lower_volatility_band"].iloc[-1]:
            signal_strength += 1
            reasons.append("Price near lower volatility band, potential bounce.")
        elif df["close"].iloc[-1] > df["upper_volatility_band"].iloc[-1]:
            signal_strength -= 1
            reasons.append("Price near upper volatility band, potential reversal.")

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

    # Volatility expansion/contraction
        prev_atr = df["atr"].iloc[-2]
        if atr > prev_atr * 1.5:
            signal_strength += 1
            reasons.append("Volatility expansion detected, potential significant move.")
        elif atr < prev_atr * 0.5:
            signal_strength -= 1
            reasons.append("Volatility contraction detected, potential range-bound market.")

    # Higher and lower timeframe confirmations
        higher_adx = higher_df["adx"].iloc[-1]
        higher_plus_di = higher_df["plus_di"].iloc[-1]
        higher_minus_di = higher_df["minus_di"].iloc[-1]

        if higher_adx > 25 and higher_plus_di > higher_minus_di:
            signal_strength += 1
            reasons.append("Higher timeframe trend is strongly bullish.")
        elif higher_adx > 25 and higher_plus_di < higher_minus_di:
            signal_strength -= 1
            reasons.append("Higher timeframe trend is strongly bearish.")

        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, symbol: str, timeframe: str, signal: str, metrics: Dict[str, float], set_number: int
    ):
        """
        Saves the generated signal to the database asynchronously.
        """
        try:
            # Use sync_to_async to create the Signal object
            await sync_to_async(Signal.objects.create)(
                symbol=symbol,
                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 {symbol} {timeframe} (Set {set_number}) saved to the database.")
        except Exception as e:
            logger.error(f"Error saving signal for {symbol} {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()

            try:
                # Fetch symbols asynchronously
                symbols = await self.get_symbols()
                logger.info(f"Fetched {len(symbols)} symbols.")
            except Exception as e:
                logger.error(f"Error fetching symbols: {e}")
                await asyncio.sleep(60)  # Wait before retrying
                continue

            tasks = []
            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...")