rusty_feeder/exchange/binance/spot/

provider.rs

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

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

use super::message_handler::BinanceSpotMessageHandler;
use super::message_handler_zerocopy::BinanceSpotZeroCopyMessageHandler;
use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use parking_lot::RwLock;
use quanta::Clock;
use reqwest::header::{HeaderMap, HeaderValue};
use rusty_common::json::Value;
use rusty_common::websocket::{WebSocketClient, WebSocketConfig};
use rusty_model::{
    data::{
        book_snapshot::OrderBookSnapshot, market_trade::MarketTrade,
        simd_orderbook::SharedSimdOrderBook,
    },
    instruments::{Instrument, InstrumentId},
    venues::Venue,
};
use smallvec::SmallVec;
use tokio::sync::{mpsc, watch};
use tokio::task::JoinHandle;

use crate::provider::prelude::*;

use super::data::{
    orderbook::{OrderbookMessage, OrderbookSnapshot, ParsedOrderbookSnapshot},
    subscription::{create_orderbook_subscription, create_trade_subscription},
    trade::TradeMessage,
};
use super::types::{
    BINANCE_SPOT_API_URL, BINANCE_SPOT_RATE_LIMITS, BINANCE_SPOT_WS_COMBINED_URL,
    BINANCE_SPOT_WS_URL,
};

use crate::exchange::binance::common::constants::DEFAULT_PING_INTERVAL_MS;
use crate::exchange::binance::instrument::BinanceInstrument;

/// Binance Spot exchange provider implementation
#[derive(Debug)]
pub struct BinanceSpotProvider {
    /// Connection configuration
    config: ConnectionConfig,

    /// 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>>>>,

    /// Cached instruments
    _instruments: Arc<RwLock<FxHashMap<String, Box<dyn Instrument>>>>,

    /// Task handles for WebSocket connections
    ws_handles: Arc<RwLock<FxHashMap<String, JoinHandle<()>>>>,

    /// Last connection attempt timestamp
    last_connection_attempt: Arc<RwLock<Instant>>,

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

impl Default for BinanceSpotProvider {
    fn default() -> Self {
        Self::new()
    }
}

impl BinanceSpotProvider {
    /// Create a new Binance Spot provider with default configuration
    #[inline]
    #[must_use]
    pub fn new() -> Self {
        Self::with_config(None)
    }

    /// Create a new Binance Spot provider with custom configuration
    #[inline]
    #[must_use]
    pub fn with_config(config: Option<ConnectionConfig>) -> Self {
        let mut default_config = ConnectionConfig::default();

        // Set default WebSocket configuration
        default_config.websocket_config.base_url = BINANCE_SPOT_WS_URL.into();
        default_config.websocket_config.ping_interval_milliseconds = DEFAULT_PING_INTERVAL_MS; // 30 seconds
        default_config.websocket_config.use_compression = false;

        // Set default REST API configuration
        default_config.rest_config.base_url = BINANCE_SPOT_API_URL.into();
        default_config.rest_config.timeout_milliseconds = 5000; // 5 seconds

        // Use custom config if provided
        let config = config.unwrap_or(default_config);
        let clock = config.clock.clone();

        // Create HTTP client with appropriate settings
        let mut headers = HeaderMap::new();
        headers.insert(
            "User-Agent",
            HeaderValue::from_str(&config.rest_config.user_agent)
                .unwrap_or_else(|_| HeaderValue::from_static("RustyHFT/1.0")),
        );

        let http_client = reqwest::Client::builder()
            .timeout(Duration::from_millis(
                config.rest_config.timeout_milliseconds,
            ))
            .connect_timeout(Duration::from_millis(
                config.rest_config.timeout_milliseconds / 2,
            ))
            .pool_max_idle_per_host(config.rest_config.connection_pool_size)
            .pool_idle_timeout(Duration::from_millis(
                config.rest_config.keep_alive_milliseconds,
            ))
            .default_headers(headers)
            .build()
            .unwrap_or_default();

        Self {
            config,
            connection_status: Arc::new(RwLock::new(ConnectionState::Disconnected)),
            stats: Arc::new(RwLock::new(ConnectionStats::default())),
            clock,
            subscriptions: Arc::new(RwLock::new(FxHashMap::default())),
            _instruments: Arc::new(RwLock::new(FxHashMap::default())),
            ws_handles: Arc::new(RwLock::new(FxHashMap::default())),
            last_connection_attempt: Arc::new(RwLock::new(Instant::now())),
            http_client,
        }
    }

