rusty_ems/exchanges/

bithumb_websocket_trading.rs

//! Bithumb WebSocket Trading Implementation
//!
//! This module provides a high-performance WebSocket trading client for Bithumb's private endpoints
//! with JWT authentication, real-time order updates, and asset tracking.
//!
//! # Features
//!
//! - **JWT authentication**: Support for Bithumb's JWT-based WebSocket authentication
//! - **Private channels**: Real-time myOrder and myAsset data streams
//! - **Order management**: Integration with REST API for order placement
//! - **Asset tracking**: Real-time balance and position updates
//! - **Connection health monitoring**: Real-time metrics and health status
//! - **Automatic reconnection**: Configurable backoff strategy with exponential delays
//! - **Zero-copy message processing**: Optimized for low-latency trading
//!
//! # Connection Management
//!
//! The client maintains connection health through:
//! - WebSocket ping/pong frames sent every 2 minutes
//! - Automatic reconnection on connection loss
//! - JWT token refresh before expiration
//! - Configurable timeout handling
//!
//! # Private Channels
//!
//! Supports Bithumb's private WebSocket channels:
//! - `myOrder`: Real-time order updates (creation/execution/cancellation)
//! - `myAsset`: Real-time asset/balance updates
//! - Event-driven updates only (no snapshot data)
//!
//! # Usage Example
//!
//! ```rust,no_run
//! use rusty_common::auth::exchanges::bithumb::BithumbAuth;
//! use rusty_ems::exchanges::bithumb_websocket_trading::BithumbWebSocketTrader;
//! use std::sync::Arc;
//!
//! #[tokio::main]
//! async fn main() -> Result<()> {
//!     let auth = Arc::new(BithumbAuth::new("access_key".into(), "secret_key".into()));
//!
//!     let trader = BithumbWebSocketTrader::new(auth);
//!     let (report_tx, report_rx) = flume::bounded(100);
//!
//!     // Connect and authenticate
//!     trader.connect(report_tx).await?;
//!
//!     Ok(())
//! }
//! ```

use anyhow::{Result, bail};
use async_trait::async_trait;
use flume::Sender;
use futures::{SinkExt, StreamExt};
use log::{debug, error, info, warn};
use parking_lot::RwLock;
use quanta::Clock;
use rust_decimal::Decimal;
use rusty_common::auth::exchanges::bithumb::{BithumbAuth, header_keys};
use rusty_common::types::Exchange;
use rusty_common::utils::id_generation;
use rusty_common::websocket::{
    Message, WebSocketConfig,
    client::ConnectionState as WsConnectionState,
    connector::{WebSocketConnector, WebSocketSink, WebSocketStream},
    stats::new_shared_stats,
};
use rusty_model::{
    enums::OrderStatus, instruments::InstrumentId, trading_order::Order, venues::Venue,
};

use super::bithumb_errors::map_websocket_order_state;
#[cfg(test)]
use super::bithumb_errors::map_websocket_order_status;
use simd_json;
use simd_json::prelude::{ValueAsArray, ValueAsScalar, ValueObjectAccess};
use smartstring::alias::String as SmartString;
use std::sync::Arc;
use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
use std::time::Duration;
use tokio::sync::RwLock as AsyncRwLock;
use tokio::task::JoinHandle;

use crate::execution_engine::ExecutionReport;

/// Bithumb private WebSocket API URL
const BITHUMB_PRIVATE_WS_URL: &str = "wss://ws-api.bithumb.com/websocket/v1/private";

/// Ping interval for keeping WebSocket connection alive (2 minutes)
const PING_INTERVAL_SECONDS: u64 = 120;

/// Pong timeout (10 seconds - if no pong received within this time, consider connection dead)
const PONG_TIMEOUT_SECONDS: u64 = 10;

/// Maximum message size allowed (10MB)
const MAX_MESSAGE_SIZE: usize = 10 * 1024 * 1024;

/// Bithumb-specific connection state for proper state machine tracking
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum BithumbConnectionState {
    /// WebSocket is not connected
    Disconnected = 0,
    /// WebSocket is attempting to connect
    Connecting = 1,
    /// WebSocket is connected but not authenticated
    Connected = 2,
    /// WebSocket is attempting to authenticate
    Authenticating = 3,
    /// WebSocket is connected and authenticated
    Authenticated = 4,
    /// WebSocket is disconnecting
    Disconnecting = 5,
}

