rusty_feeder/exchange/bithumb/

provider.rs

//! Bithumb exchange provider implementation
//!
//! High-performance provider for Bithumb markets with WebSocket streaming
//! and REST API support. Optimized for HFT applications with nanosecond precision.

use rusty_common::collections::FxHashMap;
use std::any::Any;
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::common::rest_utils::{current_timestamp_ms, get_request};
use crate::exchange::zerocopy_helpers::{deserialize_from_slice_borrowed, deserialize_from_vec};
use crate::provider::prelude::*;
use anyhow::{Result, anyhow};
use async_trait::async_trait;
use parking_lot::RwLock;
use quanta::Clock;
use rust_decimal::Decimal;
use rusty_common::auth::exchanges::bithumb::BithumbAuth;
use rusty_common::json;
use rusty_common::websocket::{
    Message, MessageHandler, WebSocketClient, WebSocketConfig, WebSocketError, WebSocketResult,
};
use rusty_model::{
    data::{
        book_snapshot::OrderBookSnapshot,
        market_trade::MarketTrade,
        orderbook::PriceLevel,
        simd_orderbook::{SharedSimdOrderBook, SimdOrderBook},
    },
    enums::OrderSide,
    instruments::{Instrument, InstrumentId},
    venues::Venue,
};
use smallvec::SmallVec;
use smartstring::alias::String;
use tokio::sync::{mpsc, watch};
use tokio::task::JoinHandle;

use super::{
    data::{
        orderbook::{Orderbook, OrderbookMessage},
        subscription::{BithumbSubscription, ResponseFormat, StatusMessage},
        trade::TradeMessage,
    },
    types::{BITHUMB_PRIVATE_WS_URL, BITHUMB_PUBLIC_WS_URL, BITHUMB_RATE_LIMITS, TicketRequest},
};

/// Bithumb instrument implementation
#[derive(Debug, Clone)]
pub struct BithumbInstrument {
    /// Trading pair symbol (e.g., "KRW-BTC")
    pub symbol: String,
    /// Base currency (e.g., "BTC")
    pub base_currency: String,
    /// Quote currency (e.g., "KRW")
    pub quote_currency: String,
    /// Minimum allowed price for orders
    pub min_price: Decimal,
    /// Maximum allowed price for orders
    pub max_price: Decimal,
    /// Minimum price increment (tick size)
    pub price_tick: Decimal,
    /// Minimum order quantity
    pub min_quantity: Decimal,
    /// Maximum order quantity
    pub max_quantity: Decimal,
}

impl Instrument for BithumbInstrument {
    fn id(&self) -> InstrumentId {
        InstrumentId::new(self.symbol.clone(), Venue::Bithumb)
    }

    fn symbol(&self) -> String {
        self.symbol.clone()
    }

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

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn clone_box(&self) -> Box<dyn Instrument> {
        Box::new(self.clone())
    }
}

/// Thread-safe wrapper for BithumbInstrument
#[derive(Debug, Clone)]
struct ThreadSafeInstrument {
    symbol: String,
    base_currency: String,
    quote_currency: String,
    min_price: Decimal,
    max_price: Decimal,
    price_tick: Decimal,
    min_quantity: Decimal,
    max_quantity: Decimal,
}

impl ThreadSafeInstrument {
    fn from_bithumb_instrument(instrument: &BithumbInstrument) -> Self {
        Self {
            symbol: instrument.symbol.clone(),
            base_currency: instrument.base_currency.clone(),
            quote_currency: instrument.quote_currency.clone(),
            min_price: instrument.min_price,
            max_price: instrument.max_price,
            price_tick: instrument.price_tick,
            min_quantity: instrument.min_quantity,
            max_quantity: instrument.max_quantity,
        }
    }

    fn to_bithumb_instrument(&self) -> BithumbInstrument {
        BithumbInstrument {
            symbol: self.symbol.clone(),
            base_currency: self.base_currency.clone(),
            quote_currency: self.quote_currency.clone(),
            min_price: self.min_price,
            max_price: self.max_price,
            price_tick: self.price_tick,
            min_quantity: self.min_quantity,
            max_quantity: self.max_quantity,
        }
    }
}

// This is safe to share between threads
unsafe impl Send for ThreadSafeInstrument {}
unsafe impl Sync for ThreadSafeInstrument {}

/// Bithumb market data provider
#[derive(Debug)]
pub struct BithumbProvider {
    /// 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, ThreadSafeInstrument>>>,

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

