2024-09-17 16:30:10.137 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: TUSD/USDT, timeframe: 1m
2024-09-17 16:30:10.146 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: IOTA/USDT, timeframe: 1m
2024-09-17 16:30:10.146 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: IOTA/USDT, timeframe: 1m
2024-09-17 16:30:10.148 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: XLM/USDT, timeframe: 1m
2024-09-17 16:30:10.148 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: XLM/USDT, timeframe: 1m
2024-09-17 16:30:10.226 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: ONT/USDT, timeframe: 1m
2024-09-17 16:30:10.226 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: TRX/USDT, timeframe: 1m
2024-09-17 16:30:10.226 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: ONT/USDT, timeframe: 1m
2024-09-17 16:30:10.226 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: TRX/USDT, timeframe: 1m
2024-09-17 16:30:10.392 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: ETC/USDT, timeframe: 1m
2024-09-17 16:30:10.392 | DEBUG    | __mp_main__:__init__:39 - Initialized SignalGenerator with symbol: ETC/USDT, timeframe: 1m
2024-09-17 16:30:16.344 | ERROR    | __mp_main__:fetch_historical_data:51 - Error fetching historical data for TUSD/USDT: 
binance does not have market symbol TUSD/USDT
2024-09-17 16:30:16.344 | ERROR    | __mp_main__:fetch_historical_data:51 - Error fetching historical data for TUSD/USDT: 
binance does not have market symbol TUSD/USDT
2024-09-17 16:30:16.383 | ERROR    | __mp_main__:fetch_historical_data:51 - Error fetching historical data for TUSD/USDT: 
binance does not have market symbol TUSD/USDT
2024-09-17 16:30:16.383 | ERROR    | __mp_main__:fetch_historical_data:51 - Error fetching historical data for TUSD/USDT: 
binance does not have market symbol TUSD/USDT
2024-09-17 16:30:16.697 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.697 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.699 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for TRX/USDT      
2024-09-17 16:30:16.699 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for TRX/USDT      
2024-09-17 16:30:16.767 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.767 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.769 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for IOTA/USDT     
2024-09-17 16:30:16.769 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for IOTA/USDT     
2024-09-17 16:30:16.778 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.778 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.780 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for XLM/USDT      
2024-09-17 16:30:16.780 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for XLM/USDT
2024-09-17 16:30:16.788 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.788 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.790 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for ETC/USDT      
2024-09-17 16:30:16.790 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for ETC/USDT      
2024-09-17 16:30:16.811 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.811 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:16.813 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for ONT/USDT      
2024-09-17 16:30:16.813 | INFO     | __mp_main__:historical_data_worker:194 - Processed historical data for ONT/USDT      
2024-09-17 16:30:17.039 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.039 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.099 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.099 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.121 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.121 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.130 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.130 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.145 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.145 | INFO     | __mp_main__:fetch_historical_data:48 - Fetched historical data for {self.symbol} successfully.
2024-09-17 16:30:17.750 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for TUSD/USDT
2024-09-17 16:30:17.750 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for TUSD/USDT
2024-09-17 16:30:18.293 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for TRX/USDT
2024-09-17 16:30:18.293 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for TRX/USDT
2024-09-17 16:30:18.377 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for XLM/USDT
2024-09-17 16:30:18.377 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for ONT/USDT
2024-09-17 16:30:18.377 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for XLM/USDT
2024-09-17 16:30:18.377 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for ONT/USDT
2024-09-17 16:30:18.378 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for ETC/USDT
2024-09-17 16:30:18.379 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for IOTA/USDT
2024-09-17 16:30:18.378 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for ETC/USDT
2024-09-17 16:30:18.379 | INFO     | __mp_main__:fetch_realtime_data:57 - Connected to websocket for IOTA/USDT
I get the above logs when I run my code below. Why do some symbols show error while others are fetched correctly? Also, why is it only running and after connecting to websockets nothing happens? It is not generating signals.
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 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
        self.url = f"wss://stream.binance.com:9443/ws/{self.symbol}@kline_{self.timeframe}"
        self.exchange = ccxt.binance({
            'options': {'defaultType': 'future'}  # Configure for futures markets
        })
        self.df = pd.DataFrame()
        logger.debug(f"Initialized SignalGenerator with symbol: {symbol}, timeframe: {timeframe}")
        

    def fetch_historical_data(self):
        try:
            since = self.exchange.parse8601('2024-07-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("Fetched historical data for {self.symbol} {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}")
            #return pd.DataFrame()
    #async def collect_data(self, kline_data):
        #logging.debug(f"{self.symbol} - Latest: {kline_data}")
        #self.df = self.df.append(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
        await self.fetch_realtime_data()

           #self.generate_signals()
        #df_with_signals = self.generate_signals(self.df)
        #if df_with_signals is not None:
            #latest_signal = df_with_signals.iloc[-1]['Signal']
            #logging.info(f"Real-time signal for {self.symbol}: {latest_signal}")
        #self.save_signals(df_with_signals)
    # Implement your data collection logic here, e.g., append to a DataFrame

    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'))
            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']
           # logging.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

    def save_signals(self, df):
        try:
            last_signal = df.iloc[-1]
            # Save all generated signals to the database
            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):
        await self.combine_data()
        if not self.df.empty:
            self.df = self.generate_signals(self.df)
            self.df = self.save_signals(self.df)
        #df = self.combined_data()
        #if not df.empty:
            #df = self.generate_signals(df)
            #df = self.save_signals(df)
        else:
            logger.error("No data fetched, cannot proceed with signal generation.")
            #stop_event.wait(1)  # Check the event every second

def historical_data_worker(symbol):
        signal_generator = SignalGenerator(symbol=symbol)
        signal_generator.df = signal_generator.fetch_historical_data()
        if not signal_generator.df.empty:
            logger.info(f"Processed historical data for {symbol}")
        try:
            asyncio.run(signal_generator.run())
        except Exception as e:
            logger.error(f"Error running signal processing for {symbol}: {e}")
      

async def main():
    
        exchange = ccxt.binance({'options': {'defaultType': 'future'}})
        markets = exchange.fetch_markets()

# Filter for USDT pairs
        usdt_pairs = [market['symbol'] for market in markets if '/USDT' in market['symbol']]
        
        with ProcessPoolExecutor(max_workers=6) as executor:
            try:
                signal_generators = list(executor.map(
                historical_data_worker,
                usdt_pairs,
                ))
                
            except KeyboardInterrupt:
                logger.info("KeyboardInterrupt received. Shutting down executor…")
                executor.shutdown(wait=True, cancel_futures=True)
                logger.info("Executor has been shut down successfully.")
        

    # Setup tasks for real-time data retrieval
        tasks = []
            #fetch_realtime_data(symbol, f'wss://your.websocket.url/{symbol}', collect_data)
            #for symbol in usdt_pairs
        #]
        for signal_generator in signal_generators:
            tasks.append(signal_generator.fetch_realtime_data())

    # Use asyncio to run the real-time tasks concurrently
        await asyncio.gather(*tasks)

          
if __name__ == "__main__":
    asyncio.run(main())