rusty_feeder/exchange/binance/futures/

provider.rs

//! Binance Futures market provider implementation using consolidated WebSocket
//!
//! This module implements the Provider trait for Binance Futures markets,
//! enabling high-performance data acquisition and normalization.

use rusty_common::collections::FxHashMap;
use simd_json::prelude::{ValueAsArray, ValueAsObject, ValueAsScalar, ValueObjectAccess};
use smartstring::alias::String;
use std::fmt::Debug;
use std::sync::Arc;

use anyhow::{Result, anyhow};
use async_trait::async_trait;
use parking_lot::RwLock;
use quanta::Clock;
use rusty_common::json::{self};
use rusty_model::{
    data::{book_snapshot::OrderBookSnapshot, market_trade::MarketTrade},
    instruments::{Instrument, InstrumentId},
    venues::Venue,
};
use smallvec::{SmallVec, smallvec};
use tokio::sync::{mpsc, watch};

// Use the new consolidated WebSocket module
use rusty_common::websocket::exchanges::binance;
use rusty_common::websocket::{
    Message, MessageHandler, WebSocketClient, WebSocketConfig, WebSocketError, WebSocketResult,
};

use crate::exchange::binance::common::messages::SubscriptionResult;
use crate::provider::ext::DataHandler;
use crate::provider::prelude::*;
use rust_decimal::prelude::FromStr;
use rusty_model::data::simd_orderbook::SharedSimdOrderBook;

use super::data::{
    orderbook::OrderbookMessage,
    subscription::{create_orderbook_subscription, create_trade_subscription},
    trade::AggTradeMessage,
};
use super::types::{
    BINANCE_COIN_FUTURES_API_URL, BINANCE_COIN_FUTURES_WS_COMBINED_URL,
    BINANCE_COIN_FUTURES_WS_URL, BINANCE_USD_FUTURES_API_URL, BINANCE_USD_FUTURES_WS_COMBINED_URL,
    BINANCE_USD_FUTURES_WS_URL,
};

/// Futures market type (USD-M or COIN-M)
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BinanceFuturesMarketType {
    /// USD-M Futures (settled in USDT or BUSD)
    UsdM,

    /// COIN-M Futures (settled in cryptocurrency)
    CoinM,
}

/// Binance Futures exchange provider implementation
pub struct BinanceFuturesProvider {
    /// Connection configuration
    config: ConnectionConfig,

    /// Market type (USD-M or COIN-M)
    market_type: BinanceFuturesMarketType,

    /// WebSocket connection status
    connection_status: Arc<RwLock<ConnectionState>>,

    /// Connection statistics
    stats: Arc<RwLock<ConnectionStats>>,

    /// Shared clock for time synchronization
    clock: Clock,

    /// Active subscriptions
    subscriptions: Arc<RwLock<FxHashMap<String, watch::Sender<bool>>>>,

    /// Message sender for internal communication
    message_tx: mpsc::Sender<ProviderMessage>,

    /// Message receiver
    message_rx: Arc<tokio::sync::RwLock<mpsc::Receiver<ProviderMessage>>>,

    /// WebSocket client
    ws_client: Option<WebSocketClient>,

    /// Data handler
    data_handler: Arc<tokio::sync::RwLock<Option<Box<dyn DataHandler + Send + Sync>>>>,

    /// HTTP client for REST API calls
    http_client: reqwest::Client,
}

impl Debug for BinanceFuturesProvider {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BinanceFuturesProvider")
            .field("market_type", &self.market_type)
            .field("connection_status", &self.connection_status)
            .field("stats", &self.stats)
            .field("ws_client", &self.ws_client.is_some())
            .field("data_handler_present", &true) // Can't check async in Debug
            .finish()
    }
}

/// Message handler for Binance Futures WebSocket
struct BinanceFuturesHandler {
    data_handler: Arc<tokio::sync::RwLock<Option<Box<dyn DataHandler + Send + Sync>>>>,
    clock: Clock,
    stats: Arc<RwLock<ConnectionStats>>,
    message_tx: mpsc::Sender<ProviderMessage>,
}

#[async_trait]
impl MessageHandler for BinanceFuturesHandler {
    async fn on_connected(&mut self) -> WebSocketResult<()> {
        log::info!("Binance Futures WebSocket connected");
        let _ = self.message_tx.send(ProviderMessage::Connected).await;
        Ok(())
    }

    async fn on_message(&mut self, message: Message) -> WebSocketResult<()> {
        match message {
            Message::Text(text) => {
                // Parse the message
                // Parse the message safely
                match unsafe { simd_json::from_str::<json::Value>(&mut text.clone()) } {
                    Ok(value) => {
                        if let Err(e) = self.handle_market_data(&value).await {
                            log::error!("Failed to handle market data: {e}");
                        }
                    }
                    Err(e) => {
                        log::error!("Failed to parse WebSocket message: {e}");
                    }
                }
            }
            Message::Binary(_) => {
                log::warn!("Received unexpected binary message from Binance");
            }
            _ => {}
        }
        Ok(())
    }

    async fn on_error(&mut self, error: WebSocketError) -> WebSocketResult<()> {
        log::error!("Binance Futures WebSocket error: {}", error);
        let _ = self
            .message_tx
            .send(ProviderMessage::Error(error.to_string().into()))
            .await;
        Ok(())
    }

