When I run my code, it doesn't generate signals as it should instead it shows below error
(venv) PS D:\Binance\binancebot> python -m signals.tradlog
2024-10-01 15:41:41.153 | INFO     | __main__:main:509 - Total Futures USDT Pairs: 0
2024-10-01 15:41:41.196 | WARNING  | __main__:main:512 - No USDT pairs found after filtering. Please check the filtering criteria.
2024-10-01 15:41:41.200 | INFO     | __main__:main:562 - Exchange client closed.
I have also included my code below. Can you please take your time to carefully review the code to identify the issue and then modify the code to make it work efficiently and smoothly. Remember to send me complete code after modifications
import asyncio
import datetime
import json
import os
import sys
import threading
import time
import django
import pytz
import websockets
import ccxt.async_support as ccxt  # Use the async support version of ccxt

# Set the event loop policy for Windows
if sys.platform.startswith('win'):
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

# Set the environment variable to point to your project’s settings
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'binancebot.settings')

# Initialize Django
django.setup()

import concurrent.futures
from asgiref.sync import sync_to_async
from concurrent.futures import ProcessPoolExecutor, as_completed
import pandas as pd
import numpy as np
import talib as ta
from django.utils import timezone
from signals.models import Signal  # Adjust this import based on your project structure
from multiprocessing import Process, Event
from loguru import logger
from channels.db import database_sync_to_async
import traceback

# Remove all existing handlers (including the default one)
logger.remove()

# Configure your custom handlers
logger.add("trading_bot.log", rotation="20 MB", retention="2 days", level="DEBUG",
           format="{time} | {level} | {message}")  # File logging
logger.add(sys.stdout, level="DEBUG")  # Console logging