impl From<u8> for BithumbConnectionState {
    fn from(value: u8) -> Self {
        match value {
            0 => Self::Disconnected,
            1 => Self::Connecting,
            2 => Self::Connected,
            3 => Self::Authenticating,
            4 => Self::Authenticated,
            5 => Self::Disconnecting,
            _ => Self::Disconnected,
        }
    }
}

/// Connection health information for Bithumb WebSocket
#[derive(Debug, Clone)]
pub struct BithumbConnectionHealth {
    /// Current connection state
    pub state: BithumbConnectionState,
    /// Whether the connection is currently active
    pub is_connected: bool,
    /// Whether the connection is authenticated
    pub is_authenticated: bool,
    /// Time elapsed since the last ping was sent
    pub time_since_last_ping: Option<Duration>,
    /// Time elapsed since the last pong was received
    pub time_since_last_pong: Option<Duration>,
    /// Time elapsed since authentication was completed
    pub time_since_auth: Option<Duration>,
    /// Total number of messages sent
    pub messages_sent: u64,
    /// Total number of messages received
    pub messages_received: u64,
    /// Number of reconnection attempts made
    pub reconnection_attempts: u64,
}

/// Connection metrics for monitoring
#[derive(Debug, Default)]
struct ConnectionMetrics {
    pings_sent: AtomicU64,
    pongs_received: AtomicU64,
    messages_sent: AtomicU64,
    messages_received: AtomicU64,
    reconnection_attempts: AtomicU64,
}

/// Task handles for lifecycle management
#[derive(Debug, Default)]
struct TaskHandles {
    ping_handle: Option<JoinHandle<()>>,
    message_handler: Option<JoinHandle<()>>,
}

/// Bithumb WebSocket Trading Client
///
/// Provides real-time market data and private account updates via WebSocket.
/// Trading operations are handled through the REST API integration.
pub struct BithumbWebSocketTrader {
    /// Authentication handler for Bithumb API
    auth: Arc<BithumbAuth>,

    /// WebSocket connection for private data
    ws_sink: Arc<AsyncRwLock<Option<WebSocketSink>>>,
    ws_stream: Arc<AsyncRwLock<Option<WebSocketStream>>>,

    /// Connection state tracking
    state: Arc<AtomicU8>,

    /// High-precision clock for timestamp generation
    clock: Clock,

    /// Last ping sent timestamp (nanoseconds, monotonic)
    last_ping_time: Arc<AtomicU64>,

    /// Last pong received timestamp (nanoseconds, monotonic)
    last_pong_time: Arc<AtomicU64>,

    /// Authentication completed timestamp (nanoseconds, monotonic)
    auth_completed_time: Arc<AtomicU64>,

    /// Task handles for lifecycle management
    task_handles: Arc<AsyncRwLock<TaskHandles>>,

    /// Connection metrics
    metrics: Arc<ConnectionMetrics>,

    /// Connection statistics
    messages_received: Arc<AtomicU64>,
    messages_sent: Arc<AtomicU64>,
    reconnect_count: Arc<AtomicU64>,
}

impl Clone for BithumbWebSocketTrader {
    fn clone(&self) -> Self {
        Self {
            auth: self.auth.clone(),
            ws_sink: self.ws_sink.clone(),
            ws_stream: self.ws_stream.clone(),
            state: self.state.clone(),
            clock: self.clock.clone(),
            last_ping_time: self.last_ping_time.clone(),
            last_pong_time: self.last_pong_time.clone(),
            auth_completed_time: self.auth_completed_time.clone(),
            task_handles: self.task_handles.clone(),
            metrics: self.metrics.clone(),
            messages_received: self.messages_received.clone(),
            messages_sent: self.messages_sent.clone(),
            reconnect_count: self.reconnect_count.clone(),
        }
    }
}