    async fn on_disconnected(&mut self) -> WebSocketResult<()> {
        log::info!("Binance Futures WebSocket disconnected");
        let _ = self.message_tx.send(ProviderMessage::Disconnected).await;
        Ok(())
    }
}

impl BinanceFuturesHandler {
    async fn handle_market_data(&self, value: &json::Value) -> Result<()> {
        // Try to parse as different message types
        // First check if it's an orderbook message
        if let Ok(msg) = json::from_value::<OrderbookMessage>(value.clone()) {
            return self.handle_orderbook_update(msg).await;
        }

        // Check if it's an aggregated trade message
        if let Ok(msg) = json::from_value::<AggTradeMessage>(value.clone()) {
            return self.handle_agg_trade(msg).await;
        }

        // Check if it's a subscription confirmation
        if let Ok(msg) = json::from_value::<SubscriptionResult>(value.clone()) {
            log::debug!("Received subscription confirmation: {msg:?}");
            return Ok(());
        }

        log::debug!("Unhandled message type");
        Ok(())
    }

    async fn handle_orderbook_update(&self, msg: OrderbookMessage) -> Result<()> {
        // Update stats
        self.stats.write().messages_received += 1;

        // Forward to data handler if available
        if let Some(handler) = &*self.data_handler.read().await {
            // Convert to OrderBookSnapshot
            let snapshot = self.convert_to_snapshot(msg)?;
            handler.handle_orderbook_snapshot(snapshot).await?;
        }

        Ok(())
    }

    async fn handle_trade(&self, msg: MarketTrade) -> Result<()> {
        // Update stats
        self.stats.write().messages_received += 1;

        // Forward to data handler
        if let Some(handler) = &*self.data_handler.read().await {
            handler.handle_market_trade(msg).await?;
        }

        Ok(())
    }

    async fn handle_agg_trade(&self, msg: AggTradeMessage) -> Result<()> {
        // Update stats
        self.stats.write().messages_received += 1;

        // Convert to MarketTrade and forward
        if let Some(handler) = &*self.data_handler.read().await {
            let trade = self.convert_agg_trade(msg)?;
            handler.handle_market_trade(trade).await?;
        }

        Ok(())
    }

    fn convert_to_snapshot(&self, msg: OrderbookMessage) -> Result<OrderBookSnapshot> {
        use rust_decimal::prelude::FromStr;
        use rusty_model::data::orderbook::PriceLevel;

        // Get system timestamp
        let timestamp_init = self.clock.raw();

        // Create instrument ID from symbol
        let instrument_id = InstrumentId::new(&msg.symbol, Venue::Binance);

        // Convert bids
        let mut bids = SmallVec::<[PriceLevel; 64]>::new();
        for bid in msg.bids {
            if bid.len() >= 2 {
                let price = rust_decimal::Decimal::from_str(&bid[0])
                    .map_err(|e| anyhow!("Failed to parse bid price: {}", e))?;
                let quantity = rust_decimal::Decimal::from_str(&bid[1])
                    .map_err(|e| anyhow!("Failed to parse bid quantity: {}", e))?;

                bids.push(PriceLevel { price, quantity });
            }
        }

        // Convert asks
        let mut asks = SmallVec::<[PriceLevel; 64]>::new();
        for ask in msg.asks {
            if ask.len() >= 2 {
                let price = rust_decimal::Decimal::from_str(&ask[0])
                    .map_err(|e| anyhow!("Failed to parse ask price: {}", e))?;
                let quantity = rust_decimal::Decimal::from_str(&ask[1])
                    .map_err(|e| anyhow!("Failed to parse ask quantity: {}", e))?;

                asks.push(PriceLevel { price, quantity });
            }
        }

        // Create snapshot
        let snapshot = OrderBookSnapshot::new(
            instrument_id,
            bids,
            asks,
            msg.final_update_id,
            msg.event_time * 1_000_000, // Convert ms to ns
            timestamp_init,
        );

        Ok(snapshot)
    }

    fn convert_agg_trade(&self, msg: AggTradeMessage) -> Result<MarketTrade> {
        use rust_decimal::prelude::FromStr;
        use rusty_model::enums::OrderSide;

        // Parse price and quantity
        let price = rust_decimal::Decimal::from_str(&msg.price)
            .map_err(|e| anyhow!("Failed to parse trade price: {}", e))?;
        let quantity = rust_decimal::Decimal::from_str(&msg.quantity)
            .map_err(|e| anyhow!("Failed to parse trade quantity: {}", e))?;

        // Determine trade side
        // In Binance, if buyer is market maker, the actual aggressor is seller (taker)
        let direction = if msg.is_buyer_market_maker {
            OrderSide::Sell // Seller is aggressor
        } else {
            OrderSide::Buy // Buyer is aggressor
        };

        // Create instrument ID
        let instrument_id = InstrumentId::new(&msg.symbol, Venue::Binance);

        // Create market trade
        let trade = MarketTrade {
            timestamp: self.clock.now(),
            exchange_time_ns: msg.trade_time * 1_000_000, // Convert ms to ns
            price,
            quantity,
            direction,
            instrument_id,
        };

        Ok(trade)
    }
}

