Please review my code below, why is it processing symbols in duplicates? I see in the logs it process btcusdt twice, bnbusdt twice, etc
import asyncio
import json
import os
import sys
import threading
import time
import django
import websockets
# 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 ccxt
import pandas as pd
import numpy as np
import talib as ta
from django.utils import timezone
from .models import Signal
from multiprocessing import Process, Event
from loguru import logger

logger.add("trading_bot.log", rotation="20 MB", retention="2 days", level="DEBUG")  # File logging
logger.add(sys.stdout, level="DEBUG")  # Console logging


class SignalGenerator:
    def __init__(self, symbol='BTC/USDT', timeframe='1m'):
        self.symbol = symbol
        self.timeframe = timeframe.lower()
        # For perpetual futures on Binance Futures
        symbol_formatted = self.symbol.replace('/', '').lower()
        self.url = f"wss://fstream.binance.com/ws/{symbol_formatted}@kline_{self.timeframe}"
        ##self.url = f"wss://stream.binance.com:9443/ws/{self.symbol}@kline_{self.timeframe}"
        self.exchange = ccxt.binanceusdm({
            'enableRateLimit': True,
        })
        self.exchange.load_markets()
        self.df = pd.DataFrame()
        self.data_lock = asyncio.Lock()
        logger.debug(f"Initialized SignalGenerator with symbol: {symbol}, timeframe: {timeframe}")
        

    def fetch_historical_data(self):
        try:
            since = self.exchange.parse8601('2023-01-01T00:00:00Z')
            historical_bars = self.exchange.fetch_ohlcv(
                self.symbol,
                self.timeframe,
                since=since,
            )
            df = pd.DataFrame(historical_bars, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
            logger.info(f"Fetched historical data for {self.symbol} successfully.")
            return df
        except Exception as e:
            logger.error(f"Error fetching historical data for {self.symbol}: {e}")
            return pd.DataFrame()

    async def fetch_realtime_data(self):
        try:
            async with websockets.connect(self.url) as websocket:
                logger.info(f"Connected to websocket for {self.symbol}")
                while True:
                    message = await websocket.recv()
                    data = json.loads(message)

                    # Extract kline data
                    kline = data['k']
                    kline_data = {
                        'timestamp': pd.to_datetime(kline['t'], unit='ms'),
                        'open': float(kline['o']),
                        'high': float(kline['h']),
                        'low': float(kline['l']),
                        'close': float(kline['c']),
                        'volume': float(kline['v']),
                    }
                    await self.collect_data(kline_data)

        except Exception as e:
            logger.error(f"Websocket connection error for {self.symbol}: {e}")
            await asyncio.sleep(5)  # Wait before retrying
            #return pd.DataFrame()
    async def collect_data(self, kline_data):
        async with self.data_lock:
            self.df = pd.concat([self.df, pd.DataFrame([kline_data])], ignore_index=True)
            self.df.drop_duplicates(subset='timestamp', keep='last', inplace=True)
            self.df.sort_values(by='timestamp', inplace=True)
    
    async def combine_data(self):
        # Fetch historical data
        self.df = self.fetch_historical_data()
        # Start real-time data gathering
        asyncio.create_task(self.fetch_realtime_data())
        # Now proceed to generate signals periodically
        while True:
            await self.generate_and_save_signals()
            await asyncio.sleep(1)  # Adjust the sleep time as needed

    async def generate_and_save_signals(self):
        if not self.df.empty:
            self.df = self.generate_signals(self.df)
            await self.save_signals(self.df)
        else:
            logger.error("No data fetched, cannot proceed with signal generation.")


        
    def generate_signals(self, df):
        if self.df.empty:
            logger.error("No data present for signal generation.")
            return
        try:
            df = self.df.copy()

            # Ensure there is enough data to calculate indicators
            if len(df) < 50:
                logger.warning("Not enough data to perform full technical analysis.")
                return
            # Calculate EMAs
            df['EMA20'] = ta.EMA(df['close'], timeperiod=20)
            df['EMA50'] = ta.EMA(df['close'], timeperiod=50)

            # Calculate Ichimoku Cloud components
            nine_period_high = df['high'].rolling(window=9).max()
            nine_period_low = df['low'].rolling(window=9).min()
            df['tenkan_sen'] = (nine_period_high + nine_period_low) / 2

            period26_high = df['high'].rolling(window=26).max()
            period26_low = df['low'].rolling(window=26).min()
            df['kijun_sen'] = (period26_high + period26_low) / 2

            df['senkou_span_a'] = ((df['tenkan_sen'] + df['kijun_sen']) / 2).shift(26)
            df['senkou_span_b'] = ((df['high'].rolling(window=52).max() + df['low'].rolling(window=52).min()) / 2).shift(26)

            df['chikou_span'] = df['close'].shift(-26)

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

            # Calculate RSI
            df['RSI'] = ta.RSI(df['close'], timeperiod=14)

            # Generate signals
            bullish_cross = (df['EMA20'] > df['EMA50']) & (df['EMA20'].shift() <= df['EMA50'].shift())
            bearish_cross = (df['EMA20'] < df['EMA50']) & (df['EMA20'].shift() >= df['EMA50'].shift())

            price_above_cloud = df['close'] > df[['senkou_span_a', 'senkou_span_b']].max(axis=1)
            price_below_cloud = df['close'] < df[['senkou_span_a', 'senkou_span_b']].min(axis=1)

            rsi_overbought = df['RSI'] > 70
            rsi_oversold = df['RSI'] < 30

            # Define buy and sell conditions
            conditions_for_buy = (bullish_cross & price_above_cloud & ~rsi_overbought)
            conditions_for_sell = (bearish_cross & price_below_cloud & ~rsi_oversold)
            df['Signal'] = np.where(conditions_for_buy, 'BUY', np.where(conditions_for_sell, 'SELL', 'HOLD'))
            #logger.debug(f"Generated signal for {self.symbol}: {last_signal}")
            return df
            #df['Signal'] = 'HOLD'
            #df.loc[bullish_cross & price_above_cloud & ~rsi_overbought, 'Signal'] = 'BUY'
            #df.loc[bearish_cross & price_below_cloud & ~rsi_oversold, 'Signal'] = 'SELL'

            #last_signal = df.iloc[-1]['Signal']
            #logger.debug(f"Generated signal for {self.symbol}: {last_signal}")
            #self.save_signals(df)

        except Exception as e:
            logger.error(f"Error generating signals for {self.symbol}: {e}")
            return df

    async def save_signals(self, df):
        try:
            last_signal = df.iloc[-1]
            # Wrap the ORM call with sync_to_async
            await sync_to_async(Signal.objects.update_or_create)(
                timestamp=timezone.make_aware(last_signal['timestamp']),
                symbol=self.symbol,
                defaults={
                    'price': last_signal['close'],
                    'signal_type': last_signal['Signal'],
                }
            )
            logger.info(f"Signal saved for {self.symbol}: {last_signal['Signal']}")
            return df
        except Exception as e:
            logger.error(f"Error saving signals for {self.symbol} to database: {e}")
            return df

    async def run(self):
        try:
            await self.combine_data()
            if not self.df.empty:
                self.df = self.generate_signals(self.df)
                await self.save_signals(self.df)
        
            else:
                logger.error("No data fetched, cannot proceed with signal generation.")
            #stop_event.wait(1)  # Check the event every second
        except asyncio.CancelledError:
            logger.info(f"Task for {self.symbol} has been cancelled.")
        # Clean up resources if necessary
            raise      

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


# Filter for USDT pairs
        try:
            usdt_pairs = []
            for symbol, market in markets.items():
                if (market['symbol'].endswith(':USDT') and
                    market.get('contract', False) and
                    market.get('linear', False) and
                    market.get('expiry') is None):
                # Normalize the symbol by removing ':USDT'
                    normalized_symbol = market['symbol'].replace(':USDT', '')
                    usdt_pairs.append(normalized_symbol)

            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.")

            tasks = []
            for symbol in usdt_pairs:
                logger.info(f"Processing symbol: {symbol}")
                signal_generator = SignalGenerator(symbol=symbol)
                tasks.append(asyncio.create_task(signal_generator.run()))

            await asyncio.gather(*tasks)


        except KeyboardInterrupt:
            logger.info("KeyboardInterrupt received. Shutting down executor…")

    # Setup tasks for real-time data retrieval
          
if __name__ == "__main__":
    asyncio.run(main())