impl BithumbWebSocketTrader {
    /// Create a `SmartString` ID with the given prefix and a UUID suffix
    ///
    /// This helper reduces boilerplate when creating unique identifiers for
    /// orders, tickets, and other entities that need UUID-based IDs.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// let order_id = Self::create_smart_id("bithumb_order");
    /// let ticket_id = Self::create_smart_id("bithumb");
    /// ```
    /// Create a new Bithumb WebSocket trader instance
    #[must_use]
    pub fn new(auth: Arc<BithumbAuth>) -> Self {
        Self {
            auth,
            ws_sink: Arc::new(AsyncRwLock::new(None)),
            ws_stream: Arc::new(AsyncRwLock::new(None)),
            state: Arc::new(AtomicU8::new(BithumbConnectionState::Disconnected as u8)),
            clock: Clock::new(),
            last_ping_time: Arc::new(AtomicU64::new(0)),
            last_pong_time: Arc::new(AtomicU64::new(0)),
            auth_completed_time: Arc::new(AtomicU64::new(0)),
            task_handles: Arc::new(AsyncRwLock::new(TaskHandles::default())),
            metrics: Arc::new(ConnectionMetrics::default()),
            messages_received: Arc::new(AtomicU64::new(0)),
            messages_sent: Arc::new(AtomicU64::new(0)),
            reconnect_count: Arc::new(AtomicU64::new(0)),
        }
    }

    /// Get current connection state
    #[must_use]
    pub fn connection_state(&self) -> BithumbConnectionState {
        BithumbConnectionState::from(self.state.load(Ordering::Acquire))
    }

    /// Set connection state
    fn set_state(&self, new_state: BithumbConnectionState) {
        let old_state = self.connection_state();
        self.state.store(new_state as u8, Ordering::Release);

        if old_state != new_state {
            debug!("Bithumb connection state transition: {old_state:?} -> {new_state:?}");
        }
    }

    /// Check if currently connected and authenticated
    #[must_use]
    pub fn is_connected(&self) -> bool {
        matches!(
            self.connection_state(),
            BithumbConnectionState::Connected | BithumbConnectionState::Authenticated
        )
    }

    /// Check if WebSocket is authenticated
    #[must_use]
    pub fn is_authenticated(&self) -> bool {
        matches!(
            self.connection_state(),
            BithumbConnectionState::Authenticated
        )
    }

    /// Get connection statistics
    #[must_use]
    pub fn get_stats(&self) -> (u64, u64, u64) {
        (
            self.messages_received.load(Ordering::Relaxed),
            self.messages_sent.load(Ordering::Relaxed),
            self.reconnect_count.load(Ordering::Relaxed),
        )
    }

    /// Get detailed connection health information
    #[must_use]
    pub fn get_connection_health(&self) -> BithumbConnectionHealth {
        let now = self.clock.raw();
        let last_ping = self.last_ping_time.load(Ordering::Acquire);
        let last_pong = self.last_pong_time.load(Ordering::Acquire);
        let auth_time = self.auth_completed_time.load(Ordering::Acquire);

        let time_since_last_ping = if last_ping > 0 {
            Some(Duration::from_nanos(now.saturating_sub(last_ping)))
        } else {
            None
        };

        let time_since_last_pong = if last_pong > 0 {
            Some(Duration::from_nanos(now.saturating_sub(last_pong)))
        } else {
            None
        };

        let time_since_auth = if auth_time > 0 {
            Some(Duration::from_nanos(now.saturating_sub(auth_time)))
        } else {
            None
        };

        BithumbConnectionHealth {
            state: self.connection_state(),
            is_connected: self.is_connected(),
            is_authenticated: self.is_authenticated(),
            time_since_last_ping,
            time_since_last_pong,
            time_since_auth,
            messages_sent: self.metrics.messages_sent.load(Ordering::Relaxed),
            messages_received: self.metrics.messages_received.load(Ordering::Relaxed),
            reconnection_attempts: self.metrics.reconnection_attempts.load(Ordering::Relaxed),
        }
    }