    /// Authentication handler (optional, for private endpoints)
    auth: Option<BithumbAuth>,
}

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

impl BithumbProvider {
    /// Create a new Bithumb provider with default configuration
    #[must_use]
    pub fn new() -> Self {
        Self::with_config(ConnectionConfig::default())
    }

    /// Create a new Bithumb provider with custom configuration
    #[must_use]
    pub fn with_config(config: ConnectionConfig) -> Self {
        let clock = config.clock.clone();

        // Create auth if credentials are provided
        let auth =
            if config.auth_config.api_key.is_some() && config.auth_config.secret_key.is_some() {
                let auth_config = BithumbAuth::new(
                    config.auth_config.api_key.clone().unwrap(),
                    config.auth_config.secret_key.clone().unwrap(),
                );
                Some(auth_config)
            } else {
                None
            };

        // Create HTTP client with appropriate settings
        let mut headers = reqwest::header::HeaderMap::new();
        headers.insert(
            "User-Agent",
            reqwest::header::HeaderValue::from_str(&config.rest_config.user_agent)
                .unwrap_or_else(|_| reqwest::header::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,
            auth,
        }
    }

    /// Create WebSocket configuration
    fn create_ws_config(&self, private: bool) -> WebSocketConfig {
        let url = if private {
            BITHUMB_PRIVATE_WS_URL.to_string()
        } else {
            BITHUMB_PUBLIC_WS_URL.to_string()
        };

        WebSocketConfig::builder(rusty_common::types::Exchange::Bithumb, url)
            .ping_interval(Duration::from_millis(
                self.config.websocket_config.ping_interval_milliseconds,
            ))
            .timeout(Duration::from_millis(
                self.config.websocket_config.timeout_milliseconds,
            ))
            .compression(rusty_common::websocket::CompressionConfig::disabled())
            .build()
    }

    /// Create a WebSocket connection for trade data
    async fn create_trade_connection(
        &self,
        symbols: SmallVec<[String; 8]>,
        tx: mpsc::Sender<TradeMessage>,
    ) -> Result<JoinHandle<()>> {
        let connection_id = format!("trade-{}", symbols.join(","));

        // Create subscription messages using Bithumb format
        let ticket = TicketRequest {
            ticket: String::from(uuid::Uuid::new_v4().to_string()),
        };
        let subscription = BithumbSubscription::trade(symbols.to_vec());
        let format = ResponseFormat::default();

        // Create a batch subscription
        let batch = simd_json::OwnedValue::Array(Box::new(vec![
            simd_json::serde::to_owned_value(&ticket)?,
            simd_json::serde::to_owned_value(&subscription)?,
            simd_json::serde::to_owned_value(&format)?,
        ]));

        // 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 stats = self.stats.clone();
        let connection_status = self.connection_status.clone();
        let ws_config = self.create_ws_config(false); // Public connection for trades

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

                // Create message handler
                let handler = BithumbMessageHandler::new_trade_handler(
                    clock.clone(),
                    stats.clone(),
                    tx.clone(),
                );

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

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

                // Run the WebSocket client
                if let Err(e) = client.run(handler).await {
                    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
    async fn create_orderbook_connection(
        &self,
        symbols: SmallVec<[String; 8]>,
        tx: mpsc::Sender<OrderbookMessage>,
    ) -> Result<JoinHandle<()>> {
        let connection_id = format!("orderbook:{}", symbols.join(","));

        // Create subscription messages using Bithumb format
        let ticket = TicketRequest {
            ticket: String::from(uuid::Uuid::new_v4().to_string()),
        };
        let subscription = BithumbSubscription::orderbook(symbols.to_vec(), None);
        let format = ResponseFormat::default();

        // Create a batch subscription
        let batch = simd_json::OwnedValue::Array(Box::new(vec![
            simd_json::serde::to_owned_value(&ticket)?,
            simd_json::serde::to_owned_value(&subscription)?,
            simd_json::serde::to_owned_value(&format)?,
        ]));

        // 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 stats = self.stats.clone();
        let connection_status = self.connection_status.clone();
        let ws_config = self.create_ws_config(false); // Public connection for orderbook

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

                // Create message handler
                let handler = BithumbMessageHandler::new_orderbook_handler(
                    clock.clone(),
                    stats.clone(),
                    tx.clone(),
                );

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

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

                // Run the WebSocket client
                if let Err(e) = client.run(handler).await {
                    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)
    }

    /// Convert exchange timestamp to nanosecond precision
    #[inline]
    const fn convert_exchange_timestamp(&self, timestamp_ms: u64) -> u64 {
        timestamp_ms * 1_000_000 // Convert ms to ns
    }
}

/// Message handler for Bithumb WebSocket
struct BithumbMessageHandler {
    clock: Clock,
    stats: Arc<RwLock<ConnectionStats>>,
    trade_tx: Option<mpsc::Sender<TradeMessage>>,
    orderbook_tx: Option<mpsc::Sender<OrderbookMessage>>,
}

#[async_trait]
impl MessageHandler for BithumbMessageHandler {
    async fn on_connected(&mut self) -> WebSocketResult<()> {
        log::info!("Bithumb WebSocket connected");
        Ok(())
    }

    async fn on_message(&mut self, message: Message) -> WebSocketResult<()> {
        match message {
            Message::Text(text) => {
                // Zero-copy parsing: parse directly from text bytes
                match deserialize_from_slice_borrowed::<json::Value>(text.as_bytes()) {
                    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(data) => {
                // Zero-copy parsing: parse directly from binary data without string conversion
                match deserialize_from_vec::<json::Value>(data) {
                    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 binary WebSocket message: {e}");
                    }
                }
            }
            _ => {}
        }
        Ok(())
    }

    async fn on_error(&mut self, error: WebSocketError) -> WebSocketResult<()> {
        log::error!("Bithumb WebSocket error: {}", error);
        Ok(())
    }

    async fn on_disconnected(&mut self) -> WebSocketResult<()> {
        log::info!("Bithumb WebSocket disconnected");
        Ok(())
    }
}

impl BithumbMessageHandler {
    const fn new_trade_handler(
        clock: Clock,
        stats: Arc<RwLock<ConnectionStats>>,
        tx: mpsc::Sender<TradeMessage>,
    ) -> Self {
        Self {
            clock,
            stats,
            trade_tx: Some(tx),
            orderbook_tx: None,
        }
    }

    const fn new_orderbook_handler(
        clock: Clock,
        stats: Arc<RwLock<ConnectionStats>>,
        tx: mpsc::Sender<OrderbookMessage>,
    ) -> Self {
        Self {
            clock,
            stats,
            trade_tx: None,
            orderbook_tx: Some(tx),
        }
    }

    async fn handle_market_data(&self, value: &json::Value) -> Result<()> {
        // Try to parse as different message types
        // First check if it's a trade message
        if let Ok(trade_msg) = json::from_value::<TradeMessage>(value.clone()) {
            return self.handle_trade_message(trade_msg).await;
        }

        // Check if it's an orderbook message
        if let Ok(orderbook_msg) = json::from_value::<OrderbookMessage>(value.clone()) {
            return self.handle_orderbook_message(orderbook_msg).await;
        }

        // Check if it's a status message
        if let Ok(_status) = json::from_value::<StatusMessage>(value.clone()) {
            log::debug!("Received status message");
            return Ok(());
        }

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

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

        // Forward to trade handler if available
        if let Some(tx) = &self.trade_tx
            && let Err(e) = tx.send(msg).await
        {
            log::error!("Failed to send trade message: {e}");
        }

        Ok(())
    }

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

        // Forward to orderbook handler if available
        if let Some(tx) = &self.orderbook_tx
            && let Err(e) = tx.send(msg).await
        {
            log::error!("Failed to send orderbook message: {e}");
        }

        Ok(())
    }
}

#[async_trait]
impl Provider for BithumbProvider {
    type TradeMessage = TradeMessage;
    type DepthMessage = OrderbookMessage;
    type InstrumentMessage = BithumbInstrument;

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

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

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

    async fn init(&mut self) -> Result<()> {
        // Initialize the Bithumb provider
        // Fetch initial instruments to populate cache
        match self.get_instruments().await {
            Ok(_) => Ok(()),
            Err(e) => Err(anyhow!("Failed to initialize Bithumb provider: {}", e)),
        }
    }

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

    async fn get_realtime_orderbook(&self, symbol: &str) -> Result<SharedSimdOrderBook> {
        // Create a shared orderbook
        let instrument_id = InstrumentId::new(symbol, Venue::Bithumb);
        let ob = rusty_model::data::orderbook::OrderBook::<64>::new_empty(instrument_id);
        let orderbook = SharedSimdOrderBook::from_orderbook(&ob);

        let (_tx, mut rx) = mpsc::channel(100);
        let mut symbol_list = SmallVec::<[String; 8]>::new();
        symbol_list.push(symbol.into());
        let _depth_rx = self.subscribe_orderbook(symbol_list).await?;

        // Subscribe to the orderbook updates
        let orderbook_clone = orderbook.clone();
        let clock = self.clock.clone();

        tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                let now = clock.raw();
                let ob = Orderbook::from_orderbook_message(&msg, now);

                // Convert to OrderBook format and update
                let bids_vec: SmallVec<[PriceLevel; 64]> = ob
                    .bids
                    .iter()
                    .map(|level| PriceLevel::new(level.price, level.quantity))
                    .collect();
                let asks_vec: SmallVec<[PriceLevel; 64]> = ob
                    .asks
                    .iter()
                    .map(|level| PriceLevel::new(level.price, level.quantity))
                    .collect();

                let orderbook = rusty_model::data::orderbook::OrderBook::new(
                    ob.code.clone(),
                    ob.timestamp_ns,
                    now,
                    bids_vec,
                    asks_vec,
                );
                orderbook_clone.write(|book| {
                    *book = SimdOrderBook::from_orderbook(&orderbook);
                });
            }
        });

        Ok(orderbook)
    }

