import asyncio
import json
import os
import sys
import websockets
import ccxt.async_support as ccxt
import pandas as pd
import traceback
from datetime import datetime
from asgiref.sync import sync_to_async
import django
from loguru import logger

# Initialize Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'binancebot.settings')
django.setup()

from .models import Kline, OrderBook

# Remove default Loguru handlers and configure new ones
logger.remove()
logger.add("fetch_data.log", rotation="20 MB", retention="7 days", level="DEBUG",
           format="{time} | {level} | {message}")  # File logging
logger.add(sys.stdout, level="INFO")  # Console logging

class DataFetcher:
    def __init__(self, symbols, timeframes=['1m', '5m', '15m', '1h']):
        self.symbols = symbols
        self.timeframes = [tf.lower() for tf in timeframes]
        self.exchange = ccxt.binanceusdm({'enableRateLimit': True})
        self.data_lock = asyncio.Lock()
        self.symbol_map = {}
        self.streams = []
        self.stream_urls = []
        self.max_streams_per_connection = 100  # Adjust based on testing
        logger.debug(f"Initialized DataFetcher with symbols: {self.symbols}, timeframes: {self.timeframes}")

    async def initialize_exchange(self):
        await self.exchange.load_markets()
        for symbol in self.symbols:
            market = self.exchange.market(symbol)
            ws_symbol = market['id'].lower()  # e.g., 'btcusdt'
            self.symbol_map[ws_symbol] = symbol

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

            # Add order book streams (assuming depth=5 for this example)
            self.streams.append(f"{ws_symbol}@depth5@100ms")  # Adjust depth and update speed as needed

        # Create WebSocket URLs with limited streams per connection
        for i in range(0, len(self.streams), self.max_streams_per_connection):
            stream_subset = self.streams[i:i + self.max_streams_per_connection]
            url = f"wss://fstream.binance.com/stream?streams={'/'.join(stream_subset)}"
            self.stream_urls.append(url)

        logger.debug(f"WebSocket URLs: {self.stream_urls}")

    async def fetch_historical_data(self):
        semaphore = asyncio.Semaphore(5)  # Adjusted semaphore for optimal concurrency
        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:
                limit = 1000  # Binance allows up to 1000 bars per request

                # Get the most recent timestamp from the database
                latest_timestamp = await self.get_latest_kline_timestamp(symbol, timeframe)
                if latest_timestamp:
                    since = int(latest_timestamp.timestamp() * 1000) + 1  # Convert to milliseconds and add 1ms
                else:
                    since = None  # Fetch all available data

                all_bars = []

                while True:
                    bars = await self.exchange.fetch_ohlcv(symbol, timeframe=timeframe, since=since, limit=limit)
                    if not bars:
                        break
                    all_bars += bars
                    since = bars[-1][0] + 1  # Prevent fetching the last bar again
                    if len(bars) < limit:
                        break

                if all_bars:
                    df = pd.DataFrame(all_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 {len(df)} new bars for {symbol} {timeframe}")

                    # Save to database
                    await self.save_klines(symbol, timeframe, df)
                else:
                    logger.info(f"No new data to fetch for {symbol} {timeframe}")

            except ccxt.NetworkError as e:
                logger.error(f"Network error while fetching data for {symbol} {timeframe}: {str(e)}")
            except ccxt.ExchangeError as e:
                logger.error(f"Exchange error while fetching data for {symbol} {timeframe}: {str(e)}")
            except Exception as e:
                logger.error(f"Unexpected error fetching data for {symbol} {timeframe}: {str(e)}\n{traceback.format_exc()}")

    @sync_to_async
    def get_latest_kline_timestamp(self, symbol, timeframe):
        try:
            latest_kline = Kline.objects.filter(symbol=symbol, timeframe=timeframe).order_by('-timestamp').first()
            if latest_kline:
                return latest_kline.timestamp
            else:
                return None
        except Exception as e:
            logger.error(f"Error getting latest kline timestamp for {symbol} {timeframe}: {str(e)}")
            return None

    @sync_to_async
    def save_klines(self, symbol, timeframe, df):
        # Validate and clean data before saving
        df.dropna(inplace=True)
        if df.empty:
            logger.warning(f"No valid data to save for {symbol} {timeframe}")
            return

        klines_to_create = []
        for _, row in df.iterrows():
            klines_to_create.append(Kline(
                symbol=symbol,
                timeframe=timeframe,
                timestamp=row['timestamp'],
                open=row['open'],
                high=row['high'],
                low=row['low'],
                close=row['close'],
                volume=row['volume']
            ))
        Kline.objects.bulk_create(klines_to_create, ignore_conflicts=True)
        logger.info(f"Saved {len(klines_to_create)} klines for {symbol} {timeframe} to the database.")

    async def handle_websocket(self, url):
        retry_count = 0
        max_retries = 5
        backoff = 1

        while retry_count < max_retries:
            try:
                async with websockets.connect(
                    url,
                    ping_interval=20,
                    ping_timeout=20,
                    close_timeout=10
                ) as websocket:
                    logger.info(f"Connected to WebSocket: {url}")
                    retry_count = 0  # Reset on successful connection

                    async for message in websocket:
                        data = json.loads(message)

                        if 'data' not in data:
                            continue

                        stream_type = data['stream'].split('@')[1]  # Extract stream type (e.g., 'kline_1m', 'depth5')

                        if stream_type.startswith('kline'):
                            await self.process_kline_data(data)
                        elif stream_type.startswith('depth'):
                            await self.process_order_book_data(data)

            except websockets.exceptions.ConnectionClosed as e:
                logger.warning(f"WebSocket connection closed: {str(e)}")
            except websockets.exceptions.InvalidStatusCode as e:
                logger.error(f"Invalid status code when connecting to WebSocket: {str(e)}")
                break
            except Exception as e:
                logger.error(f"Unexpected WebSocket error: {str(e)}\n{traceback.format_exc()}")
                retry_count += 1
                sleep_time = backoff * retry_count
                logger.info(f"Retrying WebSocket connection in {sleep_time} seconds (Attempt {retry_count}/{max_retries})")
                await asyncio.sleep(sleep_time)

    async def process_kline_data(self, data):
        kline_data = data['data']['k']
        if not kline_data['x']:
            return  # Only process closed klines

        ws_symbol = kline_data['s'].lower()
        symbol = self.symbol_map.get(ws_symbol)
        timeframe = kline_data['i']

        if timeframe not in self.timeframes:
            return

        timestamp = pd.to_datetime(kline_data['t'], unit='ms', utc=True).tz_convert('Africa/Nairobi')
        kline = Kline(
            symbol=symbol,
            timeframe=timeframe,
            timestamp=timestamp,
            open=float(kline_data['o']),
            high=float(kline_data['h']),
            low=float(kline_data['l']),
            close=float(kline_data['c']),
            volume=float(kline_data['v'])
        )

        # Save or update the kline in the database
        await self.save_kline(kline)

    async def process_order_book_data(self, data):
        order_book_data = data['data']
        ws_symbol = order_book_data['s'].lower()
        symbol = self.symbol_map.get(ws_symbol)
        timestamp = pd.to_datetime(order_book_data['T'], unit='ms', utc=True).tz_convert('Africa/Nairobi')

        # Process and save order book data
        bids = [[float(price), float(qty)] for price, qty in order_book_data['b']]
        asks = [[float(price), float(qty)] for price, qty in order_book_data['a']]

        order_book = OrderBook(
            symbol=symbol,
            timestamp=timestamp,
            bids=bids,
            asks=asks
        )
        await self.save_order_book(order_book)

    @sync_to_async
    def save_kline(self, kline):
        # Data validation and cleansing
        if any(val is None for val in [kline.open, kline.high, kline.low, kline.close, kline.volume]):
            logger.warning(f"Invalid data encountered for {kline.symbol} {kline.timeframe} at {kline.timestamp}")
            return

        Kline.objects.update_or_create(
            symbol=kline.symbol,
            timeframe=kline.timeframe,
            timestamp=kline.timestamp,
            defaults={
                'open': kline.open,
                'high': kline.high,
                'low': kline.low,
                'close': kline.close,
                'volume': kline.volume,
            }
        )
        logger.debug(f"Saved kline for {kline.symbol} {kline.timeframe} at {kline.timestamp}")

    @sync_to_async
    def save_order_book(self, order_book):
        OrderBook.objects.update_or_create(
            symbol=order_book.symbol,
            timestamp=order_book.timestamp,
            defaults={
                'bids': order_book.bids,
                'asks': order_book.asks,
            }
        )
        logger.debug(f"Saved order book for {order_book.symbol} at {order_book.timestamp}")

    async def start_websockets(self):
        tasks = []
        for url in self.stream_urls:
            tasks.append(asyncio.create_task(self.handle_websocket(url)))
        await asyncio.gather(*tasks)

    async def run(self):
        try:
            logger.info("Initializing exchange...")
            await self.initialize_exchange()
            logger.info("Fetching historical data...")
            await self.fetch_historical_data()
            logger.info("Starting WebSocket connections...")
            await self.start_websockets()
        finally:
            await self.exchange.close()
            logger.info("Exchange connection closed.")

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

    # Filter for USDT perpetual futures pairs
    usdt_pairs = []
    for symbol, market in markets.items():
        if market.get('active') and market.get('contract') and market.get('type') == 'swap' and market.get('linear') and market.get('quote') == 'USDT':
            usdt_pairs.append(symbol)

    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

    fetcher = DataFetcher(symbols=usdt_pairs, timeframes=['1m', '5m', '15m', '1h'])
    await fetcher.run()

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

Between above code and the one below, wic is better

#fetch_data.py
import asyncio
import json
import os
import sys
import websockets
import ccxt.async_support as ccxt
import pandas as pd
import traceback
from datetime import datetime
from asgiref.sync import sync_to_async
import django
from loguru import logger

# Initialize Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'binancebot.settings')
django.setup()

from .models import Kline, OrderBook

# Remove default Loguru handlers and configure new ones
logger.remove()
logger.add("fetch_data.log", rotation="20 MB", retention="7 days", level="DEBUG",
           format="{time} | {level} | {message}")  # File logging
logger.add(sys.stdout, level="INFO")  # Console logging

class DataFetch:
    def __init__(self, symbols, timeframes=['1m', '5m', '15m', '1h']):
        self.symbols = symbols
        self.timeframes = [tf.lower() for tf in timeframes]
        self.exchange = ccxt.binanceusdm({'enableRateLimit': True})
        self.data_lock = asyncio.Lock()
        self.symbol_map = {}
        self.streams = []
        self.stream_urls = []
        self.max_streams_per_connection = 100  # Adjust based on testing
        logger.debug(f"Initialized DataFetcher with symbols: {self.symbols}, timeframes: {self.timeframes}")

    async def initialize_exchange(self):
        await self.exchange.load_markets()
        for symbol in self.symbols:
            market = self.exchange.market(symbol)
            ws_symbol = market['id'].lower()  # e.g., 'btcusdt'
            self.symbol_map[ws_symbol] = symbol

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

            # Add order book streams (assuming depth=5 for this example)
            self.streams.append(f"{ws_symbol}@depth5@100ms")  # Adjust depth and update speed as needed

        # Create WebSocket URLs with limited streams per connection
        for i in range(0, len(self.streams), self.max_streams_per_connection):
            stream_subset = self.streams[i:i + self.max_streams_per_connection]
            url = f"wss://fstream.binance.com/stream?streams={'/'.join(stream_subset)}"
            self.stream_urls.append(url)

        logger.debug(f"WebSocket URLs: {self.stream_urls}")

    async def fetch_historical_data(self):
        semaphore = asyncio.Semaphore(5)  # Adjusted semaphore for optimal concurrency
        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:
                since = None  # Fetch all available data
                all_bars = []
                limit = 1000  # Binance allows up to 1000 bars per request

                while True:
                    bars = await self.exchange.fetch_ohlcv(symbol, timeframe=timeframe, since=since, limit=limit)
                    if not bars:
                        break
                    all_bars += bars
                    since = bars[-1][0] + 1  # Prevent fetching the last bar again
                    if len(bars) < limit:
                        break

                if all_bars:
                    df = pd.DataFrame(all_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 {len(df)} bars for {symbol} {timeframe}")

                    # Save to database
                    await self.save_klines(symbol, timeframe, df)

            except ccxt.NetworkError as e:
                logger.error(f"Network error while fetching data for {symbol} {timeframe}: {str(e)}")
            except ccxt.ExchangeError as e:
                logger.error(f"Exchange error while fetching data for {symbol} {timeframe}: {str(e)}")
            except Exception as e:
                logger.error(f"Unexpected error fetching data for {symbol} {timeframe}: {str(e)}\n{traceback.format_exc()}")

    @sync_to_async
    def save_klines(self, symbol, timeframe, df):
        # Validate and clean data before saving
        df.dropna(inplace=True)
        if df.empty:
            logger.warning(f"No valid data to save for {symbol} {timeframe}")
            return

        klines_to_create = []
        for _, row in df.iterrows():
            klines_to_create.append(Kline(
                symbol=symbol,
                timeframe=timeframe,
                timestamp=row['timestamp'],
                open=row['open'],
                high=row['high'],
                low=row['low'],
                close=row['close'],
                volume=row['volume']
            ))
        Kline.objects.bulk_create(klines_to_create, ignore_conflicts=True)
        logger.info(f"Saved {len(klines_to_create)} klines for {symbol} {timeframe} to the database.")

    async def handle_websocket(self, url):
        retry_count = 0
        max_retries = 5
        backoff = 1

        while retry_count < max_retries:
            try:
                async with websockets.connect(
                    url,
                    ping_interval=20,
                    ping_timeout=20,
                    close_timeout=10
                ) as websocket:
                    logger.info(f"Connected to WebSocket: {url}")
                    retry_count = 0  # Reset on successful connection

                    async for message in websocket:
                        data = json.loads(message)

                        if 'data' not in data:
                            continue

                        stream_type = data['stream'].split('@')[1]  # Extract stream type (e.g., 'kline_1m', 'depth5')

                        if stream_type.startswith('kline'):
                            await self.process_kline_data(data)
                        elif stream_type.startswith('depth'):
                            await self.process_order_book_data(data)

            except websockets.exceptions.ConnectionClosed as e:
                logger.warning(f"WebSocket connection closed: {str(e)}")
            except websockets.exceptions.InvalidStatusCode as e:
                logger.error(f"Invalid status code when connecting to WebSocket: {str(e)}")
                break
            except Exception as e:
                logger.error(f"Unexpected WebSocket error: {str(e)}\n{traceback.format_exc()}")
                retry_count += 1
                sleep_time = backoff * retry_count
                logger.info(f"Retrying WebSocket connection in {sleep_time} seconds (Attempt {retry_count}/{max_retries})")
                await asyncio.sleep(sleep_time)

    async def process_kline_data(self, data):
        kline_data = data['data']['k']
        if not kline_data['x']:
            return  # Only process closed klines

        ws_symbol = kline_data['s'].lower()
        symbol = self.symbol_map.get(ws_symbol)
        timeframe = kline_data['i']

        if timeframe not in self.timeframes:
            return

        timestamp = pd.to_datetime(kline_data['t'], unit='ms', utc=True).tz_convert('Africa/Nairobi')
        kline = Kline(
            symbol=symbol,
            timeframe=timeframe,
            timestamp=timestamp,
            open=float(kline_data['o']),
            high=float(kline_data['h']),
            low=float(kline_data['l']),
            close=float(kline_data['c']),
            volume=float(kline_data['v'])
        )

        # Save or update the kline in the database
        await self.save_kline(kline)

    async def process_order_book_data(self, data):
        order_book_data = data['data']
        ws_symbol = order_book_data['s'].lower()
        symbol = self.symbol_map.get(ws_symbol)
        timestamp = pd.to_datetime(order_book_data['T'], unit='ms', utc=True).tz_convert('Africa/Nairobi')

        # Process and save order book data
        bids = [[float(price), float(qty)] for price, qty, _ in order_book_data['b']]
        asks = [[float(price), float(qty)] for price, qty, _ in order_book_data['a']]

        order_book = OrderBook(
            symbol=symbol,
            timestamp=timestamp,
            bids=bids,
            asks=asks
        )
        await self.save_order_book(order_book)

    @sync_to_async
    def save_kline(self, kline):
        # Data validation and cleansing
        if any(val is None for val in [kline.open, kline.high, kline.low, kline.close, kline.volume]):
            logger.warning(f"Invalid data encountered for {kline.symbol} {kline.timeframe} at {kline.timestamp}")
            return

        Kline.objects.update_or_create(
            symbol=kline.symbol,
            timeframe=kline.timeframe,
            timestamp=kline.timestamp,
            defaults={
                'open': kline.open,
                'high': kline.high,
                'low': kline.low,
                'close': kline.close,
                'volume': kline.volume,
            }
        )
        logger.debug(f"Saved kline for {kline.symbol} {kline.timeframe} at {kline.timestamp}")

    @sync_to_async
    def save_order_book(self, order_book):
        # You might want to add data validation here as well
        OrderBook.objects.update_or_create(
            symbol=order_book.symbol,
            timestamp=order_book.timestamp,
            defaults={
                'bids': order_book.bids,
                'asks': order_book.asks,
            }
        )
        logger.debug(f"Saved order book for {order_book.symbol} at {order_book.timestamp}")

    async def start_websockets(self):
        tasks = []
        for url in self.stream_urls:
            tasks.append(asyncio.create_task(self.handle_websocket(url)))
        await asyncio.gather(*tasks)

    async def run(self):
        try:
            logger.info("Initializing exchange...")
            await self.initialize_exchange()
            logger.info("Fetching historical data...")
            await self.fetch_historical_data()
            logger.info("Starting WebSocket connections...")
            await self.start_websockets()
        finally:
            await self.exchange.close()
            logger.info("Exchange connection closed.")

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

    # Filter for USDT perpetual futures pairs
    usdt_pairs = []
    for symbol, market in markets.items():
        if market.get('active') and market.get('contract') and market.get('type') == 'swap' and market.get('linear') and market.get('quote') == 'USDT':
            usdt_pairs.append(symbol)

    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

    fetcher = DataFetch(symbols=usdt_pairs, timeframes=['1m', '5m', '15m', '1h'])
    await fetcher.run()

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