    /// Connect to Bithumb private WebSocket endpoint
    pub async fn connect(&self, report_tx: Sender<ExecutionReport>) -> Result<()> {
        info!("Connecting to Bithumb private WebSocket: {BITHUMB_PRIVATE_WS_URL}");

        // Set connecting state
        self.set_state(BithumbConnectionState::Connecting);

        // Generate JWT token for WebSocket headers
        // Use generate_headers to get JWT token
        let headers = self
            .auth
            .generate_headers("GET", "/websocket/v1/private", None)?;
        let auth_header = headers
            .get(&header_keys::authorization())
            .ok_or_else(|| anyhow::anyhow!("No authorization header generated"))?
            .clone();

        // Create WebSocket configuration with JWT in headers
        let config =
            WebSocketConfig::builder(Exchange::Bithumb, BITHUMB_PRIVATE_WS_URL.to_string())
                .connect_timeout(Duration::from_secs(10))
                .timeout(Duration::from_secs(30))
                .ping_interval(Duration::from_secs(PING_INTERVAL_SECONDS))
                .pong_timeout(Duration::from_secs(PONG_TIMEOUT_SECONDS))
                .max_frame_size(65536) // 64KB
                .max_message_size(MAX_MESSAGE_SIZE)
                .header(
                    header_keys::AUTHORIZATION.to_string(),
                    auth_header.to_string(),
                )
                .build();

        let stats = new_shared_stats();
        let connection_state = Arc::new(RwLock::new(WsConnectionState::Disconnected));
        let mut connector =
            WebSocketConnector::new(config, stats.clone(), connection_state.clone());

        let (sink, stream) = connector.connect_with_retry(BITHUMB_PRIVATE_WS_URL).await?;

        // Store connection references
        *self.ws_sink.write().await = Some(sink);
        *self.ws_stream.write().await = Some(stream);

        // Update state
        self.set_state(BithumbConnectionState::Connected);

        // Subscribe to private channels (no authentication message needed)
        self.subscribe_to_private_channels().await?;

        // Mark as authenticated since JWT was in connection headers
        self.set_state(BithumbConnectionState::Authenticated);
        self.auth_completed_time
            .store(self.clock.raw(), Ordering::Release);

        // Start background tasks
        self.start_ping_task().await;
        self.start_message_task(report_tx).await;

        info!("Successfully connected and authenticated to Bithumb private WebSocket");
        Ok(())
    }

    /// Authenticate and subscribe to private channels
    // Removed authenticate_and_subscribe method as authentication is handled via headers
    /// Subscribe to Bithumb private channels (myOrder and myAsset)
    async fn subscribe_to_private_channels(&self) -> Result<()> {
        debug!("Subscribing to Bithumb private channels");

        // Generate unique ticket ID for this session
        let ticket_id = id_generation::generate_report_id_with_uuid("bithumb");

        // Subscribe to myOrder channel for order updates (array format)
        let order_subscription = simd_json::json!([
            {
                "ticket": ticket_id.clone()
            },
            {
                "type": "myOrder",
                "codes": []  // Empty array for all markets
            },
            {
                "format": "DEFAULT"  // Use full field names
            }
        ]);

        // Subscribe to myAsset channel for balance updates (array format)
        let asset_subscription = simd_json::json!([
            {
                "ticket": ticket_id
            },
            {
                "type": "myAsset"
            },
            {
                "format": "DEFAULT"
            }
        ]);

        // Send order subscription
        let order_msg: SmartString = simd_json::to_string(&order_subscription)?.into();
        self.send_message(Message::Text(order_msg)).await?;

        // Send asset subscription
        let asset_msg: SmartString = simd_json::to_string(&asset_subscription)?.into();
        self.send_message(Message::Text(asset_msg)).await?;

        debug!("Bithumb private channels subscription completed");
        Ok(())
    }

    /// Send a message through the WebSocket connection
    async fn send_message(&self, message: Message) -> Result<()> {
        if let Some(sink) = self.ws_sink.write().await.as_mut() {
            let frame = message.to_frame_view();
            sink.send(frame).await?;
            self.metrics.messages_sent.fetch_add(1, Ordering::Relaxed);
            Ok(())
        } else {
            bail!("WebSocket sink not available")
        }
    }

    /// Start ping task to keep connection alive
    async fn start_ping_task(&self) {
        let sink = self.ws_sink.clone();
        let last_ping = self.last_ping_time.clone();
        let clock = self.clock.clone();
        let metrics = self.metrics.clone();

        let task = tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(PING_INTERVAL_SECONDS));