/// Internal provider messages
enum ProviderMessage {
    Connected,
    Disconnected,
    Error(String),
    Subscribe(String),
    Unsubscribe(String),
}

impl HttpClientProvider for BinanceFuturesProvider {
    fn http_client(&self) -> &reqwest::Client {
        &self.http_client
    }
}

impl BinanceFuturesProvider {
    /// Create a new Binance Futures provider
    #[must_use]
    pub fn new(config: ConnectionConfig, market_type: BinanceFuturesMarketType) -> Self {
        let (message_tx, message_rx) = mpsc::channel(1000);

        // Create HTTP client before moving config
        let http_client = Self::create_http_client(&config);

        Self {
            config,
            market_type,
            connection_status: Arc::new(RwLock::new(ConnectionState::Disconnected)),
            stats: Arc::new(RwLock::new(ConnectionStats::default())),
            clock: Clock::new(),
            subscriptions: Arc::new(RwLock::new(FxHashMap::default())),
            message_tx,
            message_rx: Arc::new(tokio::sync::RwLock::new(message_rx)),
            ws_client: None,
            data_handler: Arc::new(tokio::sync::RwLock::new(None)),
            http_client,
        }
    }

    /// Get the WebSocket URL based on market type
    const fn get_ws_url(&self, use_combined: bool) -> &'static str {
        match (self.market_type, use_combined) {
            (BinanceFuturesMarketType::UsdM, true) => BINANCE_USD_FUTURES_WS_COMBINED_URL,
            (BinanceFuturesMarketType::UsdM, false) => BINANCE_USD_FUTURES_WS_URL,
            (BinanceFuturesMarketType::CoinM, true) => BINANCE_COIN_FUTURES_WS_COMBINED_URL,
            (BinanceFuturesMarketType::CoinM, false) => BINANCE_COIN_FUTURES_WS_URL,
        }
    }

    /// Create WebSocket configuration
    fn create_ws_config(&self) -> WebSocketConfig {
        binance::default_config(self.get_ws_url(true).to_string())
    }
}

impl BinanceFuturesProvider {
    /// Get provider name
    pub const fn name(&self) -> &str {
        match self.market_type {
            BinanceFuturesMarketType::UsdM => "binance_futures_usdm",
            BinanceFuturesMarketType::CoinM => "binance_futures_coinm",
        }
    }

    /// Get venue
    pub const fn venue(&self) -> Venue {
        Venue::Binance
    }

    /// Connect to WebSocket
    pub async fn connect(&mut self) -> Result<()> {
        // Update connection state
        *self.connection_status.write() = ConnectionState::Connecting;

        // Create WebSocket client
        let config = self.create_ws_config();
        let client = WebSocketClient::new(config);

        // Create handler
        let handler = BinanceFuturesHandler {
            data_handler: self.data_handler.clone(),
            clock: self.clock.clone(),
            stats: self.stats.clone(),
            message_tx: self.message_tx.clone(),
        };

        // Store client reference
        self.ws_client = Some(client);

        // Run the client (this will handle the WebSocket connection)
        // Note: In a real implementation, you'd want to spawn this in a separate task
        // or handle it differently based on your architecture

        // Update connection state
        *self.connection_status.write() = ConnectionState::Connected;

        Ok(())
    }

    /// Disconnect from WebSocket
    pub async fn disconnect(&mut self) -> Result<()> {
        if let Some(client) = &self.ws_client {
            client.shutdown();
        }

        *self.connection_status.write() = ConnectionState::Disconnected;
        self.ws_client = None;

        Ok(())
    }

    /// Subscribe to orderbook updates
    pub async fn subscribe_orderbook(
        &mut self,
        instrument_id: &InstrumentId,
        depth: usize,
    ) -> Result<()> {
        let symbol = instrument_id.symbol.to_lowercase();
        let stream = format!("{symbol}@depth{depth}@100ms");

        // Create subscription message
        let mut symbols: SmallVec<[String; 8]> = SmallVec::<[String; 8]>::new();
        symbols.push(symbol.into());
        let sub_msg =
            create_orderbook_subscription("orderbook", symbols, Some(depth), Some(1), None, false);

        // Send subscription
        if let Some(client) = &self.ws_client {
            let message = simd_json::to_string(&sub_msg)
                .map_err(|e| anyhow!("Failed to serialize subscription: {}", e))?;

            client
                .send(Message::Text(message.into()))
                .await
                .map_err(|e| anyhow!("Failed to send subscription: {:?}", e))?;

            log::info!("Subscribed to orderbook: {stream}");
        } else {
            return Err(anyhow!("WebSocket client not connected"));
        }

        Ok(())
    }

    /// Subscribe to trade updates
    pub async fn subscribe_trades(&mut self, instrument_id: &InstrumentId) -> Result<()> {
        let symbol = instrument_id.symbol.to_lowercase();
        let stream = format!("{symbol}@aggTrade");

        // Create subscription message
        let mut symbols: SmallVec<[String; 8]> = SmallVec::<[String; 8]>::new();
        symbols.push(symbol.into());
        let sub_msg = create_trade_subscription("trades", symbols, Some(1), None, false);

        // Send subscription
        if let Some(client) = &self.ws_client {
            let message = simd_json::to_string(&sub_msg)
                .map_err(|e| anyhow!("Failed to serialize subscription: {}", e))?;

            client
                .send(Message::Text(message.into()))
                .await
                .map_err(|e| anyhow!("Failed to send subscription: {:?}", e))?;

            log::info!("Subscribed to trades: {stream}");
        } else {
            return Err(anyhow!("WebSocket client not connected"));
        }

        Ok(())
    }

