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 .models import Signal
from multiprocessing import Process, Event
from loguru import logger
from channels.db import database_sync_to_async
# 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:DD-MM-YYYY HH:mm:ss.SSS} | {level} | {message}")  # File logging
logger.add(sys.stdout, level="DEBUG")  # Console logging
# Ensure that the time displayed in logs is in Nairobi time

class CombinedSignalGenerator:
    def __init__(self, symbols, timeframe='5m'):
        self.symbols = symbols
        self.timeframe = timeframe.lower()
        self.exchange = ccxt.binanceusdm({
            'enableRateLimit': True,
        })
        self.dfs = {symbol: pd.DataFrame() for symbol in symbols}
        self.data_lock = asyncio.Lock()
        logger.debug(f"Initialized CombinedSignalGenerator with symbols: {symbols}, timeframe: {timeframe}")

        # Prepare the combined stream URL
        symbol_formatteds = [symbol.replace('/', '').lower() for symbol in symbols]
        self.streams = [f"{symbol_formatted}@kline_{self.timeframe}" for symbol_formatted in symbol_formatteds]
        self.url = f"wss://fstream.binance.com/stream?streams={'/'.join(self.streams)}"

    async def initialize_exchange(self):
        await self.exchange.load_markets()

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

    async def fetch_symbol_historical_data(self, symbol, semaphore):
        async with semaphore:
            try:
                since = self.exchange.parse8601('2024-07-07T00:00:00Z')
                historical_bars = await self.exchange.fetch_ohlcv(
                    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', utc=True)
                df['timestamp'] = df['timestamp'].dt.tz_convert('Africa/Nairobi')
                self.dfs[symbol] = df
                logger.info(f"Fetched historical data for {symbol} successfully.")
            except Exception as e:
                logger.error(f"Error fetching historical data for {symbol}: {e}")
                self.dfs[symbol] = pd.DataFrame()

    async def fetch_realtime_data(self):
        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']
                    symbol = kline['s']
                    # Convert symbol from 'ARPAUSDT' to 'ARPA/USDT'
                    if symbol.endswith('USDT'):
                        base = symbol[:-4]
                        quote = symbol[-4:]
                        symbol = f"{base}/{quote}"
                    else:
                        # Handle other quote currencies if necessary
                        continue  # Skip symbols that are not USDT pairs
                    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']),
                    }
                    await self.collect_data(symbol, kline_data)

        except Exception as e:
            logger.error(f"Websocket connection error for symbols {self.symbols}: {e}")
            await asyncio.sleep(5)  # Wait before retrying

    async def collect_data(self, symbol, kline_data):
        async with self.data_lock:
            df = self.dfs.get(symbol)
            if df is None:
                df = pd.DataFrame()
            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(1000)  # Adjust as needed
            self.dfs[symbol] = df

    async def generate_and_save_signals(self):
        for symbol, df in self.dfs.items():
            if not df.empty:
                last_signal_row = self.generate_signals(symbol, df)
                if last_signal_row is not None:
                    await self.save_signals(symbol, last_signal_row)
                else:
                    logger.error(f"Signal generation returned None for {symbol}.")
            else:
                logger.error(f"No data fetched for {symbol}, cannot proceed with signal generation.")

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

        try:
            # Ensure the DataFrame is sorted by timestamp
            df = df.sort_values(by='timestamp')

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

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

            # Calculate indicators on recent data
            # Calculate EMAs
            recent_df['EMA20'] = ta.EMA(recent_df['close'], timeperiod=20)
            recent_df['EMA50'] = ta.EMA(recent_df['close'], timeperiod=50)

            # Calculate Ichimoku Cloud components without future shifts
            # 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

            # Chikou Span (Lagging Span)
            # Shifted backward by 26 periods
            recent_df['chikou_span'] = recent_df['close'].shift(-26)
            # Drop NaN values resulting from calculations
            recent_df.dropna(inplace=True)

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

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

            # 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

            # Generate signals based on conditions
            bullish_cross = (
                (last_row['EMA20'] > last_row['EMA50']) and
                (previous_row['EMA20'] <= previous_row['EMA50'])
            )

            bearish_cross = (
                (last_row['EMA20'] < last_row['EMA50']) and
                (previous_row['EMA20'] >= previous_row['EMA50'])
            )

            # Price above/below Ichimoku Cloud
            senkou_span_a = last_row['senkou_span_a']
            senkou_span_b = last_row['senkou_span_b']
            # Handle potential NaN values due to shifting
            if pd.isna(senkou_span_a) or pd.isna(senkou_span_b):
                logger.warning(f"Senkou spans contain NaN values after shifting for {symbol}.")
                return None

            max_senkou_span = max(senkou_span_a, senkou_span_b)
            min_senkou_span = min(senkou_span_a, senkou_span_b)

            price_above_cloud = last_row['close'] > max_senkou_span
            price_below_cloud = last_row['close'] < min_senkou_span

            # RSI conditions
            rsi_overbought = last_row['RSI'] > 70
            rsi_oversold = last_row['RSI'] < 30

            # Check for NaN values in indicators
            indicators = ['EMA20', 'EMA50', 'tenkan_sen', 'kijun_sen', 'senkou_span_a', 'senkou_span_b', 'RSI']
            for indicator in indicators:
                if pd.isna(last_row[indicator]) or pd.isna(previous_row[indicator]):
                    logger.warning(f"{indicator} contains NaN values for {symbol}.")

            # Log indicator values
            logger.debug(f"{symbol} Indicators at {last_row['timestamp']}: EMA20={last_row['EMA20']}, EMA50={last_row['EMA50']}, RSI={last_row['RSI']}")
            logger.debug(f"{symbol} Ichimoku at {last_row['timestamp']}: Senkou Span A={last_row['senkou_span_a']}, Senkou Span B={last_row['senkou_span_b']}")

            # Log condition values
            logger.debug(f"{symbol} Conditions at {last_row['timestamp']}: bullish_cross={bullish_cross}, bearish_cross={bearish_cross}, price_above_cloud={price_above_cloud}, price_below_cloud={price_below_cloud}, rsi_overbought={rsi_overbought}, rsi_oversold={rsi_oversold}")

            # Determine signal based on conditions
            if bullish_cross and price_above_cloud and not rsi_overbought:
                signal = 'BUY'
            elif bearish_cross and price_below_cloud and not rsi_oversold:
                signal = 'SELL'
            else:
                signal = 'HOLD'

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

            # Log the generated signal
            logger.info(f"Generated signal for {symbol} at {last_row['timestamp']}: {signal}")

            # Return the last row as a DataFrame to keep consistency
            return last_row.to_frame().T

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

    async def save_signals(self, symbol, last_signal_row):
        try:
            last_signal = last_signal_row.iloc[0]
            timestamp = last_signal['timestamp']  # Should be timezone-aware in Nairobi time
            await database_sync_to_async(Signal.objects.update_or_create)(
                timestamp=timestamp,
                symbol=symbol,
                defaults={
                    'price': last_signal['close'],
                    'signal_type': last_signal['Signal'],
                }
            )
            logger.info(f"Signal saved for {symbol} at {timestamp}: {last_signal['Signal']}")
        except Exception as e:
            logger.error(f"Error saving signals for {symbol} to database: {e}")

    async def run(self):
        await self.initialize_exchange()
        await self.fetch_historical_data()
        asyncio.create_task(self.fetch_realtime_data())
        while True:
            await self.generate_and_save_signals()
            await asyncio.sleep(1)  # Adjust the sleep time as needed

        # Ensure we close the exchange client when done
            await self.exchange.close()

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

    # Filter for USDT pairs
    try:
        usdt_pairs = set()
        for symbol in markets:
            market = markets[symbol]
            if (market.get('contract', False) and
                market.get('linear', False) and
                market.get('expiry') is None):
                base = market['base']
                quote = market['quote']
                # Only consider USDT pairs
                if quote == 'USDT':
                    normalized_symbol = f"{base}/{quote}"
                    usdt_pairs.add(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.")

        # Fetch 24h price change data
        tickers = await exchange.fetch_tickers(list(usdt_pairs))
        # 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)))

        # 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  # Now symbols are sorted based on 24h change
        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:
            logger.info(f"Processing symbols group: {group_symbols}")
            signal_generator = CombinedSignalGenerator(symbols=group_symbols)
            tasks.append(asyncio.create_task(signal_generator.run()))

        await asyncio.gather(*tasks)

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

    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())