            loop {
                interval.tick().await;

                // Send ping frame
                if let Some(sink) = sink.write().await.as_mut() {
                    let ping_message = Message::Ping(Vec::new());
                    let frame = ping_message.to_frame_view();
                    if let Err(e) = sink.send(frame).await {
                        error!("Failed to send ping: {e}");
                        break;
                    }

                    last_ping.store(clock.raw(), Ordering::Release);
                    metrics.pings_sent.fetch_add(1, Ordering::Relaxed);
                } else {
                    break;
                }
            }
        });

        self.task_handles.write().await.ping_handle = Some(task);
    }

    /// Start message processing task
    async fn start_message_task(&self, report_tx: Sender<ExecutionReport>) {
        let stream = self.ws_stream.clone();
        let last_pong_time = self.last_pong_time.clone();
        let clock = self.clock.clone();
        let metrics = self.metrics.clone();

        let task = tokio::spawn(async move {
            loop {
                // Take a lock scope to get the stream
                let mut stream_guard = stream.write().await;
                if let Some(ws_stream) = stream_guard.as_mut() {
                    if let Some(frame) = ws_stream.next().await {
                        drop(stream_guard); // Release the lock before processing

                        let message = Message::from_frame_view(frame);
                        metrics.messages_received.fetch_add(1, Ordering::Relaxed);

                        match message {
                            Message::Text(text) => {
                                debug!("Received text message: {text}");
                                if let Err(e) =
                                    Self::process_message(&text, &report_tx, &clock).await
                                {
                                    error!("Failed to process message: {e}");
                                }
                            }
                            Message::Pong(_) => {
                                last_pong_time.store(clock.raw(), Ordering::Release);
                                metrics.pongs_received.fetch_add(1, Ordering::Relaxed);
                            }
                            Message::Close(_) => {
                                warn!("Received close message from Bithumb WebSocket");
                                break;
                            }
                            _ => {
                                debug!("Received other message type: {message:?}");
                            }
                        }
                    } else {
                        break; // Stream ended
                    }
                } else {
                    break; // No stream available
                }
            }
        });

        self.task_handles.write().await.message_handler = Some(task);
    }

    /// Process incoming WebSocket messages
    async fn process_message(
        text: &str,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) -> Result<()> {
        // Parse JSON message using simd_json for high performance
        // Use mutable bytes for in-place optimizations while maintaining safety
        let mut json_bytes = text.as_bytes().to_vec();
        let json: simd_json::OwnedValue = simd_json::from_slice(&mut json_bytes)?;

        // Determine message type
        if let Some(msg_type) = json.get("type").and_then(|v| v.as_str()) {
            match msg_type {
                "myOrder" => {
                    Self::handle_order_update(&json, report_tx, clock).await?;
                }
                "myAsset" => {
                    Self::handle_asset_update(&json).await?;
                }
                "auth" => {
                    debug!("Authentication response: {json:?}");
                }
                "subscribe" => {
                    debug!("Subscription response: {json:?}");
                }
                _ => {
                    debug!("Unknown message type: {msg_type}");
                }
            }
        } else {
            debug!("Message without type field: {text}");
        }

        Ok(())
    }

    /// Handle myOrder updates
    async fn handle_order_update(
        json: &simd_json::OwnedValue,
        report_tx: &Sender<ExecutionReport>,
        clock: &Clock,
    ) -> Result<()> {
        debug!("Processing Bithumb order update: {json:?}");

        // Use correct field names from Bithumb documentation
        let order_id = json
            .get("uuid")
            .and_then(|v| v.as_str())
            .unwrap_or("unknown")
            .into();

        let symbol = json
            .get("code")
            .and_then(|v| v.as_str())
            .unwrap_or("UNKNOWN");

        let state_str = json
            .get("state")
            .and_then(|v| v.as_str())
            .unwrap_or("unknown");

        let status = Self::map_bithumb_order_state(state_str);

        let executed_volume = json
            .get("executed_volume")
            .and_then(simd_json::prelude::ValueAsScalar::as_f64)
            .and_then(|f| Decimal::try_from(f).ok())
            .unwrap_or(Decimal::ZERO);

        let remaining_volume = json
            .get("remaining_volume")
            .and_then(simd_json::prelude::ValueAsScalar::as_f64)
            .and_then(|f| Decimal::try_from(f).ok())
            .unwrap_or(Decimal::ZERO);

        let execution_price = json
            .get("price")
            .and_then(simd_json::prelude::ValueAsScalar::as_f64)
            .and_then(|f| Decimal::try_from(f).ok());

        let timestamp = json
            .get("timestamp")
            .and_then(simd_json::prelude::ValueAsScalar::as_u64)
            .unwrap_or(0)
            * 1_000_000; // Convert milliseconds to nanoseconds

        let report = ExecutionReport {
            id: id_generation::generate_exchange_order_id("bithumb"),
            order_id,
            exchange_timestamp: timestamp,
            system_timestamp: rusty_common::time::get_epoch_timestamp_ns(),
            instrument_id: InstrumentId::new(symbol, Venue::Bithumb),
            status,
            filled_quantity: executed_volume,
            remaining_quantity: remaining_volume,
            execution_price,
            reject_reason: None,
            exchange_execution_id: json
                .get("trade_uuid")
                .and_then(|v| v.as_str())
                .map(std::convert::Into::into),
            is_final: matches!(
                status,
                OrderStatus::Filled | OrderStatus::Cancelled | OrderStatus::Rejected
            ),
        };

        if let Err(e) = report_tx.try_send(report) {
            error!("Failed to send Bithumb order update report: {e}");
        }

        Ok(())
    }

    /// Handle myAsset updates
    async fn handle_asset_update(json: &simd_json::OwnedValue) -> Result<()> {
        debug!("Processing Bithumb asset update: {json:?}");

        // Use correct field names from Bithumb documentation
        // assets is directly at the root level, not under "data"
        if let Some(assets) = json.get("assets").and_then(|v| v.as_array()) {
            for (index, asset) in assets.iter().enumerate() {
                match (
                    asset.get("currency").and_then(|v| v.as_str()),
                    asset.get("balance").and_then(|v| v.as_str()), // String in response
                    asset.get("locked").and_then(|v| v.as_str()),  // String in response
                ) {
                    (Some(currency), Some(balance), Some(locked)) => {
                        // Parse string values to f64 for logging
                        let balance_f64 = match balance.parse::<f64>() {
                            Ok(val) => val,
                            Err(e) => {
                                warn!("Failed to parse balance for {currency}: '{balance}' - {e}");
                                0.0
                            }
                        };
                        let locked_f64 = match locked.parse::<f64>() {
                            Ok(val) => val,
                            Err(e) => {
                                warn!(
                                    "Failed to parse locked amount for {currency}: '{locked}' - {e}"
                                );
                                0.0
                            }
                        };

                        info!(
                            "Bithumb asset update - {currency}: balance={balance_f64}, locked={locked_f64}"
                        );
                    }
                    (currency, balance, locked) => {
                        warn!(
                            "Incomplete asset data at index {}: currency={:?}, balance={:?}, locked={:?}",
                            index,
                            currency.is_some(),
                            balance.is_some(),
                            locked.is_some()
                        );
                    }
                }
            }
        } else {
            warn!("No 'assets' array found in myAsset update or it's not an array");
        }

        Ok(())
    }

    /// Map Bithumb order state to internal `OrderStatus`
    fn map_bithumb_order_state(state: &str) -> OrderStatus {
        map_websocket_order_state(state).unwrap_or(OrderStatus::Unknown)
    }

    /// Disconnect from WebSocket
    pub async fn disconnect(&self) -> Result<()> {
        info!("Disconnecting from Bithumb WebSocket");

        self.set_state(BithumbConnectionState::Disconnecting);

        // Cancel background tasks
        {
            let mut handles = self.task_handles.write().await;
            if let Some(handle) = handles.ping_handle.take() {
                handle.abort();
            }
            if let Some(handle) = handles.message_handler.take() {
                handle.abort();
            }
        }

        // Close WebSocket connections
        *self.ws_sink.write().await = None;
        *self.ws_stream.write().await = None;

        self.set_state(BithumbConnectionState::Disconnected);

        info!("Disconnected from Bithumb WebSocket");
        Ok(())
    }
}