    /// Helper function to update connection statistics upon message reception
    #[inline]
    fn update_receive_stats(
        stats: Arc<RwLock<ConnectionStats>>,
        message_size: usize,
        local_time: u64,
    ) {
        let mut s = stats.write();
        s.messages_received += 1;
        s.bytes_received += message_size as u64;
        s.last_message_time = local_time;
    }

    /// Helper function to create WebSocket request with proper headers
    fn create_websocket_config(
        url: String,
        use_compression: bool,
        ping_interval_milliseconds: u64,
        timeout_milliseconds: u64,
    ) -> WebSocketConfig {
        // Binance does NOT support WebSocket compression extensions
        // Adding compression headers can cause connection issues
        // Their streams are already optimized and binary messages are handled efficiently
        if use_compression {
            log::debug!(
                "Binance WebSocket does not support compression extensions, ignoring compression setting"
            );
        }

        WebSocketConfig::builder(rusty_common::types::Exchange::Binance, url.to_string())
            .ping_interval(Duration::from_millis(ping_interval_milliseconds))
            .timeout(Duration::from_millis(timeout_milliseconds))
            .compression(rusty_common::websocket::CompressionConfig::disabled())
            .build()
    }

    /// Connect to the WebSocket API
    #[inline]
    async fn connect(&self) -> Result<()> {
        if *self.connection_status.read() == ConnectionState::Connected {
            return Ok(());
        }

        // Check if we need to respect reconnection backoff
        {
            let now = Instant::now();
            let last_attempt = *self.last_connection_attempt.read();
            let backoff_ms = self
                .config
                .websocket_config
                .reconnect
                .backoff_initial_milliseconds;

            if now.duration_since(last_attempt) < Duration::from_millis(backoff_ms) {
                // Wait until backoff period is over
                tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
            }

            // Update last attempt time
            *self.last_connection_attempt.write() = Instant::now();
        }

        *self.connection_status.write() = ConnectionState::Connecting;

        // Note: Connection is now handled by individual trade/orderbook connections
        // using WebSocketClient pattern. This method is kept for compatibility.
        *self.connection_status.write() = ConnectionState::Connected;
        self.stats.write().connected_time = self.clock.raw();
        Ok(())
    }

    /// Create a WebSocket connection for trade data
    #[inline]
    async fn create_trade_connection(
        &self,
        symbols: SmallVec<[String; 8]>,
        tx: mpsc::Sender<TradeMessage>,
    ) -> Result<JoinHandle<()>> {
        // For multiple symbols, use combined streams
        let is_combined = symbols.len() > 1;

        let url = if is_combined {
            BINANCE_SPOT_WS_COMBINED_URL.to_string()
        } else {
            // For single symbol, use direct stream URL that doesn't require subscription
            if symbols.len() == 1 {
                let symbol_lower = symbols[0].to_lowercase();
                format!("wss://stream.binance.com:9443/ws/{symbol_lower}@trade")
            } else {
                BINANCE_SPOT_WS_URL.to_string()
            }
        };

        let connection_id = format!("trade-{}", symbols.join(","));

        // Create subscription message
        let subscription =
            create_trade_subscription(&connection_id, symbols.clone(), None, None, false);

        // Create stop signal
        let (stop_tx, stop_rx) = watch::channel(false);
        self.subscriptions
            .write()
            .insert(connection_id.clone().into(), stop_tx);

        // Clone parameters for the task
        let clock = self.clock.clone();
        let timeout_milliseconds = self.config.websocket_config.timeout_milliseconds;
        let ping_interval_milliseconds = self.config.websocket_config.ping_interval_milliseconds;
        let stats = self.stats.clone();
        let connection_status = self.connection_status.clone();
        let use_compression = self.config.websocket_config.use_compression;
        let use_zero_copy = self.config.websocket_config.use_zero_copy_parsing;

        // Spawn a task for the WebSocket connection
        let handle = tokio::spawn(async move {
            loop {
                // Check for stop signal
                if *stop_rx.borrow() {
                    break;
                }

                // Create WebSocket configuration
                let ws_config = Self::create_websocket_config(
                    url.clone().into(),
                    use_compression,
                    ping_interval_milliseconds,
                    timeout_milliseconds,
                );

                // Create WebSocket client
                let mut client = WebSocketClient::new(ws_config);

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

                // Run the WebSocket client with appropriate handler
                let result = if use_zero_copy {
                    let handler = BinanceSpotZeroCopyMessageHandler::new_trade_handler(
                        clock.clone(),
                        stats.clone(),
                        tx.clone(),
                        is_combined,
                        None, // We'll handle subscription differently
                    );
                    client.run(handler).await
                } else {
                    let handler = BinanceSpotMessageHandler::new_trade_handler(
                        clock.clone(),
                        stats.clone(),
                        tx.clone(),
                        is_combined,
                        None, // We'll handle subscription differently
                    );
                    client.run(handler).await
                };

                if let Err(e) = result {
                    log::error!("WebSocket client error: {e}");
                    *connection_status.write() = ConnectionState::Error;
                }

                // Check for stop signal before reconnecting
                if *stop_rx.borrow() {
                    break;
                }

                // Reconnect after a delay
                tokio::time::sleep(Duration::from_millis(1000)).await;
            }
        });

        Ok(handle)
    }