    /// Unsubscribe from orderbook updates
    pub async fn unsubscribe_orderbook(&mut self, instrument_id: &InstrumentId) -> Result<()> {
        let symbol = instrument_id.symbol.to_lowercase();

        // Remove from subscriptions
        self.subscriptions.write().remove(symbol.as_str());

        // Send unsubscribe message
        if let Some(client) = &self.ws_client {
            use super::data::subscription::WebSocketSubscribeRequest;

            // Get the depth from stored subscription info (default to 10)
            let depth = 10; // You might want to store this in subscriptions map

            let unsub_msg = WebSocketSubscribeRequest {
                id: 2,
                method: "UNSUBSCRIBE".into(),
                params: smallvec![format!("{symbol}@depth{depth}").into()],
            };

            let message = simd_json::to_string(&unsub_msg)
                .map_err(|e| anyhow!("Failed to serialize unsubscription: {}", e))?;

            client
                .send(Message::Text(message.into()))
                .await
                .map_err(|e| anyhow!("Failed to send unsubscription: {:?}", e))?;

            log::info!("Unsubscribed from orderbook: {symbol}");
        }

        Ok(())
    }

    /// Unsubscribe from trade updates
    pub async fn unsubscribe_trades(&mut self, instrument_id: &InstrumentId) -> Result<()> {
        let symbol = instrument_id.symbol.to_lowercase();

        // Remove from subscriptions
        self.subscriptions.write().remove(symbol.as_str());

        // Send unsubscribe message
        if let Some(client) = &self.ws_client {
            use super::data::subscription::WebSocketSubscribeRequest;

            let unsub_msg = WebSocketSubscribeRequest {
                id: 2,
                method: "UNSUBSCRIBE".into(),
                params: smallvec![format!("{symbol}@aggTrade").into()],
            };

            let message = simd_json::to_string(&unsub_msg)
                .map_err(|e| anyhow!("Failed to serialize unsubscription: {}", e))?;

            client
                .send(Message::Text(message.into()))
                .await
                .map_err(|e| anyhow!("Failed to send unsubscription: {:?}", e))?;

            log::info!("Unsubscribed from trades: {symbol}");
        }

        Ok(())
    }

    /// Check if connected
    pub fn is_connected(&self) -> bool {
        matches!(*self.connection_status.read(), ConnectionState::Connected)
    }

    /// Get connection statistics
    pub fn connection_stats(&self) -> ConnectionStats {
        self.stats.read().clone()
    }

    /// Run with data handler
    pub async fn run_with_handler(
        &mut self,
        handler: Box<dyn DataHandler + Send + Sync>,
    ) -> Result<()> {
        // Store the handler
        *self.data_handler.write().await = Some(handler);

        // Connect if not already connected
        if !self.is_connected() {
            self.connect().await?;
        }

        // Process internal messages
        let mut rx = self.message_rx.write().await;
        while let Some(msg) = rx.recv().await {
            match msg {
                ProviderMessage::Connected => {
                    log::info!("Provider connected");
                }
                ProviderMessage::Disconnected => {
                    log::warn!("Provider disconnected");
                    // Attempt reconnection is handled by WebSocketClient
                }
                ProviderMessage::Error(e) => {
                    log::error!("Provider error: {e}");
                }
                _ => {}
            }
        }

        Ok(())
    }

    /// Fetch available instruments
    pub async fn fetch_instruments(&self) -> Result<Vec<Box<dyn Instrument>>> {
        use reqwest::Client;
        use rusty_model::instruments::SpotInstrument;

        // Get API URL based on market type
        let api_url = match self.market_type {
            BinanceFuturesMarketType::UsdM => BINANCE_USD_FUTURES_API_URL,
            BinanceFuturesMarketType::CoinM => BINANCE_COIN_FUTURES_API_URL,
        };

        // Create HTTP client
        let client = Client::new();

        // Fetch exchange info
        let url = format!("{api_url}/fapi/v1/exchangeInfo");
        let response = client
            .get(&url)
            .send()
            .await
            .map_err(|e| anyhow!("Failed to fetch exchange info: {}", e))?;

        // Check response status
        if !response.status().is_success() {
            let status = response.status();
            let body = response.text().await.unwrap_or_default();
            return Err(anyhow!(
                "Failed to fetch instruments: {} - {}",
                status,
                body
            ));
        }

        // Parse JSON response
        let body = response
            .text()
            .await
            .map_err(|e| anyhow!("Failed to read response body: {}", e))?;
        let data: json::Value = unsafe { simd_json::from_str(&mut body.clone()) }
            .map_err(|e| anyhow!("Failed to parse exchange info: {}", e))?;

        // Extract instruments
        let mut instruments = Vec::new();

        if let Some(obj) = data.as_object()
            && let Some(symbols) = obj.get("symbols").and_then(|v| v.as_array())
        {
            for symbol_data in symbols {
                // Check if it's an object first
                if let Some(symbol_obj) = symbol_data.as_object() {
                    // Extract symbol info
                    let symbol = symbol_obj
                        .get("symbol")
                        .and_then(|v| v.as_str())
                        .ok_or_else(|| anyhow!("Missing symbol field"))?;

                    let status = symbol_obj
                        .get("status")
                        .and_then(|v| v.as_str())
                        .unwrap_or("");

                    // Only include active trading symbols
                    if status != "TRADING" {
                        continue;
                    }

                    let base_asset = symbol_obj
                        .get("baseAsset")
                        .and_then(|v| v.as_str())
                        .ok_or_else(|| anyhow!("Missing baseAsset field"))?;

                    let quote_asset = symbol_obj
                        .get("quoteAsset")
                        .and_then(|v| v.as_str())
                        .ok_or_else(|| anyhow!("Missing quoteAsset field"))?;

                    // Create instrument (using SpotInstrument for now)
                    // TODO: Create proper FuturesInstrument type with contract details
                    let instrument =
                        SpotInstrument::new(symbol, base_asset, quote_asset, Venue::Binance);

                    instruments.push(Box::new(instrument) as Box<dyn Instrument>);
                }
            }
        }

        log::info!(
            "Fetched {} instruments from Binance Futures",
            instruments.len()
        );
        Ok(instruments)
    }
}