    async fn subscribe_trades(
        &self,
        symbols: SmallVec<[String; 8]>,
    ) -> Result<mpsc::Receiver<Self::TradeMessage>> {
        // 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)
    }

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

    async fn subscribe_orderbook(
        &self,
        symbols: SmallVec<[String; 8]>,
    ) -> Result<mpsc::Receiver<Self::DepthMessage>> {
        // 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)
    }

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

    async fn get_instruments(&self) -> Result<Vec<Box<dyn Instrument>>> {
        // First, check if we have cached instruments
        {
            let instruments = self.instruments.read();
            if !instruments.is_empty() {
                // Convert the ThreadSafeInstrument to Box<dyn Instrument>
                let boxed_instruments: Vec<Box<dyn Instrument>> = instruments
                    .values()
                    .map(|ts_instrument| {
                        let bithumb_instrument = ts_instrument.to_bithumb_instrument();
                        Box::new(bithumb_instrument) as Box<dyn Instrument>
                    })
                    .collect();
                return Ok(boxed_instruments);
            }
        }

        // If not, fetch from API
        let client = &self.http_client;

        #[derive(serde::Deserialize)]
        struct TickerResponse {
            status: String,
            data: FxHashMap<String, TickerData>,
        }

        #[derive(serde::Deserialize)]
        #[allow(dead_code)]
        struct TickerData {
            opening_price: String,
            closing_price: String,
            min_price: String,
            max_price: String,
            units_traded: String,
            acc_trade_value: String,
            prev_closing_price: String,
            #[serde(rename = "units_traded_24H")]
            units_traded_24h: String,
            #[serde(rename = "acc_trade_value_24H")]
            acc_trade_value_24h: String,
            #[serde(rename = "fluctate_24H")]
            fluctate_24h: String,
            #[serde(rename = "fluctate_rate_24H")]
            fluctate_rate_24h: String,
        }

        // Get all tickers
        let endpoint = "/public/ticker/ALL_KRW";
        let url = format!("https://api.bithumb.com{endpoint}");
        let tickers: TickerResponse = get_request(client, &url).await?;

        if tickers.status != "0000" {
            return Err(anyhow!("Failed to fetch tickers: {}", tickers.status));
        }

        // Create a vector to collect our thread-safe instruments and Box<dyn Instrument> result
        let mut instrument_boxes: SmallVec<[Box<dyn Instrument>; 128]> =
            SmallVec::<[Box<dyn Instrument>; 128]>::new();
        let mut thread_safe_instruments = SmallVec::<[(String, ThreadSafeInstrument); 128]>::new();

        // Process the ticker data
        for (symbol, ticker_data) in tickers.data.iter().filter(|(symbol, _)| *symbol != "date") {
            // Try to extract min/max price from ticker data
            let min_price = ticker_data
                .min_price
                .parse::<Decimal>()
                .unwrap_or(Decimal::ZERO);
            let max_price = ticker_data
                .max_price
                .parse::<Decimal>()
                .unwrap_or(Decimal::ZERO);

            // For price_tick, we'll use a reasonable default based on the price range
            // This is an approximation as Bithumb doesn't directly provide tick size
            let price_tick = if max_price > Decimal::ZERO {
                // For higher priced assets, use larger tick size
                if max_price > Decimal::from(1_000_000) {
                    Decimal::from(100)
                } else if max_price > Decimal::from(100_000) {
                    Decimal::from(10)
                } else if max_price > Decimal::from(10_000) {
                    Decimal::from(1)
                } else if max_price > Decimal::from(1_000) {
                    Decimal::from_str("0.1").unwrap_or(Decimal::ONE)
                } else if max_price > Decimal::from(100) {
                    Decimal::from_str("0.01").unwrap_or(Decimal::ONE)
                } else {
                    Decimal::from_str("0.001").unwrap_or(Decimal::ONE)
                }
            } else {
                Decimal::ONE
            };

            // For min/max quantity, we'll use reasonable defaults
            // Bithumb typically has minimum order sizes but they're not in the API
            let min_quantity = Decimal::from_str("0.0001").unwrap_or(Decimal::ZERO); // Common minimum for crypto
            let max_quantity = Decimal::from(10000); // Large default

            // Create a BithumbInstrument
            let bithumb_instrument = BithumbInstrument {
                symbol: format!("KRW-{symbol}").into(), // Format to match WebSocket format
                base_currency: symbol.clone(),
                quote_currency: "KRW".into(),
                min_price,
                max_price,
                price_tick,
                min_quantity,
                max_quantity,
            };

            // Create a thread-safe version
            let thread_safe = ThreadSafeInstrument::from_bithumb_instrument(&bithumb_instrument);

            // Store both versions
            thread_safe_instruments.push((bithumb_instrument.symbol.clone(), thread_safe));
            instrument_boxes.push(Box::new(bithumb_instrument) as Box<dyn Instrument>);
        }

        // Cache the thread-safe instruments
        let mut cache = self.instruments.write();
        for (symbol, instrument) in thread_safe_instruments {
            cache.insert(symbol, instrument);
        }

        Ok(instrument_boxes.into_vec())
    }