    /// Create a WebSocket connection for orderbook data
    #[inline]
    async fn create_orderbook_connection(
        &self,
        symbols: SmallVec<[String; 8]>,
        tx: mpsc::Sender<OrderbookMessage>,
    ) -> Result<JoinHandle<()>> {
        // For multiple symbols, use combined streams
        let is_combined = symbols.len() > 1;

        let url = if is_combined {
            BINANCE_SPOT_WS_COMBINED_URL.to_string()
        } else {
            BINANCE_SPOT_WS_URL.to_string()
        };

        let connection_id = format!("orderbook:{}", symbols.join(","));

        // Create subscription message
        let subscription = create_orderbook_subscription(
            &connection_id,
            symbols.clone(),
            None,
            None,
            None,
            None,
            false,
        );

        // Create stop signal
        let (stop_tx, stop_rx) = watch::channel(false);
        self.subscriptions
            .write()
            .insert(connection_id.clone().into(), stop_tx);

        // Clone parameters for the task
        let clock = self.clock.clone();
        let timeout_milliseconds = self.config.websocket_config.timeout_milliseconds;
        let ping_interval_milliseconds = self.config.websocket_config.ping_interval_milliseconds;
        let stats = self.stats.clone();
        let connection_status = self.connection_status.clone();
        let use_compression = self.config.websocket_config.use_compression;
        let use_zero_copy = self.config.websocket_config.use_zero_copy_parsing;

        // Spawn a task for the WebSocket connection
        let handle = tokio::spawn(async move {
            loop {
                // Check for stop signal
                if *stop_rx.borrow() {
                    break;
                }

                // Create WebSocket configuration
                let ws_config = Self::create_websocket_config(
                    url.clone().into(),
                    use_compression,
                    ping_interval_milliseconds,
                    timeout_milliseconds,
                );

                // Create WebSocket client
                let mut client = WebSocketClient::new(ws_config);

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

                // Run the WebSocket client with appropriate handler
                let result = if use_zero_copy {
                    let handler = BinanceSpotZeroCopyMessageHandler::new_orderbook_handler(
                        clock.clone(),
                        stats.clone(),
                        tx.clone(),
                        is_combined,
                        None, // We'll handle subscription differently
                    );
                    client.run(handler).await
                } else {
                    let handler = BinanceSpotMessageHandler::new_orderbook_handler(
                        clock.clone(),
                        stats.clone(),
                        tx.clone(),
                        is_combined,
                        None, // We'll handle subscription differently
                    );
                    client.run(handler).await
                };

                if let Err(e) = result {
                    log::error!("WebSocket client error: {e}");
                    *connection_status.write() = ConnectionState::Error;
                }

                // Check for stop signal before reconnecting
                if *stop_rx.borrow() {
                    break;
                }

                // Reconnect after a delay
                tokio::time::sleep(Duration::from_millis(1000)).await;
            }
        });

        Ok(handle)
    }

    /// Fetch orderbook snapshot for a symbol
    async fn fetch_orderbook_snapshot(
        &self,
        symbol: &str,
        limit: Option<u32>,
    ) -> Result<OrderbookSnapshot> {
        let limit = limit.unwrap_or(100);
        let url = format!(
            "{}/api/v3/depth?symbol={}&limit={}",
            self.config.rest_config.base_url, symbol, limit
        );

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

        if !response.status().is_success() {
            let error_text = response.text().await?;
            return Err(anyhow!(
                "Failed to fetch orderbook snapshot: {}",
                error_text
            ));
        }

        // Parse the response using simd_json
        let bytes = response
            .bytes()
            .await
            .map_err(|e| anyhow!("Failed to get response bytes: {}", e))?;
        let mut bytes_vec = bytes.to_vec();
        let snapshot = simd_json::from_slice::<OrderbookSnapshot>(&mut bytes_vec)
            .map_err(|e| anyhow!("Failed to parse JSON: {}", e))?;
        Ok(snapshot)
    }
}

#[async_trait]
impl Provider for BinanceSpotProvider {
    type TradeMessage = TradeMessage;
    type DepthMessage = OrderbookMessage;
    type InstrumentMessage = Value;