/// Provider trait implementation for Binance Futures
#[async_trait]
impl Provider for BinanceFuturesProvider {
    type TradeMessage = AggTradeMessage;
    type DepthMessage = OrderbookMessage;
    type InstrumentMessage = json::Value;

    fn name(&self) -> &'static str {
        match self.market_type {
            BinanceFuturesMarketType::UsdM => "BinanceFuturesUSDM",
            BinanceFuturesMarketType::CoinM => "BinanceFuturesCOINM",
        }
    }

    fn venue(&self) -> Venue {
        Venue::Binance
    }

    fn config(&self) -> &ConnectionConfig {
        &self.config
    }

    async fn init(&mut self) -> Result<()> {
        // Initialize WebSocket client using consolidated implementation
        let ws_url = match self.market_type {
            BinanceFuturesMarketType::UsdM => BINANCE_USD_FUTURES_WS_COMBINED_URL.to_string(),
            BinanceFuturesMarketType::CoinM => BINANCE_COIN_FUTURES_WS_COMBINED_URL.to_string(),
        };

        let ws_config = binance::default_config(ws_url);
        self.ws_client = Some(WebSocketClient::new(ws_config));

        // Update connection status
        *self.connection_status.write() = ConnectionState::Connected;
        Ok(())
    }

    async fn subscribe_trades(
        &self,
        symbols: SmallVec<[String; 8]>,
    ) -> Result<mpsc::Receiver<Self::TradeMessage>> {
        let (tx, rx) = mpsc::channel(1024);

        // Create subscription message for multiple symbols
        let sub_msg = create_trade_subscription("trades", symbols.clone(), Some(1), None, false);

        if let Some(client) = &self.ws_client {
            let message = simd_json::to_string(&sub_msg)
                .map_err(|e| anyhow!("Failed to serialize subscription: {}", e))?;

            client
                .send(Message::Text(message.into()))
                .await
                .map_err(|e| anyhow!("Failed to send subscription: {:?}", e))?;

            log::info!("Subscribed to trades for symbols: {symbols:?}");
        } else {
            return Err(anyhow!("WebSocket client not initialized"));
        }

        Ok(rx)
    }

    async fn unsubscribe_trades(&self) -> Result<()> {
        // TODO: Implement unsubscribe logic
        Ok(())
    }

    async fn subscribe_orderbook(
        &self,
        symbols: SmallVec<[String; 8]>,
    ) -> Result<mpsc::Receiver<Self::DepthMessage>> {
        let (tx, rx) = mpsc::channel(1024);

        // Create subscription message for multiple symbols with depth
        let sub_msg = create_orderbook_subscription(
            "orderbook",
            symbols.clone(),
            Some(20), // Default depth
            Some(1),
            None,
            false,
        );

        if let Some(client) = &self.ws_client {
            let message = simd_json::to_string(&sub_msg)
                .map_err(|e| anyhow!("Failed to serialize subscription: {}", e))?;

            client
                .send(Message::Text(message.into()))
                .await
                .map_err(|e| anyhow!("Failed to send subscription: {:?}", e))?;

            log::info!("Subscribed to orderbook for symbols: {symbols:?}");
        } else {
            return Err(anyhow!("WebSocket client not initialized"));
        }

        Ok(rx)
    }

    async fn unsubscribe_orderbook(&self) -> Result<()> {
        // TODO: Implement unsubscribe logic
        Ok(())
    }

    async fn get_instruments(&self) -> Result<Vec<Box<dyn Instrument>>> {
        // Use the existing fetch_instruments method
        self.fetch_instruments().await
    }

    async fn get_historical_trades(
        &self,
        _symbol: &str,
        _limit: Option<u32>,
    ) -> Result<Vec<MarketTrade>> {
        // TODO: Implement historical trades fetching
        Err(anyhow!("Historical trades not implemented yet"))
    }

    async fn get_orderbook_snapshot(
        &self,
        symbol: &str,
        depth: Option<u32>,
    ) -> Result<OrderBookSnapshot> {
        // Fetch orderbook snapshot from REST API
        let limit = depth.unwrap_or(20).min(5000);
        let url = match self.market_type {
            BinanceFuturesMarketType::UsdM => {
                format!("{BINANCE_USD_FUTURES_API_URL}/fapi/v1/depth?symbol={symbol}&limit={limit}")
            }
            BinanceFuturesMarketType::CoinM => {
                format!(
                    "{BINANCE_COIN_FUTURES_API_URL}/dapi/v1/depth?symbol={symbol}&limit={limit}"
                )
            }
        };

        let response = self.http_client.get(&url).send().await?;

        // Check for rate limit errors
        use crate::exchange::binance::common::rate_limit::{
            handle_rate_limit_error, parse_order_count_headers,
        };
        if response.status() == 429 || response.status() == 418 {
            let headers = response.headers().clone();
            handle_rate_limit_error(response.status().as_u16(), &headers)?;
        }

        // Parse order count headers for monitoring
        let order_counts = parse_order_count_headers(response.headers());
        if !order_counts.is_empty() {
            log::debug!("Binance order count limits: {order_counts:?}");
        }

        let body = response.text().await?;
        let data: json::Value = unsafe { simd_json::from_str(&mut body.clone()) }?;

        // Parse the response
        let instrument_id = InstrumentId::new(symbol, Venue::Binance);
        let timestamp = self.clock.raw();
        let last_update_id = data
            .get("lastUpdateId")
            .and_then(|v| v.as_u64())
            .unwrap_or(0);

        let mut snapshot = OrderBookSnapshot::new_empty(instrument_id, timestamp, last_update_id);

        // Parse bids
        if let Some(bids) = data.get("bids").and_then(|v| v.as_array()) {
            for bid in bids {
                if let Some(arr) = bid.as_array()
                    && arr.len() >= 2
                    && let (Some(price_str), Some(qty_str)) = (arr[0].as_str(), arr[1].as_str())
                    && let (Ok(price), Ok(quantity)) = (
                        rust_decimal::Decimal::from_str(price_str),
                        rust_decimal::Decimal::from_str(qty_str),
                    )
                {
                    snapshot.add_bid(price, quantity);
                }
            }
        }

        // Parse asks
        if let Some(asks) = data.get("asks").and_then(|v| v.as_array()) {
            for ask in asks {
                if let Some(arr) = ask.as_array()
                    && arr.len() >= 2
                    && let (Some(price_str), Some(qty_str)) = (arr[0].as_str(), arr[1].as_str())
                    && let (Ok(price), Ok(quantity)) = (
                        rust_decimal::Decimal::from_str(price_str),
                        rust_decimal::Decimal::from_str(qty_str),
                    )
                {
                    snapshot.add_ask(price, quantity);
                }
            }
        }

        Ok(snapshot)
    }

    async fn get_realtime_orderbook(&self, _symbol: &str) -> Result<SharedSimdOrderBook> {
        // TODO: Implement realtime orderbook access
        Err(anyhow!("Realtime orderbook not implemented yet"))
    }

    async fn is_connected(&self) -> bool {
        *self.connection_status.read() == ConnectionState::Connected
    }

    async fn connection_status(&self) -> ConnectionState {
        *self.connection_status.read()
    }

    async fn get_stats(&self) -> ConnectionStats {
        self.stats.read().clone()
    }

    async fn ping(&self) -> Result<u64> {
        if let Some(client) = &self.ws_client {
            let start = std::time::Instant::now();

            // Send ping frame
            client.send(Message::Ping(vec![])).await?;

            // Calculate round-trip time
            let elapsed = start.elapsed();
            Ok(elapsed.as_millis() as u64)
        } else {
            Err(anyhow!("WebSocket client not initialized"))
        }
    }

    async fn reset_connection(&self) -> Result<()> {
        // Reset connection status
        *self.connection_status.write() = ConnectionState::Disconnected;

        // Reset stats
        *self.stats.write() = ConnectionStats::default();

        // TODO: Implement proper reconnection logic
        Ok(())
    }

    fn get_rate_limits(&self) -> Vec<RateLimit> {
        vec![
            RateLimit {
                limit_type: "REQUEST_WEIGHT",
                interval: "MINUTE",
                interval_num: 1,
                limit: 1200,
            },
            RateLimit {
                limit_type: "ORDERS",
                interval: "SECOND",
                interval_num: 10,
                limit: 300,
            },
        ]
    }

    async fn shutdown(&mut self) -> Result<()> {
        // Shutdown WebSocket client
        if let Some(client) = &self.ws_client {
            client.shutdown();
        }

        // Update connection status
        *self.connection_status.write() = ConnectionState::Disconnected;

        // Clear subscriptions
        self.subscriptions.write().clear();

        Ok(())
    }
}

