import asyncio
import datetime
import json
import os
import sys
import threading
import time
import django
import pytz
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

# 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 SignalGenerator:
    def __init__(self, symbol='BTC/USDT', timeframe='5m'):
        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('2024-07-07T00: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', utc=True)
            df['timestamp'] = df['timestamp'].dt.tz_convert('Africa/Nairobi')
            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', 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(kline_data)

            except Exception as e:
                logger.error(f"Websocket connection error for {self.symbol}: {e}")
                await asyncio.sleep(5)  # Wait before retrying
                #continue
            #return pd.DataFrame()
    async def collect_data(self, kline_data):
        async with self.data_lock:
            kline_data['timestamp'] = kline_data['timestamp'].tz_convert('Africa/Nairobi')
            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)
            # Keep only the recent data to limit memory usage
            self.df = self.df.tail(1000)  # Adjust as needed
    
    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:
            last_signal_row = self.generate_signals(self.df)
            if last_signal_row is not None:
                await self.save_signals(last_signal_row)
            else:
                logger.error(f"Signal generation returned None for {self.symbol}.")
        else:
            logger.error("No data fetched, cannot proceed with signal generation.")

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

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

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

        # Check if there's enough data to compute indicators
            if len(recent_df) < 100:
                logger.warning("Not enough data to perform full technical analysis.")
                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).shift(26)
        # 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).shift(26)

            # Shift the cloud spans back to align with current data
            recent_df['senkou_span_a_shifted'] = recent_df['senkou_span_a'].shift(-26)
            recent_df['senkou_span_b_shifted'] = recent_df['senkou_span_b'].shift(-26)

        # Chikou Span (Lagging Span)
        # Shifted backward by 26 periods
            recent_df['chikou_span'] = recent_df['close'].shift(-26)
            

        # 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_shifted']
            senkou_span_b = last_row['senkou_span_b_shifted']
            # Handle potential NaN values due to shifting
            if pd.isna(senkou_span_a) or pd.isna(senkou_span_b):
                logger.warning("Senkou spans contain NaN values after shifting.")
                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 {self.symbol}.")

            # Log indicator values
            logger.debug(f"{self.symbol} Indicators at {last_row['timestamp']}: EMA20={last_row['EMA20']}, EMA50={last_row['EMA50']}, RSI={last_row['RSI']}")
            logger.debug(f"{self.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"{self.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}")

        # Initialize signal
            #signal = 'HOLD' #if used remove else below

        # 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'    #if used remove initialized above

        # Assign the signal to the last row
            last_row['Signal'] = signal

        # Log the generated signal
            logger.info(f"Generated signal for {self.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 {self.symbol}: {e}")
            return None

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

    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 = set()
            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.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.")

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

When I run the above code I get errors as below
19-09-2024 20:00:27.991 | INFO | Fetched historical data for ZK/USDT successfully.
19-09-2024 20:00:27.997 | WARNING | Senkou spans contain NaN values after shifting.
19-09-2024 20:00:27.998 | ERROR | Signal generation returned None for ZK/USDT.
19-09-2024 20:00:28.318 | INFO | Fetched historical data for STG/USDT successfully.
19-09-2024 20:00:28.324 | WARNING | Senkou spans contain NaN values after shifting.
19-09-2024 20:00:28.325 | ERROR | Signal generation returned None for STG/USDT.
19-09-2024 20:00:28.637 | INFO | Fetched historical data for TON/USDT successfully.
19-09-2024 20:00:28.643 | WARNING | Senkou spans contain NaN values after shifting.
19-09-2024 20:00:28.644 | ERROR | Signal generation returned None for TON/USDT.
19-09-2024 20:00:28.961 | INFO | Fetched historical data for APE/USDT successfully.
19-09-2024 20:00:28.967 | WARNING | Senkou spans contain NaN values after shifting.
19-09-2024 20:00:28.967 | ERROR | Signal generation returned None for APE/USDT.
19-09-2024 20:00:29.276 | INFO | Fetched historical data for REEF/USDT successfully.
19-09-2024 20:00:29.282 | WARNING | Senkou spans contain NaN values after shifting.
19-09-2024 20:00:29.282 | ERROR | Signal generation returned None for REEF/USDT.
19-09-2024 20:00:29.590 | INFO | Fetched historical data for EGLD/USDT successfully.
19-09-2024 20:00:29.596 | WARNING | Senkou spans contain NaN values after shifting.
19-09-2024 20:00:29.597 | ERROR | Signal generation returned None for EGLD/USDT.