Why is my trading bot above just showing the logs below but it doesn't fetch actual data nor does it process it
(venv) PS D:\Binance\binancebot> python -m signals.tradlog
2024-09-20 13:59:49.202 | INFO     | __main__:main:312 - Total Futures USDT Pairs: 308
2024-09-20 13:59:49.558 | INFO     | __main__:main:339 - Processing symbols group: []
2024-09-20 13:59:49.576 | DEBUG    | __main__:__init__:52 - Initialized CombinedSignalGenerator with symbols: [], timeframe: 5m
2024-09-20 13:59:49.577 | INFO     | __main__:main:339 - Processing symbols group: []
2024-09-20 13:59:49.599 | DEBUG    | __main__:__init__:52 - Initialized CombinedSignalGenerator with symbols: [], timeframe: 5m
2024-09-20 13:59:49.601 | INFO     | __main__:main:339 - Processing symbols group: []
2024-09-20 13:59:49.636 | DEBUG    | __main__:__init__:52 - Initialized CombinedSignalGenerator with symbols: [], timeframe: 5m
2024-09-20 13:59:49.637 | INFO     | __main__:main:339 - Processing symbols group: []
2024-09-20 13:59:49.694 | DEBUG    | __main__:__init__:52 - Initialized CombinedSignalGenerator with symbols: [], timeframe: 5m
2024-09-20 13:59:49.705 | INFO     | __main__:main:339 - Processing symbols group: []
2024-09-20 13:59:49.771 | DEBUG    | __main__:__init__:52 - Initialized CombinedSignalGenerator with symbols: [], timeframe: 5m
2024-09-20 13:59:49.772 | INFO     | __main__:main:339 - Processing symbols group: []
2024-09-20 13:59:49.790 | DEBUG    | __main__:__init__:52 - Initialized CombinedSignalGenerator with symbols: [], timeframe: 5m
2024-09-20 13:59:54.641 | INFO     | __main__:fetch_realtime_data:91 - Connected to combined websocket for symbols: []
2024-09-20 13:59:54.732 | INFO     | __main__:fetch_realtime_data:91 - Connected to combined websocket for symbols: []
2024-09-20 13:59:54.783 | INFO     | __main__:fetch_realtime_data:91 - Connected to combined websocket for symbols: []
2024-09-20 13:59:54.785 | INFO     | __main__:fetch_realtime_data:91 - Connected to combined websocket for symbols: []
2024-09-20 13:59:54.788 | INFO     | __main__:fetch_realtime_data:91 - Connected to combined websocket for symbols: []
2024-09-20 13:59:54.792 | INFO     | __main__:fetch_realtime_data:91 - Connected to combined websocket for symbols: []