/// MonitorExt implementation for BinanceFuturesProvider
///
/// Provides advanced monitoring capabilities for HFT trading systems
#[async_trait]
impl crate::provider::ext::MonitorExt for BinanceFuturesProvider {
    async fn get_instrument_metrics(
        &self,
        instrument_id: &InstrumentId,
    ) -> Result<crate::provider::ext::InstrumentMetrics> {
        use crate::feeder::FeedStats;
        use crate::provider::ext::{InstrumentMetrics, LatencyMetrics, OrderFlowMetrics};

        let connection_stats = self.connection_stats();
        let current_time = self.clock.raw();

        // Calculate order flow metrics based on recent activity
        let order_flow_metrics = OrderFlowMetrics {
            volume_imbalance: 0.0, // Would be calculated from recent trades
            book_pressure: 0.0,    // Would be calculated from order book data
            trade_intensity: connection_stats.messages_received as f64 / 60.0, // Messages per minute
            avg_trade_size: rust_decimal::Decimal::ZERO, // Would track from trades
            price_volatility: 0.0,                       // Would calculate from price movements
            avg_spread_bps: 0.0,                         // Would track from order book updates
            max_spread_bps: 0.0,                         // Would track from order book updates
            market_impact_bps: 0.0,                      // Would calculate from trade analysis
        };

        // Calculate latency metrics
        let latency_metrics = LatencyMetrics {
            network_latency_ns: connection_stats.avg_latency_ns,
            parsing_latency_ns: 50_000,    // Estimated parsing time
            processing_latency_ns: 25_000, // Estimated processing time
            total_latency_ns: connection_stats.avg_latency_ns + 75_000,
            latency_jitter_ns: connection_stats.avg_latency_ns / 10, // Estimated jitter
        };

        // Calculate health score based on performance metrics
        let health_score = if connection_stats.errors == 0 {
            1.0
        } else {
            (1.0 - (connection_stats.errors as f64
                / connection_stats.messages_received.max(1) as f64))
                .max(0.0)
        };

        // Basic feed stats (would be enhanced with real-time data)
        let mut feed_stats = FeedStats::default();
        feed_stats.messages_processed = connection_stats.messages_received;
        feed_stats.messages_per_second = 0.0; // Would calculate from timing
        feed_stats.avg_process_latency_ns = connection_stats.avg_latency_ns;
        feed_stats.max_process_latency_ns = connection_stats.avg_latency_ns; // Simplified
        feed_stats.p99_process_latency_ns = connection_stats.avg_latency_ns;
        feed_stats.dropped_messages = 0; // Would track from actual feed drops
        feed_stats.last_update_time = current_time;
        feed_stats.memory_usage_bytes = std::mem::size_of::<Self>();

        Ok(InstrumentMetrics {
            instrument_id: instrument_id.clone(),
            feed_stats,
            connection_stats,
            order_flow_metrics,
            latency_metrics,
            anomaly_score: 0.0, // Would be calculated by anomaly detection
            health_score,
            last_updated: current_time,
        })
    }