#[async_trait]
impl crate::execution_engine::Exchange for BithumbWebSocketTrader {
    fn venue(&self) -> Venue {
        Venue::Bithumb
    }

    async fn place_order(
        &self,
        _order: Order,
        _report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Bithumb WebSocket doesn't support order placement - orders must be placed via REST API
        // The existing BithumbExchange handles REST order placement
        // This WebSocket client focuses on real-time updates only
        bail!("Order placement via WebSocket not supported by Bithumb - use REST API")
    }

    async fn cancel_order(
        &self,
        _order_id: SmartString,
        _report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        // Bithumb WebSocket doesn't support order cancellation - must use REST API
        bail!("Order cancellation via WebSocket not supported by Bithumb - use REST API")
    }

    async fn modify_order(
        &self,
        _order_id: SmartString,
        _new_price: Option<Decimal>,
        _new_quantity: Option<Decimal>,
        _report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        bail!("Order modification not supported by Bithumb")
    }

    async fn cancel_all_orders(
        &self,
        _instrument_id: Option<InstrumentId>,
        _report_sender: Sender<ExecutionReport>,
    ) -> Result<()> {
        bail!("Cancel all orders via WebSocket not supported by Bithumb - use REST API")
    }

    async fn get_order_status(&self, _order_id: &str) -> Result<OrderStatus> {
        bail!("Order status query via WebSocket not supported by Bithumb - use REST API")
    }