    fn name(&self) -> &'static str {
        "BinanceSpot"
    }

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

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

    async fn init(&mut self) -> Result<()> {
        self.connect().await
    }

    async fn shutdown(&mut self) -> Result<()> {
        // Unsubscribe from all active subscriptions
        let mut keys_to_remove = Vec::new();

        for (key, _) in self.subscriptions.read().iter() {
            keys_to_remove.push(key.clone());
        }

        // Send stop signal for each subscription
        for key in keys_to_remove {
            if let Some(tx) = self.subscriptions.write().remove(&key) {
                let _ = tx.send(true);
            }

            // Abort task
            if let Some(handle) = self.ws_handles.write().remove(&key) {
                handle.abort();
            }
        }

        // Set connection status to disconnected
        *self.connection_status.write() = ConnectionState::Disconnected;

        Ok(())
    }

    #[inline]
    async fn subscribe_trades(
        &self,
        symbols: SmallVec<[String; 8]>,
    ) -> Result<mpsc::Receiver<Self::TradeMessage>> {
        // Ensure connection is established
        self.connect().await?;

        // Create channel for trade messages
        let (tx, rx) = mpsc::channel(1024);

        // Create WebSocket connection
        let handle = self.create_trade_connection(symbols.clone(), tx).await?;

        // Save task handle
        let connection_id = format!("trade-{}", symbols.join(","));
        self.ws_handles.write().insert(connection_id.into(), handle);

        Ok(rx)
    }

    #[inline]
    async fn unsubscribe_trades(&self) -> Result<()> {
        // Find all trade subscriptions
        let mut keys_to_remove = Vec::new();

        for (key, _) in self.subscriptions.read().iter() {
            if key.starts_with("trade:") {
                keys_to_remove.push(key.clone());
            }
        }

        // Send stop signal for each subscription
        for key in keys_to_remove {
            if let Some(tx) = self.subscriptions.write().remove(&key) {
                let _ = tx.send(true);
            }

            // Abort task
            if let Some(handle) = self.ws_handles.write().remove(&key) {
                handle.abort();
            }
        }

        Ok(())
    }

    #[inline]
    async fn subscribe_orderbook(
        &self,
        symbols: SmallVec<[String; 8]>,
    ) -> Result<mpsc::Receiver<Self::DepthMessage>> {
        // Ensure connection is established
        self.connect().await?;

        // Create channel for orderbook messages
        let (tx, rx) = mpsc::channel(1024);

        // Create WebSocket connection
        let handle = self
            .create_orderbook_connection(symbols.clone(), tx)
            .await?;

        // Save task handle
        let connection_id = format!("orderbook:{}", symbols.join(","));
        self.ws_handles.write().insert(connection_id.into(), handle);

        Ok(rx)
    }

    #[inline]
    async fn unsubscribe_orderbook(&self) -> Result<()> {
        // Find all orderbook subscriptions
        let mut keys_to_remove = Vec::new();

        for (key, _) in self.subscriptions.read().iter() {
            if key.starts_with("orderbook:") {
                keys_to_remove.push(key.clone());
            }
        }

        // Send stop signal for each subscription
        for key in keys_to_remove {
            if let Some(tx) = self.subscriptions.write().remove(&key) {
                let _ = tx.send(true);
            }

            // Abort task
            if let Some(handle) = self.ws_handles.write().remove(&key) {
                handle.abort();
            }
        }

        Ok(())
    }