class CombinedSignalGenerator:
    def __init__(self, symbols, timeframes=['5m', '15m', '1h']):
        self.symbols = symbols  # List of symbols as provided by ccxt
        self.timeframes = [tf.lower() for tf in timeframes]
        self.exchange = ccxt.binanceusdm({'enableRateLimit': True})
        # Nested dictionary to hold DataFrames for each symbol and timeframe
        self.dfs = {symbol: {tf: pd.DataFrame() for tf in self.timeframes} for symbol in self.symbols}
        self.data_lock = asyncio.Lock()
        logger.debug(f"Initialized CombinedSignalGenerator with symbols: {self.symbols}, timeframes: {self.timeframes}")

        # Define the directory to store historical data
        self.data_directory = "historical_data"
        # Create the directory if it doesn't exist
        if not os.path.exists(self.data_directory):
            os.makedirs(self.data_directory)

        # Prepare the combined stream URLs for each symbol and timeframe
        self.symbol_map = {}
        self.streams = []

        # Await exchange.load_markets() is called in main() before this class is initialized
        for symbol in self.symbols:
            market = self.exchange.market(symbol)
            ws_symbol = market['id'].lower()  # WebSocket symbols are in lowercase
            self.symbol_map[ws_symbol] = symbol  # Map WebSocket symbol to ccxt symbol

            for timeframe in self.timeframes:
                self.streams.append(f"{ws_symbol}@kline_{timeframe}")

        self.url = f"wss://fstream.binance.com/stream?streams={'/'.join(self.streams)}"

    #def get_stream_symbol(self, symbol):
        # Strip any suffixes and format symbol for the websocket stream
        #symbol = normalize_symbol(symbol)
        #return symbol.replace('/', '').lower()

    async def initialize_exchange(self):
        pass

    async def fetch_historical_data(self):
        # Limit concurrency to prevent rate limiting
        semaphore = asyncio.Semaphore(3)
        tasks = []
        for symbol in self.symbols:
            for timeframe in self.timeframes:
                tasks.append(self.fetch_symbol_historical_data(symbol, timeframe, semaphore))
        await asyncio.gather(*tasks)

    async def fetch_symbol_historical_data(self, symbol, timeframe, semaphore):
        async with semaphore:
            try:
                # Define a filename for caching based on the symbol and timeframe
                filename = f"historical_data_{symbol.replace('/', '').replace(':', '')}_{timeframe}.csv"
                filepath = os.path.join(self.data_directory, filename)

                # Check if the file exists
                if os.path.exists(filepath):
                    # Load the existing data
                    df = pd.read_csv(filepath, parse_dates=['timestamp'])
                    if df['timestamp'].dt.tz is None:
                        # Localize to 'UTC' and convert to 'Africa/Nairobi'
                        df['timestamp'] = df['timestamp'].dt.tz_localize('UTC').dt.tz_convert('Africa/Nairobi')
                    else:
                        df['timestamp'] = df['timestamp'].dt.tz_convert('Africa/Nairobi')

                    logger.info(f"Loaded historical data for {symbol} {timeframe} from {filename}")
                    # Get the most recent timestamp
                    last_timestamp = df['timestamp'].max()
                    # Fetch data from the last timestamp onward
                    since = int(last_timestamp.tz_convert('UTC').timestamp() * 1000) + 1
                    logger.info(f"Fetching new data for {symbol} {timeframe} since {last_timestamp}")
                else:
                    # File doesn't exist, fetch recent historical data
                    logger.info(f"No historical data file found for {symbol} {timeframe}. Fetching recent data.")
                    df = pd.DataFrame()
                    since = None  # Fetch recent data

                # Fetch historical bars
                limit = None  # Fetch as much data as possible within API constraints
                bars = await self.exchange.fetch_ohlcv(
                    symbol,
                    timeframe=timeframe,
                    since=since,
                    limit=1500  # Binance allows up to 1500
                )

                if bars:
                    new_df = pd.DataFrame(bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
                    new_df['timestamp'] = pd.to_datetime(new_df['timestamp'], unit='ms', utc=True)
                    new_df['timestamp'] = new_df['timestamp'].dt.tz_convert('Africa/Nairobi')
                    df = pd.concat([df, new_df], ignore_index=True)
                    df.drop_duplicates(subset='timestamp', keep='last', inplace=True)
                    df.sort_values(by='timestamp', inplace=True)
                    df.reset_index(drop=True, inplace=True)
                    # Keep only the recent data to limit memory usage
                    df = df.tail(2000)  # Adjust as needed
                    # Save the updated data to CSV
                    df.to_csv(filepath, index=False)
                    logger.info(f"Fetched and saved historical data for {symbol} {timeframe} successfully.")
                else:
                    logger.info(f"No new data fetched for {symbol} {timeframe}.")
                self.dfs[symbol][timeframe] = df
            except Exception as e:
                logger.error(f"Error fetching historical data for {symbol} {timeframe}: {e}\n{traceback.format_exc()}")
                self.dfs[symbol][timeframe] = pd.DataFrame()

    async def fetch_realtime_data(self):
        max_retries = 3
        retry_count = 0
        while retry_count < max_retries:
            try:
                async with websockets.connect(self.url) as websocket:
                    logger.info(f"Connected to combined websocket for symbols: {self.symbols}")
                    while True:
                        message = await websocket.recv()
                        data = json.loads(message)
                        kline = data['data']['k']
                        raw_symbol = kline['s']  # Binance symbol, like 'BTCUSDT'
                        symbol_formatted = raw_symbol.lower()
                        symbol = self.symbol_map.get(symbol_formatted)
                        if not symbol:
                            # Skip symbols not in our mapping
                            continue
                        timeframe = kline['i']
                        if timeframe not in self.timeframes:
                            continue  # Skip timeframes we're not tracking
                        is_kline_closed = kline['x']  # 'x' is True when the kline is closed
                        if not is_kline_closed:
                            continue  # Skip if the kline is not closed (still forming)
                        kline_data = {
                            'timestamp': pd.to_datetime(kline['t'], unit='ms', utc=True).tz_convert('Africa/Nairobi'),
                            'open': float(kline['o']),
                            'high': float(kline['h']),
                            'low': float(kline['l']),
                            'close': float(kline['c']),
                            'volume': float(kline['v']),
                        }
                        logger.debug(f"Received kline data for {symbol} {timeframe} at {kline_data['timestamp']}")
                        await self.collect_data(symbol, timeframe, kline_data)
                        await self.generate_and_save_signals(symbol)
            except Exception as e:
                logger.error(f"Websocket connection error: {e}\n{traceback.format_exc()}")
                retry_count += 1
                if retry_count < max_retries:
                    logger.info(f"Retrying websocket connection (Attempt {retry_count + 1}/{max_retries})")
                    await asyncio.sleep(5)  # Wait before retrying
                else:
                    logger.error(f"Max retries reached. Exiting fetch_realtime_data")
                    break

    async def collect_data(self, symbol, timeframe, kline_data):
        async with self.data_lock:
            if symbol not in self.dfs or timeframe not in self.dfs[symbol]:
                logger.warning(f"Symbol {symbol} or timeframe {timeframe} not found in dfs. Skipping update.")
                logger.debug(f"Available symbols: {list(self.dfs.keys())}")
                return

            df = self.dfs[symbol][timeframe]
            # Before adding the new data, check if it's new
            if not df.empty and kline_data['timestamp'] <= df['timestamp'].iloc[-1]:
                logger.debug(f"Received old kline data for {symbol} {timeframe} at {kline_data['timestamp']}. Skipping.")
                return
            # Append new data
            df = pd.concat([df, pd.DataFrame([kline_data])], ignore_index=True)
            df.drop_duplicates(subset='timestamp', keep='last', inplace=True)
            df.sort_values(by='timestamp', inplace=True)
            # Keep only the recent data to limit memory usage
            df = df.tail(2000)  # Adjust as needed
            self.dfs[symbol][timeframe] = df
            logger.debug(f"Collected data for {symbol} {timeframe} at {kline_data['timestamp']}")
            # Save the updated data to CSV
            filename = f"historical_data_{symbol.replace('/', '').replace(':', '')}_{timeframe}.csv"
            filepath = os.path.join(self.data_directory, filename)
            df.to_csv(filepath, index=False)

    async def generate_and_save_signals(self, symbol):
        # For each symbol, we need to analyze each timeframe and perform two analyses
        analysis_results = {}
        for timeframe in self.timeframes:
            if symbol not in self.dfs or timeframe not in self.dfs[symbol]:
                logger.warning(f"Symbol {symbol} or timeframe {timeframe} not found in dfs during signal generation.")
                continue
            df = self.dfs[symbol][timeframe]
            if not df.empty:
                # Perform first analysis
                first_signal_row = self.generate_signals(symbol, timeframe, df, analysis_set='first')
                # Perform second analysis for confirmation
                second_signal_row = self.generate_signals(symbol, timeframe, df, analysis_set='second')
                # Combine the two analyses
                combined_signal_row = self.combine_signal_results(first_signal_row, second_signal_row)
                if combined_signal_row is not None:
                    analysis_results[timeframe] = combined_signal_row
                else:
                    logger.error(f"Signal generation returned None for {symbol} {timeframe}.")
            else:
                logger.error(f"No data fetched for {symbol} {timeframe}, cannot proceed with signal generation.")

        # After analyzing all timeframes, combine the results to determine the final signal
        if analysis_results:
            final_signal = self.determine_final_signal(analysis_results)
            if final_signal:
                await self.save_signal(symbol, final_signal)
            else:
                logger.error(f"Failed to determine final signal for {symbol}.")
        else:
            logger.error(f"No analysis results available for {symbol}.")

    def generate_signals(self, symbol, timeframe, df, analysis_set='first'):
        if df.empty:
            logger.error(f"No data present for signal generation for {symbol} {timeframe}.")
            return None

        try:
            # Ensure the DataFrame is sorted by timestamp
            df = df.sort_values(by='timestamp')
            logger.debug(f"DataFrame for {symbol} {timeframe} has {len(df)} rows, latest timestamp: {df['timestamp'].iloc[-1]}")

            # Use the last N rows for indicator calculations
            recent_df = df.tail(250).copy()

            # Check if there's enough data to compute indicators
            if len(recent_df) < 150:
                logger.warning(f"Not enough data to perform technical analysis for {symbol} {timeframe}.")
                return None

            # First analysis set of indicators
            if analysis_set == 'first':
                # Calculate EMA and SMA
                recent_df['EMA25'] = ta.EMA(recent_df['close'], timeperiod=25)
                recent_df['SMA25'] = ta.SMA(recent_df['close'], timeperiod=25)

                # Calculate Ichimoku Cloud components
                # Tenkan-sen (Conversion Line)
                nine_period_high = recent_df['high'].rolling(window=9).max()
                nine_period_low = recent_df['low'].rolling(window=9).min()
                recent_df['tenkan_sen'] = (nine_period_high + nine_period_low) / 2

                # Kijun-sen (Base Line)
                period26_high = recent_df['high'].rolling(window=26).max()
                period26_low = recent_df['low'].rolling(window=26).min()
                recent_df['kijun_sen'] = (period26_high + period26_low) / 2

                # Senkou Span A (Leading Span A)
                recent_df['senkou_span_a'] = ((recent_df['tenkan_sen'] + recent_df['kijun_sen']) / 2)
                # Senkou Span B (Leading Span B)
                period52_high = recent_df['high'].rolling(window=52).max()
                period52_low = recent_df['low'].rolling(window=52).min()
                recent_df['senkou_span_b'] = (period52_high + period52_low) / 2

                # No future shifts for simplicity
                # Drop NaN values resulting from calculations
                recent_df.dropna(inplace=True)

            # Second analysis set of indicators
            elif analysis_set == 'second':
                # Calculate MACD
                recent_df['macd'], recent_df['macd_signal'], recent_df['macd_hist'] = ta.MACD(
                    recent_df['close'], fastperiod=12, slowperiod=26, signalperiod=9
                )

                # Calculate Stochastic Oscillator
                recent_df['slowk'], recent_df['slowd'] = ta.STOCH(
                    recent_df['high'], recent_df['low'], recent_df['close'],
                    fastk_period=14, slowk_period=3, slowk_matype=0,
                    slowd_period=3, slowd_matype=0
                )

                # Calculate ADX
                recent_df['ADX'] = ta.ADX(recent_df['high'], recent_df['low'], recent_df['close'], timeperiod=14)

                # Drop NaN values resulting from calculations
                recent_df.dropna(inplace=True)
            else:
                logger.error(f"Invalid analysis set specified: {analysis_set}")
                return None

            # Focus on the last data point for signal generation
            last_row = recent_df.iloc[-1]
            previous_row = recent_df.iloc[-2]  # For comparing previous indicator values

            # Signal generation logic based on analysis set
            if analysis_set == 'first':
                # Generate signals based on first set of indicators
                bullish_condition = (
                    (last_row['EMA25'] > last_row['SMA25']) and
                    (previous_row['EMA25'] <= previous_row['SMA25']) and
                    (last_row['close'] > max(last_row['senkou_span_a'], last_row['senkou_span_b']))
                )

                bearish_condition = (
                    (last_row['EMA25'] < last_row['SMA25']) and
                    (previous_row['EMA25'] >= previous_row['SMA25']) and
                    (last_row['close'] < min(last_row['senkou_span_a'], last_row['senkou_span_b']))
                )
            elif analysis_set == 'second':
                # Generate signals based on second set of indicators
                bullish_condition = (
                    (last_row['macd'] > last_row['macd_signal']) and
                    (previous_row['macd'] <= previous_row['macd_signal']) and
                    (last_row['slowk'] < 80 and last_row['slowd'] < 80)  # Not overbought
                )

                bearish_condition = (
                    (last_row['macd'] < last_row['macd_signal']) and
                    (previous_row['macd'] >= previous_row['macd_signal']) and
                    (last_row['slowk'] > 20 and last_row['slowd'] > 20)  # Not oversold
                )

            # Determine the signal
            if bullish_condition:
                signal = 'BUY'
            elif bearish_condition:
                signal = 'SELL'
            else:
                # Possible sideways detection or no signal
                signal = 'NO_SIGNAL'

                # Detect sideways market in first analysis set
                if analysis_set == 'first':
                    # Calculate Bollinger Bands
                    recent_df['upper_band'], recent_df['middle_band'], recent_df['lower_band'] = ta.BBANDS(
                        recent_df['close'], timeperiod=20, nbdevup=2, nbdevdn=2
                    )
                    # Calculate Bandwidth to detect sideways market
                    recent_df['Bandwidth'] = (recent_df['upper_band'] - recent_df['lower_band']) / recent_df['middle_band']
                    # ADX condition
                    if 'ADX' not in recent_df.columns:
                        recent_df['ADX'] = ta.ADX(recent_df['high'], recent_df['low'], recent_df['close'], timeperiod=14)
                    adx_low = last_row['ADX'] < 20
                    bandwidth_low = last_row['Bandwidth'] < 0.05
                    if adx_low and bandwidth_low:
                        signal = 'SIDEWAYS'

            # Assign the signal to the last row
            last_row = last_row.copy()
            last_row['Signal'] = signal

            return last_row.to_frame().T

        except Exception as e:
            logger.error(f"Error generating signals for {symbol} {timeframe}: {e}\n{traceback.format_exc()}")
            return None

    def combine_signal_results(self, first_signal_row, second_signal_row):
        if first_signal_row is None or second_signal_row is None:
            # If either analysis is missing, we cannot confirm the signal
            return None

        # Extract the signals
        first_signal = first_signal_row['Signal'].iloc[0]
        second_signal = second_signal_row['Signal'].iloc[0]

        # Combine the signals
        if first_signal == 'BUY' and second_signal == 'BUY':
            combined_signal = 'VERY_BULLISH'
        elif first_signal == 'BUY':
            combined_signal = 'BULLISH'
        elif first_signal == 'SELL' and second_signal == 'SELL':
            combined_signal = 'VERY_BEARISH'
        elif first_signal == 'SELL':
            combined_signal = 'BEARISH'
        elif first_signal == 'SIDEWAYS':
            combined_signal = 'SIDEWAYS'
        else:
            combined_signal = 'NO_SIGNAL'

        # Create a combined signal row, you may include additional fields as needed
        combined_signal_row = first_signal_row.copy()
        combined_signal_row['Combined_Signal'] = combined_signal

        return combined_signal_row

    def determine_final_signal(self, analysis_results):
        # For each timeframe, get the combined signal
        combined_signals = [result['Combined_Signal'].iloc[0] for result in analysis_results.values()]

        # Prioritize signals
        if 'VERY_BULLISH' in combined_signals:
            final_signal = 'VERY_BULLISH'
        elif 'VERY_BEARISH' in combined_signals:
            final_signal = 'VERY_BEARISH'
        elif 'BULLISH' in combined_signals:
            final_signal = 'BULLISH'
        elif 'BEARISH' in combined_signals:
            final_signal = 'BEARISH'
        elif 'SIDEWAYS' in combined_signals:
            final_signal = 'SIDEWAYS'
        else:
            final_signal = 'NO_SIGNAL'

        # Get the latest timestamp from analysis results for saving
        timestamps = [result['timestamp'].iloc[0] for result in analysis_results.values()]
        latest_timestamp = max(timestamps)

        # Create a final signal row to return
        final_signal_row = analysis_results[self.timeframes[0]].iloc[0].copy()
        final_signal_row['Final_Signal'] = final_signal
        final_signal_row['timestamp'] = latest_timestamp

        return final_signal_row

    async def save_signal(self, symbol, final_signal_row):
        try:
            timestamp = final_signal_row['timestamp']  # Should be timezone-aware in Nairobi time
            run_date = timezone.now()  # Current time when the analysis is performed
            final_signal = final_signal_row['Final_Signal']
            close_price = final_signal_row['close']

            # Display symbol in standard format
            display_symbol = symbol

            # Fetch 24h price change for the symbol
            try:
                ticker = await self.exchange.fetch_ticker(symbol)
                price_change_24h = ticker.get('percentage', 0.0)
                if price_change_24h is None:
                    price_change_24h = 0.0  # Ensure it's a float
            except Exception as e:
                logger.error(f"Error fetching ticker for {symbol}: {e}")
                price_change_24h = 0.0  # Set default value if fetching fails

            await database_sync_to_async(Signal.objects.update_or_create)(
                timestamp=timestamp,
                symbol=display_symbol,
                defaults={
                    'price': close_price,
                    'price_change_24h': price_change_24h,
                    'signal_type': final_signal,
                    'run_date': run_date,
                    # Add any other fields you wish to save
                }
            )
            logger.info(f"Signal saved for {display_symbol} at {timestamp}: {final_signal}")
        except Exception as e:
            logger.error(f"Error saving signals for {symbol} to database: {e}\n{traceback.format_exc()}")

    async def run(self):
        try:
            await self.initialize_exchange()
            await self.fetch_historical_data()
            real_time_task = asyncio.create_task(self.fetch_realtime_data())
            while True:
                # For this implementation, we handle signal generation in fetch_realtime_data after each new data point
                await asyncio.sleep(1)  # Adjust the sleep time as needed
        finally:
            await self.exchange.close()

async def main():
    exchange = ccxt.binanceusdm({'enableRateLimit': True})
    await exchange.load_markets()
    markets = exchange.markets

    # Filter for USDT perpetual futures pairs
    try:
        usdt_pairs = []
        for symbol, market in markets.items():
            # Ensure the market is active and is a perpetual contract
            if market.get('active'):
                if market.get('contract'):
                    if market.get('contractType') == 'perpetual':
                        if market['quote'] == 'USDT':
                            usdt_pairs.append(symbol)
                            logger.debug(f"Added symbol: {symbol}")
        # Remove duplicates just in case
        usdt_pairs = list(set(usdt_pairs))
        logger.info(f"Total Futures USDT Pairs: {len(usdt_pairs)}")

        if not usdt_pairs:
            logger.warning("No USDT pairs found after filtering. Please check the filtering criteria.")
            return  # Exit the function if no symbols found

        # Fetch 24h price change data
        tickers = await exchange.fetch_tickers()
        logger.info(f"Fetched tickers for {len(tickers)} symbols.")

        # Create a list of symbols with their 24h price change percentage
        symbol_changes = []
        for symbol in usdt_pairs:
            ticker = tickers.get(symbol)
            if ticker:
                change_percentage = ticker.get('percentage', 0)
                symbol_changes.append((symbol, abs(change_percentage)))
            else:
                logger.warning(f"No ticker data found for symbol {symbol}")

        if not symbol_changes:
            logger.error("No symbols with ticker data found. Exiting.")
            return

        # Sort symbols by absolute 24h price change percentage in descending order
        symbol_changes.sort(key=lambda x: x[1], reverse=True)

        # Extract sorted symbols
        sorted_symbols = [symbol for symbol, change in symbol_changes]

        symbols = sorted_symbols
        num_groups = 6  # Limit the number of concurrent WebSocket connections
        grouped_symbols = [symbols[i::num_groups] for i in range(num_groups)]

        tasks = []
        for group_symbols in grouped_symbols:
            if not group_symbols:
                continue  # Skip empty groups
            logger.info(f"Processing symbols group: {group_symbols}")
            signal_generator = CombinedSignalGenerator(symbols=group_symbols)
            tasks.append(asyncio.create_task(signal_generator.run()))

        if tasks:
            await asyncio.gather(*tasks)
        else:
            logger.error("No tasks to run. Exiting.")

    except Exception as e:
        logger.error(f"An error occurred: {e}\n{traceback.format_exc()}")

    finally:
        # Ensure we close the exchange client
        await exchange.close()
        logger.info("Exchange client closed.")

if __name__ == "__main__":
    if sys.platform.startswith('win'):
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())