    async fn connect(&self, _report_sender: Sender<ExecutionReport>) -> Result<()> {
        bail!("Generic connect not implemented - use connect(report_tx)")
    }

    async fn disconnect(&self) -> Result<()> {
        self.disconnect().await
    }

    async fn is_connected(&self) -> bool {
        self.is_connected()
    }

    async fn get_instruments(&self) -> Result<smallvec::SmallVec<[InstrumentId; 32]>> {
        // Bithumb WebSocket doesn't provide instrument discovery - use REST API
        bail!("Instrument discovery via WebSocket not supported by Bithumb - use REST API")
    }

    async fn send_fix_message(&self, _message: Vec<u8>) -> Result<()> {
        bail!("FIX protocol not supported by Bithumb - use native WebSocket API")
    }

    async fn receive_fix_message(&self) -> Result<Vec<u8>> {
        bail!("FIX protocol not supported by Bithumb - use native WebSocket API")
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::execution_engine::Exchange;
    use rust_decimal::Decimal;
    use rusty_common::auth::exchanges::bithumb::BithumbAuth;
    use rusty_model::enums::OrderSide;

    #[test]
    fn test_bithumb_websocket_trader_creation() {
        let auth = Arc::new(BithumbAuth::new("test_key".into(), "test_secret".into()));
        let trader = BithumbWebSocketTrader::new(auth);

        assert_eq!(
            trader.connection_state(),
            BithumbConnectionState::Disconnected
        );
        assert!(!trader.is_connected());
        assert_eq!(trader.venue(), Venue::Bithumb);
    }

    #[test]
    fn test_connection_state_transitions() {
        let auth = Arc::new(BithumbAuth::new("test_key".into(), "test_secret".into()));
        let trader = BithumbWebSocketTrader::new(auth);

        // Test state transitions
        trader.set_state(BithumbConnectionState::Connecting);
        assert_eq!(
            trader.connection_state(),
            BithumbConnectionState::Connecting
        );

        trader.set_state(BithumbConnectionState::Connected);
        assert_eq!(trader.connection_state(), BithumbConnectionState::Connected);
        assert!(trader.is_connected());

        trader.set_state(BithumbConnectionState::Authenticated);
        assert_eq!(
            trader.connection_state(),
            BithumbConnectionState::Authenticated
        );
        assert!(trader.is_connected());
        assert!(trader.is_authenticated());
    }

    #[test]
    fn test_connection_health() {
        let auth = Arc::new(BithumbAuth::new("test_key".into(), "test_secret".into()));
        let trader = BithumbWebSocketTrader::new(auth);

        let health = trader.get_connection_health();
        assert_eq!(health.state, BithumbConnectionState::Disconnected);
        assert!(!health.is_connected);
        assert!(!health.is_authenticated);
        assert!(health.time_since_last_ping.is_none());
        assert_eq!(health.messages_sent, 0);
        assert_eq!(health.reconnection_attempts, 0);
    }

    #[test]
    fn test_order_status_mapping() {
        // Test WebSocket status mapping
        assert_eq!(
            map_websocket_order_status("placed").unwrap(),
            OrderStatus::New
        );
        assert_eq!(
            map_websocket_order_status("pending").unwrap(),
            OrderStatus::Open
        );
        assert_eq!(
            map_websocket_order_status("partial").unwrap(),
            OrderStatus::PartiallyFilled
        );
        assert_eq!(
            map_websocket_order_status("completed").unwrap(),
            OrderStatus::Filled
        );
        assert_eq!(
            map_websocket_order_status("cancelled").unwrap(),
            OrderStatus::Cancelled
        );
        assert_eq!(
            map_websocket_order_status("rejected").unwrap(),
            OrderStatus::Rejected
        );
        assert!(map_websocket_order_status("unknown").is_err());

        // Test WebSocket state mapping
        assert_eq!(
            map_websocket_order_state("wait").unwrap(),
            OrderStatus::Open
        );
        assert_eq!(
            map_websocket_order_state("trade").unwrap(),
            OrderStatus::PartiallyFilled
        );
        assert_eq!(
            map_websocket_order_state("done").unwrap(),
            OrderStatus::Filled
        );
        assert_eq!(
            map_websocket_order_state("cancel").unwrap(),
            OrderStatus::Cancelled
        );
        assert!(map_websocket_order_state("unknown").is_err());
    }

    #[tokio::test]
    async fn test_exchange_trait_methods_return_errors() {
        let auth = Arc::new(BithumbAuth::new("test_key".into(), "test_secret".into()));
        let trader = BithumbWebSocketTrader::new(auth);
        let (report_tx, _report_rx) = flume::bounded(100);

        // Test that trading operations return errors (WebSocket is read-only)
        let dummy_order = Order::new(
            Venue::Bithumb,
            "BTC_KRW",
            OrderSide::Buy,
            rusty_model::enums::OrderType::Limit,
            Decimal::ONE,
            Some(Decimal::from(50000)),
            rusty_model::types::ClientId::new("test_client"),
        );

        assert!(
            trader
                .place_order(dummy_order, report_tx.clone())
                .await
                .is_err()
        );
        assert!(
            trader
                .cancel_order("test".into(), report_tx.clone())
                .await
                .is_err()
        );
        assert!(
            trader
                .modify_order("test".into(), None, None, report_tx)
                .await
                .is_err()
        );
        assert!(trader.get_order_status("test").await.is_err());
        assert!(trader.get_instruments().await.is_err());
    }

    #[test]
    fn test_subscription_format() {
        // Test that subscription messages use correct array format
        let ticket = "test_ticket";

        // Test myOrder subscription format
        let order_sub = simd_json::json!([
            {"ticket": ticket},
            {"type": "myOrder", "codes": []},
            {"format": "DEFAULT"}
        ]);

        let order_str = simd_json::to_string(&order_sub).unwrap();
        assert!(order_str.contains(r"ticket"));
        assert!(order_str.contains(r"myOrder"));
        assert!(order_str.starts_with('[') && order_str.ends_with(']'));

        // Test myAsset subscription format
        let asset_sub = simd_json::json!([
            {"ticket": ticket},
            {"type": "myAsset"},
            {"format": "DEFAULT"}
        ]);

        let asset_str = simd_json::to_string(&asset_sub).unwrap();
        assert!(asset_str.contains(r"ticket"));
        assert!(asset_str.contains(r"myAsset"));
        assert!(asset_str.starts_with('[') && asset_str.ends_with(']'));
    }

    #[test]
    fn test_bithumb_order_state_mapping() {
        // Test Bithumb-specific order state mapping
        assert_eq!(
            BithumbWebSocketTrader::map_bithumb_order_state("wait"),
            OrderStatus::Open
        );
        assert_eq!(
            BithumbWebSocketTrader::map_bithumb_order_state("trade"),
            OrderStatus::PartiallyFilled
        );
        assert_eq!(
            BithumbWebSocketTrader::map_bithumb_order_state("done"),
            OrderStatus::Filled
        );
        assert_eq!(
            BithumbWebSocketTrader::map_bithumb_order_state("cancel"),
            OrderStatus::Cancelled
        );
        assert_eq!(
            BithumbWebSocketTrader::map_bithumb_order_state("unknown"),
            OrderStatus::Unknown
        );
    }

    #[tokio::test]
    async fn test_websocket_headers_include_jwt() {
        let auth = Arc::new(BithumbAuth::new("test_key".into(), "test_secret".into()));

        // Generate headers to verify JWT is created
        let headers = auth
            .generate_headers("GET", "/websocket/v1/private", None)
            .unwrap();
        let auth_key: SmartString = "Authorization".into();
        let auth_header = headers.get(&auth_key).unwrap();

        // Verify header starts with Bearer
        assert!(auth_header.starts_with("Bearer "));

        // Extract JWT and verify it has three parts (header.payload.signature)
        let jwt = auth_header.strip_prefix("Bearer ").unwrap();
        let parts: Vec<&str> = jwt.split('.').collect();
        assert_eq!(parts.len(), 3, "JWT should have 3 parts separated by dots");
    }
}