import numpy as np
from scipy import stats
import pandas as pd
from collections import deque
from sklearn.cluster import KMeans
from sklearn.metrics import silhouette_score
from sklearn.preprocessing import StandardScaler
from typing import Dict, List, Tuple
import logging
from binance.client import Client
from binance.exceptions import BinanceAPIException
import time

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class OrderbookBot:
    def __init__(self, short_period: int = 100, long_period: int = 500, volume_threshold: float = 2, ob_imbalance_threshold: float = 0.7):
        self.short_period = short_period
        self.long_period = long_period
        self.volume_threshold = volume_threshold
        self.ob_imbalance_threshold = ob_imbalance_threshold
        self.pair_data: Dict[str, Dict] = {}
        self.client = Client()

    def initialize_pair(self, pair: str):
        self.pair_data[pair] = {
            'volume_history': deque(maxlen=self.long_period),
            'ob_imbalance_history': deque(maxlen=self.long_period),
            'price_history': deque(maxlen=self.long_period),
            'volatility_history': deque(maxlen=self.long_period)
        }

    def calculate_volume_zscore(self, pair: str, current_volume: float) -> float:
        self.pair_data[pair]['volume_history'].append(current_volume)
        if len(self.pair_data[pair]['volume_history']) < self.short_period:
            return 0
        return stats.zscore(list(self.pair_data[pair]['volume_history']))[-1]

    def calculate_ob_imbalance(self, bids: List[Tuple[float, float]], asks: List[Tuple[float, float]]) -> float:
        total_bid_volume = sum(bid[1] for bid in bids)
        total_ask_volume = sum(ask[1] for ask in asks)
        return total_bid_volume / (total_bid_volume + total_ask_volume)

    def calculate_ob_imbalance_zscore(self, pair: str, current_imbalance: float) -> float:
        self.pair_data[pair]['ob_imbalance_history'].append(current_imbalance)
        if len(self.pair_data[pair]['ob_imbalance_history']) < self.short_period:
            return 0
        return stats.zscore(list(self.pair_data[pair]['ob_imbalance_history']))[-1]

    def calculate_volume_weighted_price(self, trades: List[Dict[str, float]]) -> float:
        return sum(trade['price'] * trade['volume'] for trade in trades) / sum(trade['volume'] for trade in trades)

    def calculate_ob_depth(self, order_book: Dict[str, List[Tuple[float, float]]], levels: int = 10) -> Tuple[float, float]:
        bid_depth = sum(bid[1] for bid in order_book['bids'][:levels])
        ask_depth = sum(ask[1] for ask in order_book['asks'][:levels])
        return bid_depth, ask_depth

    def detect_spoofing(self, order_book: Dict[str, List[Tuple[float, float]]], trades: List[Dict[str, float]], time_window: int = 60) -> bool:
        large_orders = [order for order in order_book['bids'] + order_book['asks'] if order[1] > self.volume_threshold * np.mean([trade['volume'] for trade in trades])]
        return any(order[0] not in [trade['price'] for trade in trades] for order in large_orders)

    def calculate_volatility(self, pair: str, current_price: float) -> float:
        self.pair_data[pair]['price_history'].append(current_price)
        if len(self.pair_data[pair]['price_history']) < self.short_period:
            return 0
        returns = np.diff(list(self.pair_data[pair]['price_history'])) / list(self.pair_data[pair]['price_history'])[:-1]
        volatility = np.std(returns) * np.sqrt(len(returns))
        self.pair_data[pair]['volatility_history'].append(volatility)
        return volatility

    def detect_market_regime(self, pair: str) -> str:
        if len(self.pair_data[pair]['volatility_history']) < self.short_period:
            return 'unknown'
        recent_volatility = list(self.pair_data[pair]['volatility_history'])[-self.short_period:]
        
        # Feature engineering: Include price momentum and volume
        price_momentum = (self.pair_data[pair]['price_history'][-1] / self.pair_data[pair]['price_history'][-self.short_period]) - 1
        volume_change = (self.pair_data[pair]['volume_history'][-1] / np.mean(self.pair_data[pair]['volume_history'][-self.short_period:])) - 1
        
        features = np.array([[v, price_momentum, volume_change] for v in recent_volatility])
        
        # Normalize features
        scaler = StandardScaler()
        normalized_features = scaler.fit_transform(features)
        
        # Use silhouette score to determine optimal number of clusters
        best_n_clusters = 2
        best_silhouette = -1
        for n_clusters in range(2, 6):
            kmeans = KMeans(n_clusters=n_clusters, random_state=0, n_init=10)
            cluster_labels = kmeans.fit_predict(normalized_features)
            silhouette = silhouette_score(normalized_features, cluster_labels)
            if silhouette > best_silhouette:
                best_silhouette = silhouette
                best_n_clusters = n_clusters
        
        kmeans = KMeans(n_clusters=best_n_clusters, random_state=0, n_init=10).fit(normalized_features)
        current_regime = kmeans.predict(normalized_features[-1].reshape(1, -1))[0]
        
        regimes = ['low_volatility', 'medium_volatility', 'high_volatility', 'extreme_volatility', 'unknown']
        return regimes[min(current_regime, len(regimes) - 1)]

    def adaptive_thresholds(self, market_regime: str) -> Tuple[float, float]:
        if market_regime == 'high_volatility':
            return self.volume_threshold * 1.5, self.ob_imbalance_threshold * 1.2
        elif market_regime == 'low_volatility':
            return self.volume_threshold * 0.75, self.ob_imbalance_threshold * 0.9
        elif market_regime == 'extreme_volatility':
            return self.volume_threshold * 2, self.ob_imbalance_threshold * 1.5
        else:
            return self.volume_threshold, self.ob_imbalance_threshold

    def calculate_stop_loss_take_profit(self, signal: str, current_price: float, volatility: float) -> Tuple[float, float]:
        atr_multiplier = 2  # Adjust this value based on risk tolerance
        stop_loss = current_price * (1 - atr_multiplier * volatility) if signal == "Buy" else current_price * (1 + atr_multiplier * volatility)
        take_profit = current_price * (1 + 2 * atr_multiplier * volatility) if signal == "Buy" else current_price * (1 - 2 * atr_multiplier * volatility)
        return stop_loss, take_profit

    def generate_signal(self, pair: str, current_volume: float, bids: List[Tuple[float, float]], asks: List[Tuple[float, float]], 
                        trades: List[Dict[str, float]], order_book: Dict[str, List[Tuple[float, float]]], current_price: float) -> Tuple[str, Dict]:
        try:
            if pair not in self.pair_data:
                self.initialize_pair(pair)

            volume_zscore = self.calculate_volume_zscore(pair, current_volume)
            ob_imbalance = self.calculate_ob_imbalance(bids, asks)
            ob_imbalance_zscore = self.calculate_ob_imbalance_zscore(pair, ob_imbalance)
            vwap = self.calculate_volume_weighted_price(trades)
            bid_depth, ask_depth = self.calculate_ob_depth(order_book)
            spoofing_detected = self.detect_spoofing(order_book, trades)
            volatility = self.calculate_volatility(pair, current_price)
            market_regime = self.detect_market_regime(pair)
            
            adaptive_volume_threshold, adaptive_ob_threshold = self.adaptive_thresholds(market_regime)

            if volume_zscore > adaptive_volume_threshold and ob_imbalance > adaptive_ob_threshold:
                if ob_imbalance_zscore > 0 and bid_depth > ask_depth:
                    signal = "Buy"
                elif ob_imbalance_zscore < 0 and ask_depth > bid_depth:
                    signal = "Sell"
                else:
                    signal = "Neutral"
            else:
                signal = "Neutral"

            if spoofing_detected:
                signal = "Neutral"

            signal_strength = abs(volume_zscore * ob_imbalance_zscore)
            stop_loss, take_profit = self.calculate_stop_loss_take_profit(signal, current_price, volatility)

            metrics = {
                'volume_zscore': volume_zscore,
                'ob_imbalance': ob_imbalance,
                'ob_imbalance_zscore': ob_imbalance_zscore,
                'vwap': vwap,
                'bid_depth': bid_depth,
                'ask_depth': ask_depth,
                'spoofing_detected': spoofing_detected,
                'volatility': volatility,
                'market_regime': market_regime,
                'signal_strength': signal_strength,
                'stop_loss': stop_loss,
                'take_profit': take_profit
            }

            logger.info(f"Generated signal for {pair}: {signal}")
            return signal, metrics
        except Exception as e:
            logger.error(f"Error generating signal for {pair}: {str(e)}")
            return "Neutral", {}

    def get_binance_futures_usdt_pairs(self):
        try:
            exchange_info = self.client.futures_exchange_info()
            return [symbol['symbol'] for symbol in exchange_info['symbols'] if symbol['quoteAsset'] == 'USDT']
        except BinanceAPIException as e:
            logger.error(f"Error fetching USDT pairs from Binance: {str(e)}")
            return []

    def fetch_market_data(self, pair: str):
        try:
            depth = self.client.futures_order_book(symbol=pair)
            trades = self.client.futures_recent_trades(symbol=pair)
            ticker = self.client.futures_ticker(symbol=pair)

            bids = [(float(bid[0]), float(bid[1])) for bid in depth['bids']]
            asks = [(float(ask[0]), float(ask[1])) for ask in depth['asks']]
            recent_trades = [{'price': float(trade['price']), 'volume': float(trade['qty'])} for trade in trades]
            current_price = float(ticker['lastPrice'])
            current_volume = float(ticker['volume'])

            return current_volume, bids, asks, recent_trades, {'bids': bids, 'asks': asks}, current_price
        except BinanceAPIException as e:
            logger.error(f"Error fetching market data for {pair}: {str(e)}")
            return None

    def run(self):
        pairs = self.get_binance_futures_usdt_pairs()
        while True:
            for pair in pairs:
                market_data = self.fetch_market_data(pair)
                if market_data:
                    current_volume, bids, asks, trades, order_book, current_price = market_data
                    signal, metrics = self.generate_signal(pair, current_volume, bids, asks, trades, order_book, current_price)
                    logger.info(f"Pair: {pair}, Signal: {signal}, Metrics: {metrics}")
                time.sleep(1)  # Add delay to avoid API rate limits
            time.sleep(60)  # Wait for 1 minute before next iteration

if __name__ == "__main__":
    bot = OrderbookBot()
    bot.run()

In above code, I want you to modify it so that instead of fetching data on it's own, it uses data stored in the Django ORM database. It is actually fetched by the code below and stored in database. Please use both historical and realtime price data for analysis but only generate signals for the latest datapoints i.e Signals that I can use to enter into trades in real time. Also, use asyncio to concurrently perform analysis for all available pairs for different timeframes i.e 1 minute, 5 minute, 15 minutes, 1 hour all without exceeding the limit. When saving signals please indicate the timestamp, symbol (the trading pair that the signal was generated), Signal type (whether it is a strong buy, buy, weak buy, strong sell, weak sell, neutral), timeframe (if it's 5m, 15m, 1h) price of asset at the time of generating signal, 24hour price change, the stop loss (price level at which I should exit trade to stop further losses), Take profit (price level at which I should take profit and exit the trade) Also include a detailed orderbook report function that will show all relevant performance of the bot. Include all metrics that will assist to monitor performance and finetune the bot.

#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 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:
                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 = 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())