    async fn get_historical_trades(
        &self,
        symbol: &str,
        limit: Option<u32>,
    ) -> Result<Vec<MarketTrade>> {
        let client = &self.http_client;
        let mut params: FxHashMap<String, String> = FxHashMap::default();

        // Parse symbol to get base currency (KRW-BTC -> BTC)
        let parts: SmallVec<[&str; 4]> = symbol.split('-').collect();
        let base_currency = if parts.len() >= 2 { parts[1] } else { symbol };
        let quote_currency = if parts.len() >= 2 { parts[0] } else { "KRW" };

        params.insert("order_currency".into(), base_currency.into());
        params.insert("payment_currency".into(), quote_currency.into());

        if let Some(limit) = limit {
            params.insert("count".into(), limit.to_string().into());
        }

        let endpoint = "/public/transaction_history";

        #[derive(serde::Deserialize)]
        struct TransactionHistoryResponse {
            status: String,
            data: Vec<TransactionHistoryEntry>,
        }

        #[derive(serde::Deserialize)]
        #[allow(dead_code)]
        struct TransactionHistoryEntry {
            transaction_date: String,
            r#type: String, // "bid" for buy, "ask" for sell
            units_traded: String,
            price: String,
            total: String,
        }

        // Build URL with query parameters
        let mut url = format!("https://api.bithumb.com{endpoint}");
        let query_string = params
            .iter()
            .map(|(k, v)| format!("{k}={v}"))
            .collect::<SmallVec<[_; 8]>>()
            .join("&");
        if !query_string.is_empty() {
            url.push('?');
            url.push_str(&query_string);
        }

        let trades: TransactionHistoryResponse = get_request(client, &url).await?;

        if trades.status != "0000" {
            return Err(anyhow!("Failed to fetch trades: {}", trades.status));
        }

        // Convert to our Trade type
        let clock = Clock::new();
        let result = trades
            .data
            .into_iter()
            .filter_map(|t| {
                let price = match t.price.parse() {
                    Ok(p) => p,
                    Err(_) => return None,
                };

                let quantity = match t.units_traded.parse() {
                    Ok(q) => q,
                    Err(_) => return None,
                };

                let direction = match t.r#type.as_str() {
                    "bid" => OrderSide::Buy,
                    "ask" => OrderSide::Sell,
                    _ => return None,
                };

                // Parse the transaction date to get timestamp
                let exchange_time = match t.transaction_date.parse::<u64>() {
                    Ok(ts) => ts * 1_000_000, // Convert from ms to ns
                    Err(_) => clock.raw(),    // Fallback to current time
                };

                Some(MarketTrade {
                    timestamp: clock.now(),
                    exchange_time_ns: exchange_time,
                    price,
                    quantity,
                    direction,
                    instrument_id: InstrumentId::new(symbol, Venue::Bithumb),
                })
            })
            .collect();

