diff --git a/README.md b/README.md index 4dc403a..72bc59d 100644 --- a/README.md +++ b/README.md @@ -512,8 +512,16 @@ This project is licensed under the MIT License - see the [LICENSE](LICENSE) file - **Discussions**: [GitHub Discussions](https://github.com/Praying/ccxt-rust/discussions) - **Documentation**: [Project Docs](docs/) +## ❤️ Donations + +If the project is profitable, consider donating as showing gratitude for its development: + +**USDT or USDC Binance Smart Chain BEP20**: + +`0x8e5d858f92938b028065d39450421d0e080d15f7` + --- **Status**: 🚧 Active Development | **Version**: 0.1.2 | **Updated**: 2026-01 -⚠️ **Note**: This library is under active development. APIs may change before v1.0. Not recommended for production use yet. \ No newline at end of file +⚠️ **Note**: This library is under active development. APIs may change before v1.0. Not recommended for production use yet. diff --git a/ccxt-core/Cargo.toml b/ccxt-core/Cargo.toml index fe0f2f2..efecc3d 100644 --- a/ccxt-core/Cargo.toml +++ b/ccxt-core/Cargo.toml @@ -71,6 +71,8 @@ zeroize = { workspace = true } # Test configuration (optional, enabled by test-utils feature) dotenvy = { workspace = true, optional = true } envy = { workspace = true, optional = true } +async-stream = "0.3.6" +derive_more = { version = "2.1.1", features = ["debug"] } [dev-dependencies] tokio-test = { workspace = true } @@ -112,4 +114,4 @@ backtrace = [] [[bench]] name = "timestamp_benchmark" -harness = false \ No newline at end of file +harness = false diff --git a/ccxt-core/src/ws_client/mod.rs b/ccxt-core/src/ws_client/mod.rs index 20f5c31..fd2a068 100644 --- a/ccxt-core/src/ws_client/mod.rs +++ b/ccxt-core/src/ws_client/mod.rs @@ -24,6 +24,7 @@ pub use state::{WsConnectionState, WsStats, WsStatsSnapshot}; pub use subscription::{Subscription, SubscriptionManager}; use crate::error::{Error, Result}; +use derive_more::Debug; use futures_util::{SinkExt, StreamExt, stream::SplitSink}; use serde_json::Value; use std::collections::HashMap; @@ -53,6 +54,7 @@ type WsWriter = SplitSink>, Message>; /// - `DropOldest`: Removes the oldest message to make room (default) /// - `DropNewest`: Discards the incoming message /// - `Block`: Waits until space is available (may stall the read loop) +#[derive(Debug)] pub struct WsClient { config: WsConfig, state: Arc, @@ -64,6 +66,7 @@ pub struct WsClient { shutdown_tx: Arc>>>, stats: Arc, cancel_token: Arc>>, + #[debug(skip)] event_callback: Arc>>, /// Counter for dropped messages due to backpressure dropped_messages: Arc, diff --git a/ccxt-core/src/ws_client/reconnect.rs b/ccxt-core/src/ws_client/reconnect.rs index da80fa1..ec3466e 100644 --- a/ccxt-core/src/ws_client/reconnect.rs +++ b/ccxt-core/src/ws_client/reconnect.rs @@ -259,4 +259,31 @@ impl AutoReconnectCoordinator { info!("Auto-reconnect loop terminated"); } + + /// Convert into a stream (requires `tokio-stream`). + #[cfg(feature = "websocket")] + pub fn into_stream(self) -> impl futures::Stream { + let client = self.client.clone(); + let coordinator = Arc::new(self); + + async_stream::stream! { + // Start reconnection in background + coordinator.start().await; + + loop { + if let Some(msg) = client.receive().await { + yield msg; + } else { + // Check if we should stop + if !coordinator.is_enabled() { + break; + } + // Wait a bit before retry if channel closed (shouldn't happen with reconnect) + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + + coordinator.stop().await; + } + } } diff --git a/ccxt-exchanges/Cargo.toml b/ccxt-exchanges/Cargo.toml index d7abd27..ddc47f5 100644 --- a/ccxt-exchanges/Cargo.toml +++ b/ccxt-exchanges/Cargo.toml @@ -55,6 +55,7 @@ futures-util = { workspace = true } # Security zeroize = { workspace = true } +tokio-stream = "0.1.18" [dev-dependencies] tokio-test = { workspace = true } @@ -98,4 +99,4 @@ full = ["binance", "websocket"] [[bench]] name = "binance_benchmark" -harness = false \ No newline at end of file +harness = false diff --git a/ccxt-exchanges/src/hyperliquid/mod.rs b/ccxt-exchanges/src/hyperliquid/mod.rs index decf360..cc41ef4 100644 --- a/ccxt-exchanges/src/hyperliquid/mod.rs +++ b/ccxt-exchanges/src/hyperliquid/mod.rs @@ -73,6 +73,8 @@ pub struct HyperLiquid { options: HyperLiquidOptions, /// Authentication instance (optional, for private API). auth: Option, + /// Persistent WebSocket connection reference. + pub(crate) ws_connection: std::sync::Arc>>, } /// HyperLiquid-specific options. @@ -146,6 +148,7 @@ impl HyperLiquid { base, options, auth, + ws_connection: std::sync::Arc::new(tokio::sync::RwLock::new(None)), }) } @@ -289,12 +292,11 @@ impl HyperLiquid { .map(auth::HyperLiquidAuth::wallet_address) } - // TODO: Implement in task 11 (WebSocket Implementation) - // /// Creates a public WebSocket client. - // pub fn create_ws(&self) -> ws::HyperLiquidWs { - // let urls = self.urls(); - // ws::HyperLiquidWs::new(urls.ws) - // } + /// Creates a WebSocket client for public data streams. + pub fn create_ws(&self) -> ws::HyperLiquidWs { + let urls = self.urls(); + ws::HyperLiquidWs::new(urls.ws) + } } /// HyperLiquid API URLs. diff --git a/ccxt-exchanges/src/hyperliquid/ws.rs b/ccxt-exchanges/src/hyperliquid/ws.rs index b903417..5577ca2 100644 --- a/ccxt-exchanges/src/hyperliquid/ws.rs +++ b/ccxt-exchanges/src/hyperliquid/ws.rs @@ -8,12 +8,15 @@ use ccxt_core::{ error::Result, ws_client::{WsClient, WsConfig, WsConnectionState}, }; -use serde_json::Value; +use serde_json::{Value, json}; use std::sync::Arc; -use tokio::sync::RwLock; +use std::sync::atomic::{AtomicBool, Ordering}; +use tokio::sync::{Mutex, RwLock}; +use tokio::task::JoinHandle; +use tokio::time::{Duration, interval}; /// Default ping interval for HyperLiquid WebSocket (30 seconds). -const DEFAULT_PING_INTERVAL_MS: u64 = 30000; +const DEFAULT_PING_INTERVAL_MS: u64 = 50000; /// Default reconnect delay (5 seconds). const DEFAULT_RECONNECT_INTERVAL_MS: u64 = 5000; @@ -24,11 +27,14 @@ const MAX_RECONNECT_ATTEMPTS: u32 = 10; /// HyperLiquid WebSocket client. /// /// Provides real-time data streaming for HyperLiquid exchange. +#[derive(Debug, Clone)] pub struct HyperLiquidWs { /// WebSocket client instance. client: Arc, /// Active subscriptions. subscriptions: Arc>>, + ping_active: Arc, + ping_task: Arc>>>, } /// Subscription information. @@ -81,16 +87,22 @@ impl HyperLiquidWs { Self { client: Arc::new(WsClient::new(config)), subscriptions: Arc::new(RwLock::new(Vec::new())), + ping_active: Arc::new(AtomicBool::new(false)), + ping_task: Arc::new(Mutex::new(None)), } } /// Connects to the WebSocket server. pub async fn connect(&self) -> Result<()> { - self.client.connect().await + self.client.connect().await?; + self.start_ping_loop().await; + self.resubscribe_all().await?; + Ok(()) } /// Disconnects from the WebSocket server. pub async fn disconnect(&self) -> Result<()> { + self.stop_ping_loop().await; self.client.disconnect().await } @@ -115,33 +127,8 @@ impl HyperLiquidWs { /// Subscribes to all mid prices. pub async fn subscribe_all_mids(&self) -> Result<()> { - let mut subscription_map = serde_json::Map::new(); - subscription_map.insert( - "type".to_string(), - serde_json::Value::String("allMids".to_string()), - ); - - let mut msg_map = serde_json::Map::new(); - msg_map.insert( - "method".to_string(), - serde_json::Value::String("subscribe".to_string()), - ); - msg_map.insert( - "subscription".to_string(), - serde_json::Value::Object(subscription_map), - ); - - let msg = serde_json::Value::Object(msg_map); - - self.client.send_json(&msg).await?; - - let mut subs = self.subscriptions.write().await; - subs.push(Subscription { - sub_type: SubscriptionType::AllMids, - symbol: None, - }); - - Ok(()) + self.send_subscription(SubscriptionType::AllMids, None) + .await } /// Subscribes to L2 order book updates. @@ -150,37 +137,8 @@ impl HyperLiquidWs { /// /// * `symbol` - Trading pair symbol (e.g., "BTC") pub async fn subscribe_l2_book(&self, symbol: &str) -> Result<()> { - let mut subscription_map = serde_json::Map::new(); - subscription_map.insert( - "type".to_string(), - serde_json::Value::String("l2Book".to_string()), - ); - subscription_map.insert( - "coin".to_string(), - serde_json::Value::String(symbol.to_string()), - ); - - let mut msg_map = serde_json::Map::new(); - msg_map.insert( - "method".to_string(), - serde_json::Value::String("subscribe".to_string()), - ); - msg_map.insert( - "subscription".to_string(), - serde_json::Value::Object(subscription_map), - ); - - let msg = serde_json::Value::Object(msg_map); - - self.client.send_json(&msg).await?; - - let mut subs = self.subscriptions.write().await; - subs.push(Subscription { - sub_type: SubscriptionType::L2Book, - symbol: Some(symbol.to_string()), - }); - - Ok(()) + self.send_subscription(SubscriptionType::L2Book, Some(symbol.to_string())) + .await } /// Subscribes to trade updates. @@ -189,37 +147,8 @@ impl HyperLiquidWs { /// /// * `symbol` - Trading pair symbol (e.g., "BTC") pub async fn subscribe_trades(&self, symbol: &str) -> Result<()> { - let mut subscription_map = serde_json::Map::new(); - subscription_map.insert( - "type".to_string(), - serde_json::Value::String("trades".to_string()), - ); - subscription_map.insert( - "coin".to_string(), - serde_json::Value::String(symbol.to_string()), - ); - - let mut msg_map = serde_json::Map::new(); - msg_map.insert( - "method".to_string(), - serde_json::Value::String("subscribe".to_string()), - ); - msg_map.insert( - "subscription".to_string(), - serde_json::Value::Object(subscription_map), - ); - - let msg = serde_json::Value::Object(msg_map); - - self.client.send_json(&msg).await?; - - let mut subs = self.subscriptions.write().await; - subs.push(Subscription { - sub_type: SubscriptionType::Trades, - symbol: Some(symbol.to_string()), - }); - - Ok(()) + self.send_subscription(SubscriptionType::Trades, Some(symbol.to_string())) + .await } /// Subscribes to candle updates. @@ -229,41 +158,7 @@ impl HyperLiquidWs { /// * `symbol` - Trading pair symbol (e.g., "BTC") /// * `interval` - Candle interval (e.g., "1m", "1h") pub async fn subscribe_candle(&self, symbol: &str, interval: &str) -> Result<()> { - let mut subscription_map = serde_json::Map::new(); - subscription_map.insert( - "type".to_string(), - serde_json::Value::String("candle".to_string()), - ); - subscription_map.insert( - "coin".to_string(), - serde_json::Value::String(symbol.to_string()), - ); - subscription_map.insert( - "interval".to_string(), - serde_json::Value::String(interval.to_string()), - ); - - let mut msg_map = serde_json::Map::new(); - msg_map.insert( - "method".to_string(), - serde_json::Value::String("subscribe".to_string()), - ); - msg_map.insert( - "subscription".to_string(), - serde_json::Value::Object(subscription_map), - ); - - let msg = serde_json::Value::Object(msg_map); - - self.client.send_json(&msg).await?; - - let mut subs = self.subscriptions.write().await; - subs.push(Subscription { - sub_type: SubscriptionType::Candle, - symbol: Some(symbol.to_string()), - }); - - Ok(()) + self.send_candle_subscription(symbol, interval).await } // ======================================================================== @@ -276,37 +171,8 @@ impl HyperLiquidWs { /// /// * `address` - User's Ethereum address pub async fn subscribe_user_events(&self, address: &str) -> Result<()> { - let mut subscription_map = serde_json::Map::new(); - subscription_map.insert( - "type".to_string(), - serde_json::Value::String("userEvents".to_string()), - ); - subscription_map.insert( - "user".to_string(), - serde_json::Value::String(address.to_string()), - ); - - let mut msg_map = serde_json::Map::new(); - msg_map.insert( - "method".to_string(), - serde_json::Value::String("subscribe".to_string()), - ); - msg_map.insert( - "subscription".to_string(), - serde_json::Value::Object(subscription_map), - ); - - let msg = serde_json::Value::Object(msg_map); - - self.client.send_json(&msg).await?; - - let mut subs = self.subscriptions.write().await; - subs.push(Subscription { - sub_type: SubscriptionType::UserEvents, - symbol: None, - }); - - Ok(()) + self.send_subscription(SubscriptionType::UserEvents, Some(address.to_string())) + .await } /// Subscribes to user fills. @@ -315,37 +181,8 @@ impl HyperLiquidWs { /// /// * `address` - User's Ethereum address pub async fn subscribe_user_fills(&self, address: &str) -> Result<()> { - let mut subscription_map = serde_json::Map::new(); - subscription_map.insert( - "type".to_string(), - serde_json::Value::String("userFills".to_string()), - ); - subscription_map.insert( - "user".to_string(), - serde_json::Value::String(address.to_string()), - ); - - let mut msg_map = serde_json::Map::new(); - msg_map.insert( - "method".to_string(), - serde_json::Value::String("subscribe".to_string()), - ); - msg_map.insert( - "subscription".to_string(), - serde_json::Value::Object(subscription_map), - ); - - let msg = serde_json::Value::Object(msg_map); - - self.client.send_json(&msg).await?; - - let mut subs = self.subscriptions.write().await; - subs.push(Subscription { - sub_type: SubscriptionType::UserFills, - symbol: None, - }); - - Ok(()) + self.send_subscription(SubscriptionType::UserFills, Some(address.to_string())) + .await } /// Subscribes to order updates. @@ -354,49 +191,159 @@ impl HyperLiquidWs { /// /// * `address` - User's Ethereum address pub async fn subscribe_order_updates(&self, address: &str) -> Result<()> { - let mut subscription_map = serde_json::Map::new(); - subscription_map.insert( - "type".to_string(), - serde_json::Value::String("orderUpdates".to_string()), - ); - subscription_map.insert( - "user".to_string(), - serde_json::Value::String(address.to_string()), - ); - - let mut msg_map = serde_json::Map::new(); - msg_map.insert( - "method".to_string(), - serde_json::Value::String("subscribe".to_string()), - ); - msg_map.insert( - "subscription".to_string(), - serde_json::Value::Object(subscription_map), - ); - - let msg = serde_json::Value::Object(msg_map); + self.send_subscription(SubscriptionType::OrderUpdates, Some(address.to_string())) + .await + } + + /// Returns a reference to the WebSocket client. + pub fn client(&self) -> &Arc { + &self.client + } + /// Returns a reference to the subscriptions. + pub fn subscriptions(&self) -> &Arc>> { + &self.subscriptions + } + + async fn send_subscription( + &self, + sub_type: SubscriptionType, + symbol: Option, + ) -> Result<()> { + let mut subs = self.subscriptions.write().await; + if subs + .iter() + .any(|sub| sub.sub_type == sub_type && sub.symbol == symbol) + { + return Ok(()); + } + #[allow(clippy::disallowed_methods)] + let msg = Self::build_subscription_message(&sub_type, symbol.as_deref()); self.client.send_json(&msg).await?; + subs.push(Subscription { sub_type, symbol }); + Ok(()) + } + async fn send_candle_subscription(&self, symbol: &str, interval: &str) -> Result<()> { let mut subs = self.subscriptions.write().await; + let key = format!("{}:{}", symbol, interval); + if subs.iter().any(|sub| { + sub.sub_type == SubscriptionType::Candle && sub.symbol.as_deref() == Some(&key) + }) { + return Ok(()); + } + let mut subscription_map = serde_json::Map::new(); + subscription_map.insert("type".to_string(), Value::String("candle".to_string())); + subscription_map.insert("coin".to_string(), Value::String(symbol.to_string())); + subscription_map.insert("interval".to_string(), Value::String(interval.to_string())); + #[allow(clippy::disallowed_methods)] + let msg = json!({"method": "subscribe", "subscription": subscription_map}); + self.client.send_json(&msg).await?; subs.push(Subscription { - sub_type: SubscriptionType::OrderUpdates, - symbol: None, + sub_type: SubscriptionType::Candle, + symbol: Some(key), }); + Ok(()) + } + #[allow(clippy::disallowed_methods)] + fn build_subscription_message(sub_type: &SubscriptionType, symbol: Option<&str>) -> Value { + let mut subscription_map = serde_json::Map::new(); + match sub_type { + SubscriptionType::AllMids => { + subscription_map.insert("type".to_string(), Value::String("allMids".to_string())); + } + SubscriptionType::L2Book => { + subscription_map.insert("type".to_string(), Value::String("l2Book".to_string())); + if let Some(symbol) = symbol { + subscription_map.insert("coin".to_string(), Value::String(symbol.to_string())); + } + } + SubscriptionType::Trades => { + subscription_map.insert("type".to_string(), Value::String("trades".to_string())); + if let Some(symbol) = symbol { + subscription_map.insert("coin".to_string(), Value::String(symbol.to_string())); + } + } + SubscriptionType::UserEvents => { + subscription_map + .insert("type".to_string(), Value::String("userEvents".to_string())); + if let Some(address) = symbol { + subscription_map.insert("user".to_string(), Value::String(address.to_string())); + } + } + SubscriptionType::UserFills => { + subscription_map.insert("type".to_string(), Value::String("userFills".to_string())); + if let Some(address) = symbol { + subscription_map.insert("user".to_string(), Value::String(address.to_string())); + } + } + SubscriptionType::OrderUpdates => { + subscription_map.insert( + "type".to_string(), + Value::String("orderUpdates".to_string()), + ); + if let Some(address) = symbol { + subscription_map.insert("user".to_string(), Value::String(address.to_string())); + } + } + SubscriptionType::Candle => {} + } + json!({"method": "subscribe", "subscription": subscription_map}) + } + + async fn resubscribe_all(&self) -> Result<()> { + let subs = self.subscriptions.read().await.clone(); + for sub in subs { + if sub.sub_type == SubscriptionType::Candle { + if let Some(symbol) = sub.symbol.as_deref() { + if let Some((coin, interval)) = symbol.split_once(':') { + let mut subscription_map = serde_json::Map::new(); + subscription_map + .insert("type".to_string(), Value::String("candle".to_string())); + subscription_map + .insert("coin".to_string(), Value::String(coin.to_string())); + subscription_map + .insert("interval".to_string(), Value::String(interval.to_string())); + #[allow(clippy::disallowed_methods)] + let msg = json!({"method": "subscribe", "subscription": subscription_map}); + self.client.send_json(&msg).await?; + continue; + } + } + } + #[allow(clippy::disallowed_methods)] + let msg = Self::build_subscription_message(&sub.sub_type, sub.symbol.as_deref()); + self.client.send_json(&msg).await?; + } Ok(()) } -} -impl HyperLiquidWs { - /// Returns a reference to the WebSocket client. - pub fn client(&self) -> &Arc { - &self.client + async fn start_ping_loop(&self) { + if self.ping_active.swap(true, Ordering::SeqCst) { + return; + } + let client = Arc::clone(&self.client); + let active = Arc::clone(&self.ping_active); + let mut guard = self.ping_task.lock().await; + *guard = Some(tokio::spawn(async move { + let mut interval = interval(Duration::from_millis(DEFAULT_PING_INTERVAL_MS)); + loop { + interval.tick().await; + if !active.load(Ordering::SeqCst) { + break; + } + #[allow(clippy::disallowed_methods)] + let _ = client.send_json(&json!({"method": "ping"})).await; + } + })); } - /// Returns a reference to the subscriptions. - pub fn subscriptions(&self) -> &Arc>> { - &self.subscriptions + async fn stop_ping_loop(&self) { + self.ping_active.store(false, Ordering::SeqCst); + if let Some(handle) = self.ping_task.lock().await.take() { + handle.abort(); + } } } diff --git a/ccxt-exchanges/src/hyperliquid/ws_exchange_impl.rs b/ccxt-exchanges/src/hyperliquid/ws_exchange_impl.rs index c43f0c1..4488b28 100644 --- a/ccxt-exchanges/src/hyperliquid/ws_exchange_impl.rs +++ b/ccxt-exchanges/src/hyperliquid/ws_exchange_impl.rs @@ -1,53 +1,203 @@ //! WsExchange trait implementation for HyperLiquid //! //! This module implements the unified `WsExchange` trait from `ccxt-core` for HyperLiquid. -//! -//! Note: Full WsExchange implementation requires async_stream and additional setup. -//! This is a placeholder that will be completed when WebSocket streaming is needed. use async_trait::async_trait; use ccxt_core::{ Error, Result, - types::{Balance, Ohlcv, Order, OrderBook, Ticker, Timeframe, Trade}, + types::financial::{Amount, Price}, + types::{Balance, Market, Ohlcv, Order, OrderBook, Ticker, Timeframe, Trade}, ws_client::WsConnectionState, ws_exchange::{MessageStream, WsExchange}, }; +use rust_decimal::Decimal; +use rust_decimal::prelude::FromStr; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio::time::{Duration, timeout}; +use tokio_stream::wrappers::UnboundedReceiverStream; + +use super::{HyperLiquid, parser}; -use super::HyperLiquid; +async fn get_ws(exchange: &HyperLiquid) -> Result { + let mut guard = exchange.ws_connection.write().await; + if let Some(ws) = guard.as_ref() { + return Ok(ws.clone()); + } + let ws = exchange.create_ws(); + ws.connect().await?; + let ws_clone = ws.clone(); + *guard = Some(ws); + Ok(ws_clone) +} + +async fn resolve_market( + exchange: &HyperLiquid, + symbol: &str, +) -> Result<(String, String, Option>)> { + if !symbol.contains('/') && !symbol.contains(':') { + let normalized = format!("{}/USDC:USDC", symbol); + return Ok((symbol.to_string(), normalized, None)); + } + exchange.load_markets(false).await?; + if let Ok(market) = exchange.base().market(symbol).await { + return Ok((market.base.clone(), market.symbol.clone(), Some(market))); + } + let normalized = format!("{}/USDC:USDC", symbol); + if let Ok(market) = exchange.base().market(&normalized).await { + return Ok((market.base.clone(), market.symbol.clone(), Some(market))); + } + Ok((symbol.to_string(), normalized, None)) +} #[async_trait] impl WsExchange for HyperLiquid { // ==================== Connection Management ==================== async fn ws_connect(&self) -> Result<()> { - Err(Error::not_implemented( - "ws_connect - WebSocket not yet initialized", - )) + let _ = get_ws(self).await?; + Ok(()) } async fn ws_disconnect(&self) -> Result<()> { - Err(Error::not_implemented( - "ws_disconnect - WebSocket not yet initialized", - )) + let mut guard = self.ws_connection.write().await; + if let Some(ws) = guard.take() { + ws.disconnect().await?; + } + Ok(()) } fn ws_is_connected(&self) -> bool { + if let Ok(guard) = self.ws_connection.try_read() { + if let Some(ws) = guard.as_ref() { + return ws.is_connected(); + } + } false } fn ws_state(&self) -> WsConnectionState { + if let Ok(guard) = self.ws_connection.try_read() { + if let Some(ws) = guard.as_ref() { + return ws.state(); + } + } WsConnectionState::Disconnected } fn subscriptions(&self) -> Vec { + if let Ok(guard) = self.ws_connection.try_read() { + if let Some(ws) = guard.as_ref() { + if let Ok(subs) = ws.subscriptions().try_read() { + return subs.iter().map(|s| format!("{:?}", s.sub_type)).collect(); + } + } + } Vec::new() } // ==================== Data Streams ==================== - async fn watch_ticker(&self, _symbol: &str) -> Result> { - Err(Error::not_implemented( - "watch_ticker - WebSocket streaming requires additional setup", - )) + async fn watch_ticker(&self, symbol: &str) -> Result> { + let (base, symbol_name, market) = resolve_market(self, symbol).await?; + + let ws = get_ws(self).await?; + ws.subscribe_all_mids().await?; + ws.subscribe_l2_book(&base).await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let ws = ws.clone(); + tokio::spawn(async move { + loop { + if !ws.is_connected() { + let _ = ws.connect().await; + } + + #[allow(clippy::manual_unwrap_or_default)] + let msg = match timeout(Duration::from_secs(2), ws.receive()).await { + Ok(msg) => msg, + Err(_) => None, + }; + + if let Some(msg) = msg { + let channel = msg.get("channel").and_then(|v| v.as_str()); + if channel == Some("allMids") { + let data = msg.get("data").unwrap_or(&msg); + if let Some(obj) = data.as_object() { + if let Some(mid) = obj.get(&base).and_then(|v| v.as_str()) { + if let Ok(price) = Decimal::from_str(mid) { + match parser::parse_ticker( + &symbol_name, + price, + market.as_deref(), + ) { + Ok(ticker) => { + if tx.send(Ok(ticker)).is_err() { + break; + } + } + Err(e) => { + let _ = tx.send(Err(e)); + } + } + } + } + } + continue; + } + + if channel == Some("l2Book") { + let data = msg.get("data").unwrap_or(&msg); + if let Some(coin) = data.get("coin").and_then(|v| v.as_str()) { + if coin != base { + continue; + } + } + if let Ok(book) = parser::parse_orderbook(data, symbol_name.clone()) { + if let (Some(best_bid), Some(best_ask)) = + (book.bids.first(), book.asks.first()) + { + let mid = (best_bid.price.as_decimal() + + best_ask.price.as_decimal()) + / Decimal::from(2); + let timestamp = chrono::Utc::now().timestamp_millis(); + let ticker = Ticker { + symbol: symbol_name.clone(), + timestamp, + datetime: parser::timestamp_to_datetime(timestamp), + high: None, + low: None, + bid: Some(Price::new(best_bid.price.as_decimal())), + bid_volume: Some(Amount::new(best_bid.amount.as_decimal())), + ask: Some(Price::new(best_ask.price.as_decimal())), + ask_volume: Some(Amount::new(best_ask.amount.as_decimal())), + vwap: None, + open: None, + close: Some(Price::new(mid)), + last: Some(Price::new(mid)), + previous_close: None, + change: None, + percentage: None, + average: None, + base_volume: None, + quote_volume: None, + funding_rate: None, + open_interest: None, + index_price: None, + mark_price: None, + info: std::collections::HashMap::new(), + }; + if tx.send(Ok(ticker)).is_err() { + break; + } + } + } + } + } + } + }); + + let stream = UnboundedReceiverStream::new(rx); + Ok(Box::pin(stream)) } async fn watch_tickers(&self, _symbols: &[String]) -> Result>> { @@ -56,18 +206,104 @@ impl WsExchange for HyperLiquid { async fn watch_order_book( &self, - _symbol: &str, + symbol: &str, _limit: Option, ) -> Result> { - Err(Error::not_implemented( - "watch_order_book - WebSocket streaming requires additional setup", - )) + let (base, symbol_name, _market) = resolve_market(self, symbol).await?; + + let ws = get_ws(self).await?; + ws.subscribe_l2_book(&base).await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let ws = ws.clone(); + tokio::spawn(async move { + loop { + if !ws.is_connected() { + let _ = ws.connect().await; + } + + #[allow(clippy::manual_unwrap_or_default)] + let msg = match timeout(Duration::from_secs(2), ws.receive()).await { + Ok(msg) => msg, + Err(_) => None, + }; + + if let Some(msg) = msg { + let channel = msg.get("channel").and_then(|v| v.as_str()); + if channel != Some("l2Book") { + continue; + } + let data = msg.get("data").unwrap_or(&msg); + if let Some(coin) = data.get("coin").and_then(|v| v.as_str()) { + if coin != base { + continue; + } + } + match parser::parse_orderbook(data, symbol_name.clone()) { + Ok(book) => { + if tx.send(Ok(book)).is_err() { + break; + } + } + Err(e) => { + let _ = tx.send(Err(e)); + } + } + } + } + }); + + let stream = UnboundedReceiverStream::new(rx); + Ok(Box::pin(stream)) } - async fn watch_trades(&self, _symbol: &str) -> Result>> { - Err(Error::not_implemented( - "watch_trades - WebSocket streaming requires additional setup", - )) + async fn watch_trades(&self, symbol: &str) -> Result>> { + let (base, _symbol_name, market) = resolve_market(self, symbol).await?; + + let ws = get_ws(self).await?; + ws.subscribe_trades(&base).await?; + + let (tx, rx) = mpsc::unbounded_channel::>>(); + let ws = ws.clone(); + tokio::spawn(async move { + loop { + if !ws.is_connected() { + let _ = ws.connect().await; + } + + #[allow(clippy::manual_unwrap_or_default)] + let msg = match timeout(Duration::from_secs(2), ws.receive()).await { + Ok(msg) => msg, + Err(_) => None, + }; + + if let Some(msg) = msg { + let channel = msg.get("channel").and_then(|v| v.as_str()); + if channel != Some("trades") { + continue; + } + let data = msg.get("data").unwrap_or(&msg); + let mut trades = Vec::new(); + if let Some(items) = data.as_array() { + for item in items { + if let Ok(trade) = parser::parse_trade(item, market.as_deref()) { + trades.push(trade); + } + } + } else if let Ok(trade) = parser::parse_trade(data, market.as_deref()) { + trades.push(trade); + } + if !trades.is_empty() { + if tx.send(Ok(trades)).is_err() { + break; + } + } + } + } + }); + + let stream = UnboundedReceiverStream::new(rx); + Ok(Box::pin(stream)) } async fn watch_ohlcv( @@ -83,11 +319,147 @@ impl WsExchange for HyperLiquid { } async fn watch_orders(&self, _symbol: Option<&str>) -> Result> { - Err(Error::not_implemented("watch_orders")) + let address = self.wallet_address().ok_or_else(|| { + Error::authentication("watch_orders requires authentication (wallet address)") + })?; + + self.load_markets(false).await?; + let ws = get_ws(self).await?; + + ws.subscribe_order_updates(address).await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let ws = ws.clone(); + let symbol_filter = _symbol.map(ToString::to_string); + let market_cache = self.base().market_cache.clone(); + tokio::spawn(async move { + loop { + if !ws.is_connected() { + let _ = ws.connect().await; + } + + #[allow(clippy::manual_unwrap_or_default)] + let msg = match timeout(Duration::from_secs(2), ws.receive()).await { + Ok(msg) => msg, + Err(_) => None, + }; + + if let Some(msg) = msg { + let channel = msg.get("channel").and_then(|v| v.as_str()); + if channel != Some("orderUpdates") { + continue; + } + let data = msg.get("data").unwrap_or(&msg); + let mut items = Vec::new(); + if let Some(array) = data.as_array() { + items.extend(array.iter().cloned()); + } else if let Some(inner) = data.get("data") { + if let Some(array) = inner.as_array() { + items.extend(array.iter().cloned()); + } else { + items.push(inner.clone()); + } + } else { + items.push(data.clone()); + } + + for item in items { + let market = if let Some(coin) = item.get("coin").and_then(|v| v.as_str()) { + let symbol = format!("{}/USDC:USDC", coin); + market_cache.read().await.get_market(&symbol) + } else { + None + }; + + if let Ok(order) = parser::parse_order(&item, market.as_deref()) { + if let Some(ref filter) = symbol_filter { + if order.symbol != *filter { + continue; + } + } + if tx.send(Ok(order)).is_err() { + return; + } + } + } + } + } + }); + + let stream = UnboundedReceiverStream::new(rx); + Ok(Box::pin(stream)) } async fn watch_my_trades(&self, _symbol: Option<&str>) -> Result> { - Err(Error::not_implemented("watch_my_trades")) + let address = self.wallet_address().ok_or_else(|| { + Error::authentication("watch_my_trades requires authentication (wallet address)") + })?; + + self.load_markets(false).await?; + let ws = get_ws(self).await?; + + ws.subscribe_user_fills(address).await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let ws = ws.clone(); + let symbol_filter = _symbol.map(ToString::to_string); + let market_cache = self.base().market_cache.clone(); + tokio::spawn(async move { + loop { + if !ws.is_connected() { + let _ = ws.connect().await; + } + + #[allow(clippy::manual_unwrap_or_default)] + let msg = match timeout(Duration::from_secs(2), ws.receive()).await { + Ok(msg) => msg, + Err(_) => None, + }; + + if let Some(msg) = msg { + let channel = msg.get("channel").and_then(|v| v.as_str()); + if channel != Some("userFills") { + continue; + } + let data = msg.get("data").unwrap_or(&msg); + let mut items = Vec::new(); + if let Some(array) = data.as_array() { + items.extend(array.iter().cloned()); + } else if let Some(inner) = data.get("data") { + if let Some(array) = inner.as_array() { + items.extend(array.iter().cloned()); + } else { + items.push(inner.clone()); + } + } else { + items.push(data.clone()); + } + + for item in items { + let market = if let Some(coin) = item.get("coin").and_then(|v| v.as_str()) { + let symbol = format!("{}/USDC:USDC", coin); + market_cache.read().await.get_market(&symbol) + } else { + None + }; + + if let Ok(trade) = parser::parse_trade(&item, market.as_deref()) { + if let Some(ref filter) = symbol_filter { + if trade.symbol != *filter { + continue; + } + } + if tx.send(Ok(trade)).is_err() { + return; + } + } + } + } + } + }); + + let stream = UnboundedReceiverStream::new(rx); + Ok(Box::pin(stream)) } async fn subscribe(&self, _channel: &str, _symbol: Option<&str>) -> Result<()> { diff --git a/ccxt-exchanges/tests/hyperliquid_ws_test.rs b/ccxt-exchanges/tests/hyperliquid_ws_test.rs new file mode 100644 index 0000000..34dc37d --- /dev/null +++ b/ccxt-exchanges/tests/hyperliquid_ws_test.rs @@ -0,0 +1,109 @@ +use ccxt_core::{ExchangeConfig, ws_exchange::WsExchange}; +use ccxt_exchanges::hyperliquid::{HyperLiquid, HyperLiquidOptions}; +use futures_util::StreamExt; +use std::time::Duration; +use tokio::time::timeout; + +#[tokio::test] +async fn test_hyperliquid_ws_connection_lifecycle() { + // Skip if no internet connection or in CI environment that blocks external connections + if std::env::var("CI").is_ok() { + return; + } + + let config = ExchangeConfig::default(); + // Use testnet for safety + let options = HyperLiquidOptions { + testnet: true, + ..Default::default() + }; + + let exchange = HyperLiquid::new_with_options(config, options, None).unwrap(); + + // 1. Initial State + assert!(!exchange.ws_is_connected()); + + // 2. Connect + let connect_result = timeout(Duration::from_secs(10), exchange.ws_connect()).await; + assert!(connect_result.is_ok(), "Connection timed out"); + assert!(connect_result.unwrap().is_ok(), "Connection failed"); + + assert!(exchange.ws_is_connected()); + + // 3. Disconnect + let disconnect_result = exchange.ws_disconnect().await; + assert!(disconnect_result.is_ok()); + + assert!(!exchange.ws_is_connected()); +} + +#[tokio::test] +async fn test_hyperliquid_ws_public_streams() { + if std::env::var("CI").is_ok() { + return; + } + + let config = ExchangeConfig::default(); + let options = HyperLiquidOptions { + testnet: true, + ..Default::default() + }; + + let exchange = HyperLiquid::new_with_options(config, options, None).unwrap(); + exchange.ws_connect().await.unwrap(); + + // Give connection a moment to stabilize + tokio::time::sleep(Duration::from_millis(100)).await; + + // Test OrderBook Subscription + let mut orderbook_stream = exchange.watch_order_book("ETH", None).await.unwrap(); + let orderbook_item = timeout(Duration::from_secs(10), orderbook_stream.next()) + .await + .ok() + .flatten(); + assert!(matches!(orderbook_item, Some(Ok(_)))); + + // Test Trades Subscription + let mut trades_stream = exchange.watch_trades("ETH").await.unwrap(); + let trades_item = timeout(Duration::from_secs(10), trades_stream.next()) + .await + .ok() + .flatten(); + assert!(matches!(trades_item, Some(Ok(trades)) if !trades.is_empty())); + + // Test Ticker Subscription + let mut ticker_stream = exchange.watch_ticker("ETH").await.unwrap(); + let ticker_item = timeout(Duration::from_secs(10), ticker_stream.next()) + .await + .ok() + .flatten(); + assert!(matches!(ticker_item, Some(Ok(_)))); + + exchange.ws_disconnect().await.unwrap(); +} + +#[tokio::test] +async fn test_hyperliquid_ws_private_streams() { + if std::env::var("CI").is_ok() { + return; + } + + // Requires API key/private key to derive address, or explicit address + // Since we can't easily mock auth here without credentials, we'll test the logic + // but expect authentication error if no credentials provided, which is correct behavior. + + let config = ExchangeConfig::default(); + let options = HyperLiquidOptions { + testnet: true, + ..Default::default() + }; + + let exchange = HyperLiquid::new_with_options(config, options, None).unwrap(); + + // Should fail without auth + let result = exchange.watch_orders(None).await; + assert!(result.is_err()); + + let result = exchange.watch_my_trades(None).await; + assert!(result.is_err()); +}