    #[inline]
    async fn get_instruments(&self) -> Result<Vec<Box<dyn Instrument>>> {
        // Define the endpoint for exchange info
        let url = "https://api.binance.com/api/v3/exchangeInfo";

        // Make the HTTP request
        let response = self
            .http_client
            .get(url)
            .send()
            .await
            .context("Failed to fetch Binance exchange info")?;

        // Check if the request was successful
        if !response.status().is_success() {
            return Err(anyhow!(
                "Failed to fetch Binance exchange info: HTTP {}",
                response.status()
            ));
        }

        // Parse the response using simd_json
        let bytes = response
            .bytes()
            .await
            .context("Failed to get response bytes")?;
        let mut bytes_vec = bytes.to_vec();
        let exchange_info: simd_json::OwnedValue = simd_json::from_slice(&mut bytes_vec)
            .context("Failed to parse Binance exchange info response")?;

        // Extract the symbols from the response
        let symbols = exchange_info["symbols"].as_array().ok_or_else(|| {
            anyhow!("Invalid response format: 'symbols' field not found or not an array")
        })?;

        // Convert to instruments
        let mut instruments = Vec::with_capacity(symbols.len());

        for symbol_data in symbols {
            // Skip symbols that are not trading
            let status = symbol_data["status"].as_str().unwrap_or("");
            if status != "TRADING" {
                continue;
            }

            let symbol = symbol_data["symbol"].as_str().ok_or_else(|| {
                anyhow!("Invalid symbol data: 'symbol' field not found or not a String")
            })?;

            // Create instrument
            let instrument_id = InstrumentId::new(symbol, Venue::Binance);
            let instrument =
                Box::new(BinanceInstrument { id: instrument_id }) as Box<dyn Instrument>;

            instruments.push(instrument);
        }

        Ok(instruments)
    }

    #[inline]
    async fn get_historical_trades(
        &self,
        _symbol: &str,
        _limit: Option<u32>,
    ) -> Result<Vec<MarketTrade>> {
        // Fetch historical trades from Binance API
        // Not fully implemented yet
        Err(anyhow!("Not implemented yet"))
    }

    #[inline]
    async fn get_orderbook_snapshot(
        &self,
        symbol: &str,
        depth: Option<u32>,
    ) -> Result<OrderBookSnapshot> {
        // Fetch orderbook snapshot
        let snapshot = self.fetch_orderbook_snapshot(symbol, depth).await?;
        let parsed = ParsedOrderbookSnapshot::from(snapshot);

        // Convert to model OrderBookSnapshot
        let instrument_id = InstrumentId::new(symbol, Venue::Binance);

        // Create OrderBookSnapshot with current timestamp (nanoseconds since epoch)
        let timestamp = self.clock.raw();

        // Convert to OrderBookSnapshot
        let mut order_book_depth =
            OrderBookSnapshot::new_empty(instrument_id, timestamp, parsed.last_update_id);

        // Add bids and asks
        for (price, quantity) in parsed.bids {
            order_book_depth.add_bid(price, quantity);
        }

        for (price, quantity) in parsed.asks {
            order_book_depth.add_ask(price, quantity);
        }

        Ok(order_book_depth)
    }

    #[inline]
    async fn get_realtime_orderbook(&self, _symbol: &str) -> Result<SharedSimdOrderBook> {
        // Not yet implemented
        Err(anyhow!("Not implemented yet"))
    }

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

    #[inline]
    async fn connection_status(&self) -> ConnectionState {
        *self.connection_status.read()
    }

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

    #[inline]
    async fn ping(&self) -> Result<u64> {
        // For WebSocketClient pattern, ping/pong is handled automatically
        // Return a simple network latency test via HTTP
        let start = self.clock.raw();

        let response = self
            .http_client
            .get(format!("{}/api/v3/ping", self.config.rest_config.base_url))
            .send()
            .await?;

        if response.status().is_success() {
            Ok(self.clock.raw().saturating_sub(start))
        } else {
            Err(anyhow!("Ping failed with status: {}", response.status()))
        }
    }

    #[inline]
    async fn reset_connection(&self) -> Result<()> {
        // First, unsubscribe all active subscriptions
        let mut keys_to_remove = Vec::new();

        for (key, _) in self.subscriptions.read().iter() {
            keys_to_remove.push(key.clone());
        }

        // Send stop signal for each subscription
        for key in &keys_to_remove {
            if let Some(tx) = self.subscriptions.write().remove(key) {
                let _ = tx.send(true);
            }

            // Abort task
            if let Some(handle) = self.ws_handles.write().remove(key) {
                handle.abort();
            }
        }

        // Set status to disconnected
        *self.connection_status.write() = ConnectionState::Disconnected;

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

        // Connect again
        self.connect().await
    }

    #[inline]
    fn get_rate_limits(&self) -> Vec<RateLimit> {
        BINANCE_SPOT_RATE_LIMITS.to_vec()
    }
}