        Ok(result)
    }

    async fn get_orderbook_snapshot(
        &self,
        symbol: &str,
        depth: Option<u32>,
    ) -> Result<OrderBookSnapshot> {
        let client = &self.http_client;
        let mut params: FxHashMap<String, String> = FxHashMap::default();

        // Parse symbol to get base currency (KRW-BTC -> BTC)
        let parts: SmallVec<[&str; 4]> = symbol.split('-').collect();
        let base_currency = if parts.len() >= 2 { parts[1] } else { symbol };
        let quote_currency = if parts.len() >= 2 { parts[0] } else { "KRW" };

        params.insert("order_currency".into(), base_currency.into());
        params.insert("payment_currency".into(), quote_currency.into());

        if let Some(depth) = depth {
            params.insert("count".into(), depth.to_string().into());
        }

        let endpoint = "/public/orderbook";

        #[derive(serde::Deserialize)]
        struct OrderbookRestResponse {
            status: String,
            data: OrderbookRestData,
        }

        #[derive(serde::Deserialize)]
        #[allow(dead_code)]
        struct OrderbookRestData {
            timestamp: String,
            order_currency: String,
            payment_currency: String,
            bids: Vec<OrderbookRestEntry>,
            asks: Vec<OrderbookRestEntry>,
        }

        #[derive(serde::Deserialize)]
        struct OrderbookRestEntry {
            quantity: String,
            price: String,
        }

        // Build URL with query parameters
        let mut url = format!("https://api.bithumb.com{endpoint}");
        let query_string = params
            .iter()
            .map(|(k, v)| format!("{k}={v}"))
            .collect::<SmallVec<[_; 8]>>()
            .join("&");
        if !query_string.is_empty() {
            url.push('?');
            url.push_str(&query_string);
        }

        let orderbook: OrderbookRestResponse = get_request(client, &url).await?;

        if orderbook.status != "0000" {
            return Err(anyhow!("Failed to fetch orderbook: {}", orderbook.status));
        }

        // Convert to our OrderBookSnapshot type
        let mut bids: SmallVec<[PriceLevel; 64]> =
            SmallVec::with_capacity(orderbook.data.bids.len());
        let mut asks: SmallVec<[PriceLevel; 64]> =
            SmallVec::with_capacity(orderbook.data.asks.len());

        for bid in &orderbook.data.bids {
            let price = match bid.price.parse() {
                Ok(p) => p,
                Err(_) => continue,
            };

            let quantity = match bid.quantity.parse() {
                Ok(q) => q,
                Err(_) => continue,
            };

            bids.push(PriceLevel::new(price, quantity));
        }

        for ask in &orderbook.data.asks {
            let price = match ask.price.parse() {
                Ok(p) => p,
                Err(_) => continue,
            };

            let quantity = match ask.quantity.parse() {
                Ok(q) => q,
                Err(_) => continue,
            };

            asks.push(PriceLevel::new(price, quantity));
        }

        // Sort bids in descending order, asks in ascending order
        bids.sort_by(|a, b| {
            b.price
                .partial_cmp(&a.price)
                .unwrap_or(std::cmp::Ordering::Equal)
        });
        asks.sort_by(|a, b| {
            a.price
                .partial_cmp(&b.price)
                .unwrap_or(std::cmp::Ordering::Equal)
        });

        let timestamp_ms = match orderbook.data.timestamp.parse::<u64>() {
            Ok(t) => t,
            Err(_) => current_timestamp_ms(),
        };

        // Convert millisecond timestamp to nanoseconds for HFT precision
        let exchange_time = self.convert_exchange_timestamp(timestamp_ms);

        let book = OrderBookSnapshot::new(
            rusty_model::instruments::InstrumentId::new(symbol, Venue::Bithumb),
            bids,
            asks,
            0, // No sequence number from Bithumb
            exchange_time,
            self.clock.raw(),
        );

        Ok(book)
    }

    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> {
        // 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("https://api.bithumb.com/public/ticker/ALL_KRW")
            .send()
            .await?;

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

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

        Ok(())
    }

    fn get_rate_limits(&self) -> Vec<RateLimit> {
        BITHUMB_RATE_LIMITS.to_vec()
    }
}