    async fn get_aggregated_metrics(&self) -> Result<crate::provider::ext::AggregatedMetrics> {
        use crate::provider::ext::AggregatedMetrics;

        let connection_stats = self.connection_stats();
        let subscriptions = self.subscriptions.read();
        let current_time = self.clock.raw();

        Ok(AggregatedMetrics {
            total_instruments: subscriptions.len(),
            total_messages_processed: connection_stats.messages_received,
            total_dropped_messages: 0, // Would track from feed stats
            avg_latency_ns: connection_stats.avg_latency_ns,
            max_latency_ns: connection_stats.avg_latency_ns,
            p99_latency_ns: connection_stats.avg_latency_ns, // Simplified
            total_memory_usage_bytes: std::mem::size_of::<Self>(),
            total_errors: connection_stats.errors,
            overall_health_score: if connection_stats.errors == 0 {
                1.0
            } else {
                0.8
            },
            active_alerts: 0, // Would track active alerts
            timestamp: current_time,
        })
    }

    async fn register_metrics_collector(
        &mut self,
        _collector: Box<dyn crate::provider::ext::MetricsCollector>,
    ) -> Result<()> {
        // Would store collector and start collection
        log::info!("Metrics collector registered for BinanceFuturesProvider");
        Ok(())
    }

    async fn enable_alert(
        &mut self,
        alert_config: crate::provider::ext::AlertConfig,
    ) -> Result<crate::provider::ext::AlertHandle> {
        // Would implement alerting system
        log::info!(
            "Alert enabled: {} with threshold {}",
            alert_config.name,
            alert_config.threshold
        );
        Ok(crate::provider::ext::AlertHandle(1)) // Placeholder handle
    }

    async fn disable_alert(&mut self, _handle: crate::provider::ext::AlertHandle) -> Result<()> {
        // Would disable specific alert
        log::info!("Alert disabled");
        Ok(())
    }

    async fn get_health_status(&self) -> Result<crate::provider::ext::HealthStatus> {
        use crate::provider::ext::{ComponentHealth, HealthState, HealthStatus};

        let connection_stats = self.connection_stats();
        let connected = self.is_connected();
        let current_time = self.clock.raw();

        let overall_status = if connected {
            if connection_stats.errors == 0 {
                HealthState::Healthy
            } else {
                HealthState::Degraded
            }
        } else {
            HealthState::Offline
        };

        let component_health = vec![
            ComponentHealth {
                component_name: "WebSocket Connection".to_string(),
                status: if connected {
                    HealthState::Healthy
                } else {
                    HealthState::Offline
                },
                last_heartbeat: current_time,
                error_rate: connection_stats.errors as f64
                    / connection_stats.messages_received.max(1) as f64,
                latency_p99_ns: connection_stats.avg_latency_ns,
            },
            ComponentHealth {
                component_name: "Message Processing".to_string(),
                status: HealthState::Healthy, // Would check actual message processing stats
                last_heartbeat: current_time,
                error_rate: 0.0, // Would calculate from actual processing stats
                latency_p99_ns: connection_stats.avg_latency_ns,
            },
        ];

        let performance_grade = match overall_status {
            HealthState::Healthy => 'A',
            HealthState::Degraded => 'B',
            HealthState::Critical => 'D',
            HealthState::Offline => 'F',
        };

        Ok(HealthStatus {
            overall_status,
            component_health,
            active_issues: vec![], // Would track actual issues
            performance_grade,
            uptime_seconds: (current_time - connection_stats.connected_time) / 1_000_000_000, // Convert ns to seconds
            last_check_time: current_time,
        })
    }

    async fn start_anomaly_detection(
        &mut self,
        _config: crate::provider::ext::AnomalyConfig,
    ) -> Result<()> {
        // Would implement anomaly detection algorithms
        log::info!("Anomaly detection started for BinanceFuturesProvider");
        Ok(())
    }

    async fn stop_anomaly_detection(&mut self) -> Result<()> {
        // Would stop anomaly detection
        log::info!("Anomaly detection stopped for BinanceFuturesProvider");
        Ok(())
    }

    async fn get_latency_distribution(
        &self,
        _instrument_id: &InstrumentId,
    ) -> Result<crate::provider::ext::LatencyDistribution> {
        use crate::provider::ext::{LatencyBucket, LatencyDistribution};

        let connection_stats = self.connection_stats();

        // Simplified latency distribution based on available stats
        Ok(LatencyDistribution {
            p50_ns: connection_stats.avg_latency_ns,
            p90_ns: connection_stats.avg_latency_ns + (connection_stats.avg_latency_ns / 4),
            p95_ns: connection_stats.avg_latency_ns + (connection_stats.avg_latency_ns / 2),
            p99_ns: connection_stats.avg_latency_ns,
            p99_9_ns: connection_stats.avg_latency_ns + (connection_stats.avg_latency_ns / 10),
            max_ns: connection_stats.avg_latency_ns,
            histogram: vec![
                LatencyBucket {
                    min_ns: 0,
                    max_ns: 100_000,
                    count: 10,
                },
                LatencyBucket {
                    min_ns: 100_000,
                    max_ns: 500_000,
                    count: 50,
                },
                LatencyBucket {
                    min_ns: 500_000,
                    max_ns: 1_000_000,
                    count: 30,
                },
                LatencyBucket {
                    min_ns: 1_000_000,
                    max_ns: 5_000_000,
                    count: 10,
                },
            ],
        })
    }

    async fn export_prometheus_metrics(&self) -> Result<std::string::String> {
        let connection_stats = self.connection_stats();
        let subscriptions = self.subscriptions.read();
        let provider_name = self.name();

        let prometheus_metrics = format!(
            "# HELP binance_futures_messages_received_total Total messages received\n\
             # TYPE binance_futures_messages_received_total counter\n\
             binance_futures_messages_received_total{{provider=\"{provider_name}\"}} {}\n\
             # HELP binance_futures_messages_dropped_total Total messages dropped\n\
             # TYPE binance_futures_messages_dropped_total counter\n\
             binance_futures_messages_dropped_total{{provider=\"{provider_name}\"}} {}\n\
             # HELP binance_futures_latency_avg_ns Average latency in nanoseconds\n\
             # TYPE binance_futures_latency_avg_ns gauge\n\
             binance_futures_latency_avg_ns{{provider=\"{provider_name}\"}} {}\n\
             # HELP binance_futures_active_subscriptions Number of active subscriptions\n\
             # TYPE binance_futures_active_subscriptions gauge\n\
             binance_futures_active_subscriptions{{provider=\"{provider_name}\"}} {}\n",
            connection_stats.messages_received,
            0, // Would track actual dropped messages
            connection_stats.avg_latency_ns,
            subscriptions.len()
        );

        Ok(prometheus_metrics)
    }

    async fn reset_statistics(&mut self) -> Result<()> {
        // Reset connection statistics
        let mut stats = self.stats.write();
        *stats = ConnectionStats::default();
        log::info!("Statistics reset for BinanceFuturesProvider");
        Ok(())
    }

    async fn get_trading_metrics(
        &self,
        _instrument_id: &InstrumentId,
    ) -> Result<crate::provider::ext::TradingMetrics> {
        use crate::provider::ext::{
            ExecutionQualityMetrics, MicrostructureSignals, TradingMetrics,
        };

        // Would calculate from actual trading data
        Ok(TradingMetrics {
            order_flow_imbalance: 0.0,
            vwap: rust_decimal::Decimal::ZERO,
            twap: rust_decimal::Decimal::ZERO,
            microstructure_signals: MicrostructureSignals {
                book_imbalance: 0.0,
                trade_sign_accuracy: 0.5,
                alpha_signal: 0.0,
                momentum: 0.0,
                mean_reversion: 0.0,
            },
            execution_quality: ExecutionQualityMetrics {
                slippage_bps: 0.0,
                market_impact_bps: 0.0,
                fill_rate: 1.0,
                avg_fill_time_ns: 1_000_000,
                implementation_shortfall_bps: 0.0,
            },
        })
    }
}