From 33d4e5ef11e7c4bd03bc425a804899ee0a85369e Mon Sep 17 00:00:00 2001 From: quran Date: Sat, 7 Feb 2026 14:54:47 +0800 Subject: [PATCH] feat(exchange): add futures modules and position/funding parsers for Bitget and OKX - Add futures modules for both Bitget and OKX exchanges - Implement position parsing functions for both exchanges with comprehensive field mapping - Add funding rate and funding rate history parsers for both exchanges - Create helper function for parsing f64 fields from JSON values - Add private WebSocket methods for authentication and order/balance tracking - Implement message detection helpers for account and order updates - Add comprehensive unit tests for position and funding rate parsing - Include rate limiter updates for Binance with new weight field - Add margin implementation modules for both exchanges - Implement private WebSocket creation methods for both exchanges - Add symbol module for Bitget exchange --- ccxt-exchanges/src/binance/rate_limiter.rs | 3 + ccxt-exchanges/src/bitget/margin_impl.rs | 190 ++++ ccxt-exchanges/src/bitget/mod.rs | 12 + ccxt-exchanges/src/bitget/parser.rs | 163 ++++ .../src/bitget/rest/futures/funding.rs | 99 ++ .../src/bitget/rest/futures/leverage.rs | 88 ++ .../src/bitget/rest/futures/margin.rs | 38 + ccxt-exchanges/src/bitget/rest/futures/mod.rs | 9 + .../src/bitget/rest/futures/positions.rs | 121 +++ ccxt-exchanges/src/bitget/rest/mod.rs | 1 + ccxt-exchanges/src/bitget/symbol.rs | 274 ++++++ ccxt-exchanges/src/bitget/ws.rs | 845 +++++++++++++++++- ccxt-exchanges/src/bitget/ws_exchange_impl.rs | 25 +- ccxt-exchanges/src/okx/margin_impl.rs | 192 ++++ ccxt-exchanges/src/okx/mod.rs | 11 + ccxt-exchanges/src/okx/parser.rs | 367 ++++++++ .../src/okx/rest/futures/funding.rs | 97 ++ .../src/okx/rest/futures/leverage.rs | 105 +++ ccxt-exchanges/src/okx/rest/futures/margin.rs | 59 ++ ccxt-exchanges/src/okx/rest/futures/mod.rs | 9 + .../src/okx/rest/futures/positions.rs | 213 +++++ ccxt-exchanges/src/okx/rest/mod.rs | 1 + ccxt-exchanges/src/okx/ws.rs | 576 +++++++++++- ccxt-exchanges/src/okx/ws_exchange_impl.rs | 25 +- 24 files changed, 3499 insertions(+), 24 deletions(-) create mode 100644 ccxt-exchanges/src/bitget/margin_impl.rs create mode 100644 ccxt-exchanges/src/bitget/rest/futures/funding.rs create mode 100644 ccxt-exchanges/src/bitget/rest/futures/leverage.rs create mode 100644 ccxt-exchanges/src/bitget/rest/futures/margin.rs create mode 100644 ccxt-exchanges/src/bitget/rest/futures/mod.rs create mode 100644 ccxt-exchanges/src/bitget/rest/futures/positions.rs create mode 100644 ccxt-exchanges/src/bitget/symbol.rs create mode 100644 ccxt-exchanges/src/okx/margin_impl.rs create mode 100644 ccxt-exchanges/src/okx/rest/futures/funding.rs create mode 100644 ccxt-exchanges/src/okx/rest/futures/leverage.rs create mode 100644 ccxt-exchanges/src/okx/rest/futures/margin.rs create mode 100644 ccxt-exchanges/src/okx/rest/futures/mod.rs create mode 100644 ccxt-exchanges/src/okx/rest/futures/positions.rs diff --git a/ccxt-exchanges/src/binance/rate_limiter.rs b/ccxt-exchanges/src/binance/rate_limiter.rs index af43fce..20c859e 100644 --- a/ccxt-exchanges/src/binance/rate_limiter.rs +++ b/ccxt-exchanges/src/binance/rate_limiter.rs @@ -27,6 +27,7 @@ //! // Update from response headers //! let info = RateLimitInfo { //! used_weight_1m: Some(500), +//! used_weight_1s: None, //! order_count_10s: Some(5), //! order_count_1d: Some(1000), //! retry_after: None, @@ -485,6 +486,7 @@ mod tests { let info = RateLimitInfo { used_weight_1m: Some(500), + used_weight_1s: None, order_count_10s: Some(5), order_count_1d: Some(1000), retry_after: None, @@ -555,6 +557,7 @@ mod tests { let info = RateLimitInfo { used_weight_1m: Some(1000), + used_weight_1s: None, order_count_10s: Some(50), order_count_1d: Some(5000), retry_after: Some(10), diff --git a/ccxt-exchanges/src/bitget/margin_impl.rs b/ccxt-exchanges/src/bitget/margin_impl.rs new file mode 100644 index 0000000..ede434f --- /dev/null +++ b/ccxt-exchanges/src/bitget/margin_impl.rs @@ -0,0 +1,190 @@ +//! Margin trait implementation for Bitget. +//! +//! Implements the `Margin` trait from `ccxt-core` for Bitget, providing +//! position management, leverage configuration, margin mode, and funding rate operations. + +use async_trait::async_trait; +use ccxt_core::{ + Result, + capability::ExchangeCapabilities, + traits::{Margin, PublicExchange}, + types::{ + FundingRate, FundingRateHistory, Position, Timeframe, + params::{LeverageParams, MarginMode}, + }, +}; + +use super::Bitget; + +// ============================================================================ +// PublicExchange Implementation +// ============================================================================ + +impl PublicExchange for Bitget { + fn id(&self) -> &'static str { + "bitget" + } + + fn name(&self) -> &'static str { + "Bitget" + } + + fn version(&self) -> &'static str { + "v2" + } + + fn certified(&self) -> bool { + false + } + + fn capabilities(&self) -> ExchangeCapabilities { + use ccxt_core::exchange::Capability; + + ExchangeCapabilities::builder() + // Market Data + .market_data() + .without_capability(Capability::FetchCurrencies) + .without_capability(Capability::FetchStatus) + .without_capability(Capability::FetchTime) + // Trading + .trading() + .without_capability(Capability::CancelAllOrders) + .without_capability(Capability::EditOrder) + .without_capability(Capability::FetchOrders) + .without_capability(Capability::FetchCanceledOrders) + // Account + .capability(Capability::FetchBalance) + .capability(Capability::FetchMyTrades) + // Margin / Futures + .capability(Capability::FetchPositions) + .capability(Capability::SetLeverage) + .capability(Capability::SetMarginMode) + .capability(Capability::FetchFundingRate) + .capability(Capability::FetchFundingRates) + // WebSocket + .capability(Capability::Websocket) + .capability(Capability::WatchTicker) + .capability(Capability::WatchOrderBook) + .capability(Capability::WatchTrades) + .capability(Capability::WatchBalance) + .capability(Capability::WatchOrders) + .capability(Capability::WatchMyTrades) + .build() + } + + fn timeframes(&self) -> Vec { + vec![ + Timeframe::M1, + Timeframe::M5, + Timeframe::M15, + Timeframe::M30, + Timeframe::H1, + Timeframe::H4, + Timeframe::H6, + Timeframe::H12, + Timeframe::D1, + Timeframe::D3, + Timeframe::W1, + Timeframe::Mon1, + ] + } + + fn rate_limit(&self) -> u32 { + 20 + } + + fn has_websocket(&self) -> bool { + true + } +} + +// ============================================================================ +// Margin Trait Implementation +// ============================================================================ + +#[async_trait] +impl Margin for Bitget { + async fn fetch_positions_for(&self, symbols: &[&str]) -> Result> { + self.fetch_positions_impl(symbols).await + } + + async fn fetch_position(&self, symbol: &str) -> Result { + self.fetch_position_impl(symbol).await + } + + async fn set_leverage_with_params(&self, params: LeverageParams) -> Result<()> { + let margin_mode = params.margin_mode.map(|m| match m { + MarginMode::Isolated => "isolated", + MarginMode::Cross => "cross", + }); + self.set_leverage_impl(¶ms.symbol, params.leverage, margin_mode) + .await + } + + async fn get_leverage(&self, symbol: &str) -> Result { + self.get_leverage_impl(symbol).await + } + + async fn set_margin_mode(&self, symbol: &str, mode: MarginMode) -> Result<()> { + let mode_str = match mode { + MarginMode::Isolated => "isolated", + MarginMode::Cross => "cross", + }; + self.set_margin_mode_impl(symbol, mode_str).await + } + + async fn fetch_funding_rate(&self, symbol: &str) -> Result { + self.fetch_funding_rate_impl(symbol).await + } + + async fn fetch_funding_rates(&self, symbols: &[&str]) -> Result> { + self.fetch_funding_rates_impl(symbols).await + } + + async fn fetch_funding_rate_history( + &self, + symbol: &str, + since: Option, + limit: Option, + ) -> Result> { + self.fetch_funding_rate_history_impl(symbol, since, limit) + .await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use ccxt_core::ExchangeConfig; + + #[test] + fn test_bitget_public_exchange_impl() { + let bitget = Bitget::new(ExchangeConfig::default()).unwrap(); + let pe: &dyn PublicExchange = &bitget; + + assert_eq!(pe.id(), "bitget"); + assert_eq!(pe.name(), "Bitget"); + assert_eq!(pe.version(), "v2"); + assert!(!pe.certified()); + assert!(pe.has_websocket()); + assert_eq!(pe.rate_limit(), 20); + } + + #[test] + fn test_bitget_capabilities_include_margin() { + let bitget = Bitget::new(ExchangeConfig::default()).unwrap(); + let caps = PublicExchange::capabilities(&bitget); + + assert!(caps.has("fetchPositions")); + assert!(caps.has("setLeverage")); + assert!(caps.has("setMarginMode")); + assert!(caps.has("fetchFundingRate")); + assert!(caps.has("fetchFundingRates")); + } + + #[test] + fn test_bitget_margin_trait_object_safety() { + let bitget = Bitget::new(ExchangeConfig::default()).unwrap(); + let _margin: Box = Box::new(bitget); + } +} diff --git a/ccxt-exchanges/src/bitget/mod.rs b/ccxt-exchanges/src/bitget/mod.rs index 2967c6c..9e1c3c3 100644 --- a/ccxt-exchanges/src/bitget/mod.rs +++ b/ccxt-exchanges/src/bitget/mod.rs @@ -12,9 +12,11 @@ pub mod builder; pub mod endpoint_router; pub mod error; mod exchange_impl; +mod margin_impl; pub mod parser; pub mod rest; pub mod signed_request; +pub mod symbol; pub mod ws; mod ws_exchange_impl; @@ -317,6 +319,16 @@ impl Bitget { ws::BitgetWs::new(urls.ws_public) } + /// Creates a new private WebSocket client for Bitget. + /// + /// Returns a `BitgetWs` instance configured with the private WebSocket URL. + /// The caller must call `login()` on the returned client before subscribing + /// to private channels. + pub fn create_private_ws(&self) -> ws::BitgetWs { + let urls = self.urls(); + ws::BitgetWs::new(urls.ws_private) + } + /// Creates a signed request builder for authenticated API requests. /// /// # Arguments diff --git a/ccxt-exchanges/src/bitget/parser.rs b/ccxt-exchanges/src/bitget/parser.rs index 59ce78e..cbea0c1 100644 --- a/ccxt-exchanges/src/bitget/parser.rs +++ b/ccxt-exchanges/src/bitget/parser.rs @@ -642,6 +642,169 @@ fn parse_balance_entry(data: &Value, balances: &mut HashMap Result { + use ccxt_core::types::position::PositionSide; + + let hold_side = data["holdSide"].as_str().unwrap_or("long"); + let position_side = match hold_side.to_lowercase().as_str() { + "short" => PositionSide::Short, + "long" => PositionSide::Long, + _ => PositionSide::Both, + }; + + let side = match position_side { + PositionSide::Long => Some("long".to_string()), + PositionSide::Short => Some("short".to_string()), + PositionSide::Both => None, + }; + + let total = parse_f64_field(data, "total") + .or_else(|| parse_f64_field(data, "openDelegateSize")) + .unwrap_or(0.0); + let avg_px = + parse_f64_field(data, "averageOpenPrice").or_else(|| parse_f64_field(data, "openPriceAvg")); + let mark_px = parse_f64_field(data, "markPrice"); + let upl = + parse_f64_field(data, "unrealizedPL").or_else(|| parse_f64_field(data, "unrealizedPl")); + let lever = parse_f64_field(data, "leverage"); + let liq_px = parse_f64_field(data, "liquidationPrice"); + let margin = parse_f64_field(data, "margin"); + let realized_pnl = parse_f64_field(data, "achievedProfits"); + + let mgn_mode = data["marginMode"].as_str().unwrap_or("crossed"); + let margin_mode = Some(match mgn_mode { + "isolated" => "isolated".to_string(), + _ => "cross".to_string(), + }); + + let timestamp = parse_timestamp(data, "uTime").or_else(|| parse_timestamp(data, "cTime")); + let datetime = timestamp.and_then(timestamp_to_datetime); + + let initial_margin_percentage = lever.map(|l| if l > 0.0 { 1.0 / l } else { 0.0 }); + + let notional = match (avg_px, Some(total)) { + (Some(price), Some(qty)) if qty > 0.0 => Some(price * qty), + _ => None, + }; + + let percentage = match (upl, margin) { + (Some(pnl), Some(m)) if m > 0.0 => Some((pnl / m) * 100.0), + _ => None, + }; + + Ok(ccxt_core::types::Position { + info: data.clone(), + id: data["posId"] + .as_str() + .or_else(|| data["trackingNo"].as_str()) + .map(ToString::to_string), + symbol: symbol.to_string(), + side, + position_side: Some(position_side), + dual_side_position: None, + contracts: Some(total), + contract_size: parse_f64_field(data, "contractSize"), + entry_price: avg_px, + mark_price: mark_px, + notional, + leverage: lever, + collateral: margin, + initial_margin: margin, + initial_margin_percentage, + maintenance_margin: None, + maintenance_margin_percentage: None, + unrealized_pnl: upl, + realized_pnl, + liquidation_price: liq_px, + margin_ratio: None, + margin_mode, + hedged: None, + percentage, + timestamp, + datetime, + }) +} + +/// Parse funding rate data from Bitget response. +pub fn parse_funding_rate(data: &Value, symbol: &str) -> Result { + let funding_rate = parse_f64_field(data, "fundingRate"); + let funding_time = + parse_timestamp(data, "fundingTime").or_else(|| parse_timestamp(data, "nextFundingTime")); + + let timestamp = parse_timestamp(data, "ts").or_else(|| parse_timestamp(data, "fundingTime")); + let datetime = timestamp.and_then(timestamp_to_datetime); + let funding_datetime = funding_time.and_then(timestamp_to_datetime); + + Ok(ccxt_core::types::FundingRate { + info: data.clone(), + symbol: symbol.to_string(), + mark_price: parse_f64_field(data, "markPrice"), + index_price: parse_f64_field(data, "indexPrice"), + interest_rate: None, + estimated_settle_price: None, + funding_rate, + funding_timestamp: funding_time, + funding_datetime, + previous_funding_rate: None, + previous_funding_timestamp: None, + previous_funding_datetime: None, + timestamp, + datetime, + }) +} + +/// Parse funding rate history data from Bitget response. +pub fn parse_funding_rate_history( + data: &Value, + symbol: &str, +) -> Result { + let funding_rate = parse_f64_field(data, "fundingRate"); + let timestamp = + parse_timestamp(data, "fundingTime").or_else(|| parse_timestamp(data, "settleTime")); + let datetime = timestamp.and_then(timestamp_to_datetime); + + Ok(ccxt_core::types::FundingRateHistory { + info: data.clone(), + symbol: symbol.to_string(), + funding_rate, + timestamp, + datetime, + }) +} + +/// Helper to parse a string field as f64. +fn parse_f64_field(data: &Value, field: &str) -> Option { + data[field] + .as_str() + .and_then(|s| { + if s.is_empty() { + None + } else { + s.parse::().ok() + } + }) + .or_else(|| data[field].as_f64()) +} + // ============================================================================ // Unit Tests // ============================================================================ diff --git a/ccxt-exchanges/src/bitget/rest/futures/funding.rs b/ccxt-exchanges/src/bitget/rest/futures/funding.rs new file mode 100644 index 0000000..7b5d2bc --- /dev/null +++ b/ccxt-exchanges/src/bitget/rest/futures/funding.rs @@ -0,0 +1,99 @@ +//! Funding rate operations for Bitget futures/swap. + +use super::super::super::{Bitget, parser, symbol::BitgetSymbolConverter}; +use ccxt_core::{ + Error, ParseError, Result, + types::{FundingRate, FundingRateHistory}, +}; +use std::collections::HashMap; +use tracing::warn; + +impl Bitget { + /// Fetch current funding rate for a symbol. + /// + /// Uses Bitget GET `/api/v2/mix/market/current-fund-rate` (public endpoint). + pub async fn fetch_funding_rate_impl(&self, symbol: &str) -> Result { + let exchange_id = BitgetSymbolConverter::unified_to_exchange(symbol); + let product_type = BitgetSymbolConverter::product_type_from_symbol(symbol); + + let mut params = HashMap::new(); + params.insert("symbol".to_string(), exchange_id); + params.insert("productType".to_string(), product_type.to_string()); + + let data = self + .public_request("GET", "/api/v2/mix/market/current-fund-rate", Some(¶ms)) + .await?; + + let rates_array = data["data"].as_array().ok_or_else(|| { + Error::from(ParseError::invalid_format("data", "Expected data array")) + })?; + + if rates_array.is_empty() { + return Err(Error::from(ParseError::missing_field_owned(format!( + "No funding rate found for symbol: {}", + symbol + )))); + } + + parser::parse_funding_rate(&rates_array[0], symbol) + } + + /// Fetch funding rates for multiple symbols. + pub async fn fetch_funding_rates_impl(&self, symbols: &[&str]) -> Result> { + let mut rates = Vec::new(); + for symbol in symbols { + match self.fetch_funding_rate_impl(symbol).await { + Ok(rate) => rates.push(rate), + Err(e) => { + warn!(error = %e, symbol = %symbol, "Failed to fetch Bitget funding rate"); + } + } + } + Ok(rates) + } + + /// Fetch funding rate history for a symbol. + /// + /// Uses Bitget GET `/api/v2/mix/market/history-fund-rate` (public endpoint). + pub async fn fetch_funding_rate_history_impl( + &self, + symbol: &str, + since: Option, + limit: Option, + ) -> Result> { + let exchange_id = BitgetSymbolConverter::unified_to_exchange(symbol); + let product_type = BitgetSymbolConverter::product_type_from_symbol(symbol); + + let mut params = HashMap::new(); + params.insert("symbol".to_string(), exchange_id); + params.insert("productType".to_string(), product_type.to_string()); + + if let Some(l) = limit { + params.insert("pageSize".to_string(), l.to_string()); + } + + if let Some(s) = since { + params.insert("startTime".to_string(), s.to_string()); + } + + let data = self + .public_request("GET", "/api/v2/mix/market/history-fund-rate", Some(¶ms)) + .await?; + + let history_array = data["data"].as_array().ok_or_else(|| { + Error::from(ParseError::invalid_format("data", "Expected data array")) + })?; + + let mut history = Vec::new(); + for item in history_array { + match parser::parse_funding_rate_history(item, symbol) { + Ok(record) => history.push(record), + Err(e) => { + warn!(error = %e, "Failed to parse Bitget funding rate history"); + } + } + } + + Ok(history) + } +} diff --git a/ccxt-exchanges/src/bitget/rest/futures/leverage.rs b/ccxt-exchanges/src/bitget/rest/futures/leverage.rs new file mode 100644 index 0000000..92735bc --- /dev/null +++ b/ccxt-exchanges/src/bitget/rest/futures/leverage.rs @@ -0,0 +1,88 @@ +//! Leverage operations for Bitget futures/swap. + +use super::super::super::signed_request::HttpMethod; +use super::super::super::{Bitget, symbol::BitgetSymbolConverter}; +use ccxt_core::{Error, ParseError, Result}; + +impl Bitget { + /// Set leverage for a symbol. + /// + /// Uses Bitget POST `/api/v2/mix/account/set-leverage` endpoint. + pub async fn set_leverage_impl( + &self, + symbol: &str, + leverage: u32, + margin_mode: Option<&str>, + ) -> Result<()> { + let exchange_id = BitgetSymbolConverter::unified_to_exchange(symbol); + let product_type = BitgetSymbolConverter::product_type_from_symbol(symbol); + let margin_coin = Self::margin_coin_from_symbol(symbol); + + let hold_side = if margin_mode == Some("isolated") { + // For isolated mode, set for both sides + "long" + } else { + // For cross mode, no holdSide needed + "" + }; + + let mut builder = self + .signed_request("/api/v2/mix/account/set-leverage") + .method(HttpMethod::Post) + .param("symbol", &exchange_id) + .param("productType", product_type) + .param("marginCoin", margin_coin) + .param("leverage", leverage.to_string()); + + if !hold_side.is_empty() { + builder = builder.param("holdSide", hold_side); + } + + let _data = builder.execute().await?; + + // If isolated mode, also set for short side + if margin_mode == Some("isolated") { + self.signed_request("/api/v2/mix/account/set-leverage") + .method(HttpMethod::Post) + .param("symbol", &exchange_id) + .param("productType", product_type) + .param("marginCoin", margin_coin) + .param("leverage", leverage.to_string()) + .param("holdSide", "short") + .execute() + .await?; + } + + Ok(()) + } + + /// Get current leverage for a symbol. + /// + /// Uses Bitget GET `/api/v2/mix/account/account` endpoint and extracts leverage. + pub async fn get_leverage_impl(&self, symbol: &str) -> Result { + let exchange_id = BitgetSymbolConverter::unified_to_exchange(symbol); + let product_type = BitgetSymbolConverter::product_type_from_symbol(symbol); + let margin_coin = Self::margin_coin_from_symbol(symbol); + + let data = self + .signed_request("/api/v2/mix/account/account") + .param("symbol", &exchange_id) + .param("productType", product_type) + .param("marginCoin", margin_coin) + .execute() + .await?; + + // Try to extract leverage from the account data + let lever_str = data["data"]["crossMarginLeverage"] + .as_str() + .or_else(|| data["data"]["fixedLongLeverage"].as_str()) + .ok_or_else(|| Error::from(ParseError::missing_field("leverage")))?; + + lever_str.parse::().map(|v| v as u32).map_err(|_| { + Error::from(ParseError::invalid_value( + "leverage", + format!("Cannot parse leverage: {}", lever_str), + )) + }) + } +} diff --git a/ccxt-exchanges/src/bitget/rest/futures/margin.rs b/ccxt-exchanges/src/bitget/rest/futures/margin.rs new file mode 100644 index 0000000..749fce7 --- /dev/null +++ b/ccxt-exchanges/src/bitget/rest/futures/margin.rs @@ -0,0 +1,38 @@ +//! Margin mode operations for Bitget futures/swap. + +use super::super::super::signed_request::HttpMethod; +use super::super::super::{Bitget, symbol::BitgetSymbolConverter}; +use ccxt_core::{Error, Result}; + +impl Bitget { + /// Set margin mode (cross or isolated) for a symbol. + /// + /// Uses Bitget POST `/api/v2/mix/account/set-margin-mode` endpoint. + pub async fn set_margin_mode_impl(&self, symbol: &str, mode: &str) -> Result<()> { + let exchange_id = BitgetSymbolConverter::unified_to_exchange(symbol); + let product_type = BitgetSymbolConverter::product_type_from_symbol(symbol); + let margin_coin = Self::margin_coin_from_symbol(symbol); + + let margin_mode = match mode.to_lowercase().as_str() { + "isolated" => "isolated", + "cross" | "crossed" => "crossed", + _ => { + return Err(Error::invalid_request(format!( + "Invalid margin mode: {}. Must be 'isolated' or 'cross'", + mode + ))); + } + }; + + self.signed_request("/api/v2/mix/account/set-margin-mode") + .method(HttpMethod::Post) + .param("symbol", &exchange_id) + .param("productType", product_type) + .param("marginCoin", margin_coin) + .param("marginMode", margin_mode) + .execute() + .await?; + + Ok(()) + } +} diff --git a/ccxt-exchanges/src/bitget/rest/futures/mod.rs b/ccxt-exchanges/src/bitget/rest/futures/mod.rs new file mode 100644 index 0000000..368c1fe --- /dev/null +++ b/ccxt-exchanges/src/bitget/rest/futures/mod.rs @@ -0,0 +1,9 @@ +//! Bitget futures/swap operations. +//! +//! This module contains methods for managing positions, leverage, +//! funding rates, and margin mode for Bitget perpetual swaps and futures contracts. + +mod funding; +mod leverage; +mod margin; +mod positions; diff --git a/ccxt-exchanges/src/bitget/rest/futures/positions.rs b/ccxt-exchanges/src/bitget/rest/futures/positions.rs new file mode 100644 index 0000000..504d7ae --- /dev/null +++ b/ccxt-exchanges/src/bitget/rest/futures/positions.rs @@ -0,0 +1,121 @@ +//! Position management methods for Bitget futures/swap. + +use super::super::super::{Bitget, parser, symbol::BitgetSymbolConverter}; +use ccxt_core::{Error, ParseError, Result, types::Position}; +use tracing::warn; + +impl Bitget { + /// Fetch a single position for a symbol. + /// + /// Uses Bitget GET `/api/v2/mix/position/single-position` endpoint. + pub async fn fetch_position_impl(&self, symbol: &str) -> Result { + let exchange_id = BitgetSymbolConverter::unified_to_exchange(symbol); + let product_type = BitgetSymbolConverter::product_type_from_symbol(symbol); + + let data = self + .signed_request("/api/v2/mix/position/single-position") + .param("symbol", &exchange_id) + .param("productType", product_type) + .param("marginCoin", Self::margin_coin_from_symbol(symbol)) + .execute() + .await?; + + let positions_array = data["data"].as_array().ok_or_else(|| { + Error::from(ParseError::invalid_format("data", "Expected data array")) + })?; + + if positions_array.is_empty() { + return Err(Error::from(ParseError::missing_field_owned(format!( + "No position found for symbol: {}", + symbol + )))); + } + + parser::parse_position(&positions_array[0], symbol) + } + + /// Fetch positions for specific symbols, or all positions if empty. + /// + /// Uses Bitget GET `/api/v2/mix/position/all-position` endpoint. + pub async fn fetch_positions_impl(&self, symbols: &[&str]) -> Result> { + let product_type = if symbols.is_empty() { + self.options().effective_product_type() + } else { + BitgetSymbolConverter::product_type_from_symbol(symbols[0]) + }; + + let margin_coin = if symbols.is_empty() { + "USDT".to_string() + } else { + Self::margin_coin_from_symbol(symbols[0]).to_string() + }; + + let data = self + .signed_request("/api/v2/mix/position/all-position") + .param("productType", product_type) + .param("marginCoin", &margin_coin) + .execute() + .await?; + + let positions_array = data["data"].as_array().ok_or_else(|| { + Error::from(ParseError::invalid_format("data", "Expected data array")) + })?; + + let mut positions = Vec::new(); + for position_data in positions_array { + let bitget_symbol = position_data["symbol"].as_str().unwrap_or_default(); + let symbol_hint = + BitgetSymbolConverter::exchange_to_unified_hint(bitget_symbol, product_type); + + match parser::parse_position(position_data, &symbol_hint) { + Ok(position) => { + if symbols.is_empty() || symbols.contains(&position.symbol.as_str()) { + positions.push(position); + } + } + Err(e) => { + warn!(error = %e, symbol = %bitget_symbol, "Failed to parse Bitget position"); + } + } + } + + Ok(positions) + } + + /// Extract margin coin from a unified symbol. + /// + /// - "BTC/USDT:USDT" → "USDT" + /// - "BTC/USD:BTC" → "BTC" + /// - "BTC/USDT" → "USDT" (fallback to quote) + pub(crate) fn margin_coin_from_symbol(symbol: &str) -> &str { + if let Some(pos) = symbol.find(':') { + let settle_part = &symbol[pos + 1..]; + // Strip expiry date if present + if let Some(dash_pos) = settle_part.find('-') { + &settle_part[..dash_pos] + } else { + settle_part + } + } else if let Some(pos) = symbol.find('/') { + &symbol[pos + 1..] + } else { + "USDT" + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_margin_coin_from_symbol() { + assert_eq!(Bitget::margin_coin_from_symbol("BTC/USDT:USDT"), "USDT"); + assert_eq!(Bitget::margin_coin_from_symbol("BTC/USD:BTC"), "BTC"); + assert_eq!(Bitget::margin_coin_from_symbol("BTC/USDT"), "USDT"); + assert_eq!( + Bitget::margin_coin_from_symbol("BTC/USDT:USDT-241231"), + "USDT" + ); + } +} diff --git a/ccxt-exchanges/src/bitget/rest/mod.rs b/ccxt-exchanges/src/bitget/rest/mod.rs index 2ae706d..ccc43b3 100644 --- a/ccxt-exchanges/src/bitget/rest/mod.rs +++ b/ccxt-exchanges/src/bitget/rest/mod.rs @@ -3,6 +3,7 @@ //! Implements all REST API endpoint operations for the Bitget exchange. mod account; +pub mod futures; mod market_data; mod trading; diff --git a/ccxt-exchanges/src/bitget/symbol.rs b/ccxt-exchanges/src/bitget/symbol.rs new file mode 100644 index 0000000..8c8215e --- /dev/null +++ b/ccxt-exchanges/src/bitget/symbol.rs @@ -0,0 +1,274 @@ +//! Bitget symbol converter implementation. +//! +//! This module provides conversion between unified CCXT symbols and Bitget-specific +//! exchange IDs for spot, swap (perpetual), and futures markets. +//! +//! # Bitget Symbol Formats +//! +//! | Market Type | Unified Format | Bitget Format | Product Type | +//! |-------------|----------------|---------------|--------------| +//! | Spot | BTC/USDT | BTCUSDT | spot | +//! | Linear Swap | BTC/USDT:USDT | BTCUSDT | USDT-FUTURES | +//! | Inverse Swap | BTC/USD:BTC | BTCUSD | COIN-FUTURES | +//! +//! # Example +//! +//! ```rust +//! use ccxt_exchanges::bitget::symbol::BitgetSymbolConverter; +//! +//! assert_eq!(BitgetSymbolConverter::unified_to_exchange("BTC/USDT"), "BTCUSDT"); +//! assert_eq!(BitgetSymbolConverter::unified_to_exchange("BTC/USDT:USDT"), "BTCUSDT"); +//! assert_eq!(BitgetSymbolConverter::product_type_from_symbol("BTC/USDT:USDT"), "USDT-FUTURES"); +//! assert_eq!(BitgetSymbolConverter::product_type_from_symbol("BTC/USDT"), "spot"); +//! ``` + +/// Bitget symbol converter. +/// +/// Provides methods to convert between unified CCXT symbols and Bitget-specific +/// exchange IDs, and to determine the appropriate product type. +pub struct BitgetSymbolConverter; + +impl BitgetSymbolConverter { + /// Convert a unified CCXT symbol to Bitget exchange ID. + /// + /// Strips the settlement currency suffix and removes separators: + /// - "BTC/USDT" → "BTCUSDT" + /// - "BTC/USDT:USDT" → "BTCUSDT" + /// - "BTC/USD:BTC" → "BTCUSD" + /// - "BTC/USDT:USDT-241231" → "BTCUSDT" + pub fn unified_to_exchange(symbol: &str) -> String { + // Strip settlement part (everything after ':') + let base_quote = if let Some(pos) = symbol.find(':') { + &symbol[..pos] + } else { + symbol + }; + + // Remove the '/' separator + base_quote.replace('/', "") + } + + /// Determine the Bitget product type from a unified symbol. + /// + /// Returns: + /// - "USDT-FUTURES" for linear perpetual/futures (e.g., "BTC/USDT:USDT") + /// - "COIN-FUTURES" for inverse perpetual/futures (e.g., "BTC/USD:BTC") + /// - "spot" for spot markets (e.g., "BTC/USDT") + pub fn product_type_from_symbol(symbol: &str) -> &'static str { + if let Some(pos) = symbol.find(':') { + let settle_part = &symbol[pos + 1..]; + // Strip expiry date if present + let settle = if let Some(dash_pos) = settle_part.find('-') { + &settle_part[..dash_pos] + } else { + settle_part + }; + + // Extract quote currency + let base_quote = &symbol[..pos]; + let quote = if let Some(slash_pos) = base_quote.find('/') { + &base_quote[slash_pos + 1..] + } else { + "" + }; + + // If settle == quote, it's linear (USDT-FUTURES) + // If settle != quote (settle == base), it's inverse (COIN-FUTURES) + if settle == quote { + "USDT-FUTURES" + } else { + "COIN-FUTURES" + } + } else { + "spot" + } + } + + /// Check if a symbol is a contract (has settlement currency). + pub fn is_contract(symbol: &str) -> bool { + symbol.contains(':') + } + + /// Check if a symbol is a spot market. + pub fn is_spot(symbol: &str) -> bool { + !symbol.contains(':') + } + + /// Convert a Bitget exchange ID back to a rough unified symbol hint. + /// + /// This is a best-effort conversion: + /// - "BTCUSDT" with product_type "USDT-FUTURES" → "BTC/USDT:USDT" + /// - "BTCUSD" with product_type "COIN-FUTURES" → "BTC/USD:BTC" + /// - "BTCUSDT" with product_type "spot" → "BTC/USDT" + pub fn exchange_to_unified_hint(exchange_id: &str, product_type: &str) -> String { + // Try to split the exchange ID into base and quote + // Common quote currencies in order of length (longest first to avoid partial matches) + let quote_currencies = ["USDT", "USDC", "USD", "BTC", "ETH"]; + + for quote in "e_currencies { + if exchange_id.ends_with(quote) && exchange_id.len() > quote.len() { + let base = &exchange_id[..exchange_id.len() - quote.len()]; + return match product_type { + "USDT-FUTURES" | "usdt-futures" => { + format!("{}/{}:{}", base, quote, quote) + } + "COIN-FUTURES" | "coin-futures" => { + format!("{}/{}:{}", base, quote, base) + } + _ => { + format!("{}/{}", base, quote) + } + }; + } + } + + // Fallback: return as-is + exchange_id.to_string() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // ======================================================================== + // unified_to_exchange tests + // ======================================================================== + + #[test] + fn test_spot_to_exchange() { + assert_eq!( + BitgetSymbolConverter::unified_to_exchange("BTC/USDT"), + "BTCUSDT" + ); + assert_eq!( + BitgetSymbolConverter::unified_to_exchange("ETH/USDT"), + "ETHUSDT" + ); + assert_eq!( + BitgetSymbolConverter::unified_to_exchange("SOL/BTC"), + "SOLBTC" + ); + } + + #[test] + fn test_linear_swap_to_exchange() { + assert_eq!( + BitgetSymbolConverter::unified_to_exchange("BTC/USDT:USDT"), + "BTCUSDT" + ); + assert_eq!( + BitgetSymbolConverter::unified_to_exchange("ETH/USDT:USDT"), + "ETHUSDT" + ); + } + + #[test] + fn test_inverse_swap_to_exchange() { + assert_eq!( + BitgetSymbolConverter::unified_to_exchange("BTC/USD:BTC"), + "BTCUSD" + ); + } + + #[test] + fn test_futures_with_expiry_to_exchange() { + assert_eq!( + BitgetSymbolConverter::unified_to_exchange("BTC/USDT:USDT-241231"), + "BTCUSDT" + ); + } + + // ======================================================================== + // product_type_from_symbol tests + // ======================================================================== + + #[test] + fn test_product_type_spot() { + assert_eq!( + BitgetSymbolConverter::product_type_from_symbol("BTC/USDT"), + "spot" + ); + assert_eq!( + BitgetSymbolConverter::product_type_from_symbol("ETH/BTC"), + "spot" + ); + } + + #[test] + fn test_product_type_linear() { + assert_eq!( + BitgetSymbolConverter::product_type_from_symbol("BTC/USDT:USDT"), + "USDT-FUTURES" + ); + assert_eq!( + BitgetSymbolConverter::product_type_from_symbol("ETH/USDT:USDT"), + "USDT-FUTURES" + ); + } + + #[test] + fn test_product_type_inverse() { + assert_eq!( + BitgetSymbolConverter::product_type_from_symbol("BTC/USD:BTC"), + "COIN-FUTURES" + ); + assert_eq!( + BitgetSymbolConverter::product_type_from_symbol("ETH/USD:ETH"), + "COIN-FUTURES" + ); + } + + #[test] + fn test_product_type_futures_with_expiry() { + assert_eq!( + BitgetSymbolConverter::product_type_from_symbol("BTC/USDT:USDT-241231"), + "USDT-FUTURES" + ); + } + + // ======================================================================== + // Helper function tests + // ======================================================================== + + #[test] + fn test_is_contract() { + assert!(BitgetSymbolConverter::is_contract("BTC/USDT:USDT")); + assert!(BitgetSymbolConverter::is_contract("BTC/USD:BTC")); + assert!(!BitgetSymbolConverter::is_contract("BTC/USDT")); + } + + #[test] + fn test_is_spot() { + assert!(BitgetSymbolConverter::is_spot("BTC/USDT")); + assert!(!BitgetSymbolConverter::is_spot("BTC/USDT:USDT")); + } + + // ======================================================================== + // exchange_to_unified_hint tests + // ======================================================================== + + #[test] + fn test_exchange_to_unified_spot() { + assert_eq!( + BitgetSymbolConverter::exchange_to_unified_hint("BTCUSDT", "spot"), + "BTC/USDT" + ); + } + + #[test] + fn test_exchange_to_unified_linear() { + assert_eq!( + BitgetSymbolConverter::exchange_to_unified_hint("BTCUSDT", "USDT-FUTURES"), + "BTC/USDT:USDT" + ); + } + + #[test] + fn test_exchange_to_unified_inverse() { + assert_eq!( + BitgetSymbolConverter::exchange_to_unified_hint("BTCUSD", "COIN-FUTURES"), + "BTC/USD:BTC" + ); + } +} diff --git a/ccxt-exchanges/src/bitget/ws.rs b/ccxt-exchanges/src/bitget/ws.rs index 48d9da6..81a0ea0 100644 --- a/ccxt-exchanges/src/bitget/ws.rs +++ b/ccxt-exchanges/src/bitget/ws.rs @@ -4,17 +4,26 @@ //! Supports public streams (ticker, orderbook, trades) and private streams //! (balance, orders) with automatic reconnection. +use crate::bitget::auth::BitgetAuth; use crate::bitget::parser::{parse_orderbook, parse_ticker, parse_trade}; use ccxt_core::error::{Error, Result}; -use ccxt_core::types::{Market, OrderBook, Ticker, Trade}; +use ccxt_core::types::{ + Balance, BalanceEntry, Fee, Market, Order, OrderBook, OrderSide, OrderStatus, OrderType, + Ticker, Trade, + financial::{Amount, Cost, Price}, +}; use ccxt_core::ws_client::{WsClient, WsConfig, WsConnectionState}; use ccxt_core::ws_exchange::MessageStream; use futures::Stream; +use rust_decimal::Decimal; +use rust_decimal::prelude::FromStr; use serde_json::Value; +use std::collections::HashMap; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; use tokio::sync::{RwLock, mpsc}; +use tracing::{debug, warn}; /// Default ping interval for Bitget WebSocket (30 seconds). const DEFAULT_PING_INTERVAL_MS: u64 = 30000; @@ -444,6 +453,255 @@ impl BitgetWs { self.subscriptions.read().await.clone() } + // ==================== Private WebSocket Methods ==================== + + /// Authenticates with the Bitget private WebSocket. + /// + /// Bitget WS login uses the same HMAC-SHA256 signing as REST API: + /// sign(timestamp + "GET" + "/user/verify", secret) + pub async fn login(&self, auth: &BitgetAuth) -> Result<()> { + let timestamp = (chrono::Utc::now().timestamp_millis() / 1000).to_string(); + let sign = auth.sign(×tamp, "GET", "/user/verify", ""); + + #[allow(clippy::disallowed_methods)] + let msg = serde_json::json!({ + "op": "login", + "args": [{ + "apiKey": auth.api_key(), + "passphrase": auth.passphrase(), + "timestamp": timestamp, + "sign": sign + }] + }); + + self.client.send_json(&msg).await?; + + // Wait for login response + let timeout = tokio::time::Duration::from_secs(10); + let start = tokio::time::Instant::now(); + + while start.elapsed() < timeout { + if let Some(resp) = self.client.receive().await { + if let Some(event) = resp.get("event").and_then(|e| e.as_str()) { + if event == "login" { + let code = resp.get("code").and_then(|c| c.as_str()).unwrap_or("1"); + if code == "0" { + debug!("Bitget WebSocket login successful"); + return Ok(()); + } + let msg_text = resp + .get("msg") + .and_then(|m| m.as_str()) + .unwrap_or("Unknown error"); + return Err(Error::authentication(format!( + "Bitget WebSocket login failed: {} (code: {})", + msg_text, code + ))); + } + } + } + } + + Err(Error::authentication( + "Bitget WebSocket login timed out waiting for response", + )) + } + + /// Subscribes to the orders channel (private). + /// + /// Receives real-time order updates for the specified instrument type. + pub async fn subscribe_orders(&self, inst_type: Option<&str>) -> Result<()> { + let inst_type = inst_type.unwrap_or("SPOT"); + + #[allow(clippy::disallowed_methods)] + let msg = serde_json::json!({ + "op": "subscribe", + "args": [{ + "instType": inst_type, + "channel": "orders", + "instId": "default" + }] + }); + + self.client.send_json(&msg).await?; + self.subscriptions.write().await.push("orders".to_string()); + Ok(()) + } + + /// Subscribes to the account channel (private). + /// + /// Receives real-time balance/account updates. + pub async fn subscribe_account(&self) -> Result<()> { + #[allow(clippy::disallowed_methods)] + let msg = serde_json::json!({ + "op": "subscribe", + "args": [{ + "instType": "SPOT", + "channel": "account", + "coin": "default" + }] + }); + + self.client.send_json(&msg).await?; + self.subscriptions.write().await.push("account".to_string()); + Ok(()) + } + + /// Watches balance updates via the private WebSocket. + /// + /// Requires prior authentication via `login()`. + pub async fn watch_balance(&self) -> Result> { + if !self.is_connected() { + self.connect().await?; + } + + self.subscribe_account().await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let client = Arc::clone(&self.client); + + tokio::spawn(async move { + while let Some(msg) = client.receive().await { + if is_account_message(&msg) { + match parse_ws_balance(&msg) { + Ok(balance) => { + if tx.send(Ok(balance)).is_err() { + break; + } + } + Err(e) => { + warn!(error = %e, "Failed to parse Bitget balance update"); + if tx.send(Err(e)).is_err() { + break; + } + } + } + } + } + }); + + Ok(Box::pin(ReceiverStream::new(rx))) + } + + /// Watches order updates via the private WebSocket. + /// + /// Requires prior authentication via `login()`. + pub async fn watch_orders( + &self, + symbol_filter: Option, + ) -> Result> { + if !self.is_connected() { + self.connect().await?; + } + + self.subscribe_orders(None).await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let client = Arc::clone(&self.client); + + tokio::spawn(async move { + while let Some(msg) = client.receive().await { + if is_orders_message(&msg) { + if let Some(data_array) = msg.get("data").and_then(|d| d.as_array()) { + for data in data_array { + // Apply symbol filter if specified + if let Some(ref filter) = symbol_filter { + let inst_id = + data.get("instId").and_then(|i| i.as_str()).unwrap_or(""); + let unified = format_unified_symbol(inst_id); + if unified != *filter && inst_id != filter.as_str() { + continue; + } + } + + match parse_ws_order(data) { + Ok(order) => { + if tx.send(Ok(order)).is_err() { + return; + } + } + Err(e) => { + warn!(error = %e, "Failed to parse Bitget order update"); + if tx.send(Err(e)).is_err() { + return; + } + } + } + } + } + } + } + }); + + Ok(Box::pin(ReceiverStream::new(rx))) + } + + /// Watches trade fills via the private WebSocket (orders channel). + /// + /// Extracts trade information from order execution reports. + /// Requires prior authentication via `login()`. + pub async fn watch_my_trades( + &self, + symbol_filter: Option, + ) -> Result> { + if !self.is_connected() { + self.connect().await?; + } + + self.subscribe_orders(None).await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let client = Arc::clone(&self.client); + + tokio::spawn(async move { + while let Some(msg) = client.receive().await { + if is_orders_message(&msg) { + if let Some(data_array) = msg.get("data").and_then(|d| d.as_array()) { + for data in data_array { + // Only emit trades for filled/partially filled orders + let fill_sz = data + .get("fillSz") + .or_else(|| data.get("baseVolume")) + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0.0); + + if fill_sz <= 0.0 { + continue; + } + + // Apply symbol filter + if let Some(ref filter) = symbol_filter { + let inst_id = + data.get("instId").and_then(|i| i.as_str()).unwrap_or(""); + let unified = format_unified_symbol(inst_id); + if unified != *filter && inst_id != filter.as_str() { + continue; + } + } + + match parse_ws_my_trade(data) { + Ok(trade) => { + if tx.send(Ok(trade)).is_err() { + return; + } + } + Err(e) => { + warn!(error = %e, "Failed to parse Bitget trade update"); + if tx.send(Err(e)).is_err() { + return; + } + } + } + } + } + } + } + }); + + Ok(Box::pin(ReceiverStream::new(rx))) + } + /// Watches ticker updates for a symbol. /// /// Returns a stream of `Ticker` updates for the specified symbol. @@ -738,6 +996,28 @@ fn is_trade_message(msg: &Value, symbol: &str) -> bool { } } +// ============================================================================ +// Private Message Type Detection Helpers +// ============================================================================ + +/// Check if a WebSocket message is an account/balance update. +fn is_account_message(msg: &Value) -> bool { + if let Some(arg) = msg.get("arg") { + arg.get("channel").and_then(|c| c.as_str()) == Some("account") + } else { + false + } +} + +/// Check if a WebSocket message is an orders update. +fn is_orders_message(msg: &Value) -> bool { + if let Some(arg) = msg.get("arg") { + arg.get("channel").and_then(|c| c.as_str()) == Some("orders") + } else { + false + } +} + /// Format a Bitget symbol (e.g., "BTCUSDT") to unified format (e.g., "BTC/USDT"). fn format_unified_symbol(symbol: &str) -> String { // Common quote currencies to detect @@ -815,6 +1095,286 @@ pub fn parse_ws_trades(msg: &Value, market: Option<&Market>) -> Result Result { + let data_array = msg + .get("data") + .and_then(|d| d.as_array()) + .ok_or_else(|| Error::invalid_request("Missing data in account message"))?; + + let mut balances = HashMap::new(); + + for data in data_array { + // Bitget V2 sends individual coin updates + let currency = data + .get("coin") + .and_then(|c| c.as_str()) + .unwrap_or_default() + .to_string(); + + if currency.is_empty() { + continue; + } + + let free = data + .get("available") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let used = data + .get("frozen") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let total = free + used; + + balances.insert(currency, BalanceEntry { free, used, total }); + } + + Ok(Balance { + balances, + info: HashMap::new(), + }) +} + +/// Parse a WebSocket order update message into an Order. +/// +/// Bitget order update format: +/// ```json +/// { +/// "instId": "BTCUSDT", +/// "ordId": "123456", +/// "clOrdId": "client123", +/// "side": "buy", +/// "ordType": "limit", +/// "px": "50000", +/// "sz": "0.1", +/// "fillPx": "49999", +/// "fillSz": "0.05", +/// "avgPx": "49999", +/// "status": "partially_filled", +/// "fee": "-0.5", +/// "feeCcy": "USDT", +/// "uTime": "1700000000000", +/// "cTime": "1700000000000" +/// } +/// ``` +pub fn parse_ws_order(data: &Value) -> Result { + let inst_id = data + .get("instId") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + let symbol = format_unified_symbol(inst_id); + + let id = data + .get("ordId") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + + let client_order_id = data + .get("clOrdId") + .and_then(|v| v.as_str()) + .map(ToString::to_string); + + let side = match data.get("side").and_then(|v| v.as_str()) { + Some("sell") => OrderSide::Sell, + _ => OrderSide::Buy, + }; + + let order_type = match data.get("ordType").and_then(|v| v.as_str()) { + Some("market") => OrderType::Market, + _ => OrderType::Limit, + }; + + let price = data + .get("px") + .or_else(|| data.get("price")) + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()); + + let amount = data + .get("sz") + .or_else(|| data.get("size")) + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let filled = data + .get("accFillSz") + .or_else(|| data.get("baseVolume")) + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()); + + let average = data + .get("avgPx") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .filter(|d| !d.is_zero()); + + let cost = match (filled, average) { + (Some(f), Some(a)) => Some(f * a), + _ => None, + }; + + let status = match data.get("status").and_then(|v| v.as_str()) { + Some("filled" | "full-fill") => OrderStatus::Closed, + Some("cancelled" | "canceled") => OrderStatus::Cancelled, + _ => OrderStatus::Open, + }; + + let fee = { + let fee_amount = data + .get("fee") + .or_else(|| data.get("feeDetail")) + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()); + let fee_currency = data + .get("feeCcy") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + fee_amount.map(|f| Fee::new(fee_currency, f.abs())) + }; + + let timestamp = data + .get("cTime") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()); + + Ok(Order { + id, + client_order_id, + symbol, + order_type, + side, + price, + amount, + filled, + remaining: None, + cost, + average, + status, + fee, + fees: None, + timestamp, + datetime: None, + last_trade_timestamp: None, + time_in_force: None, + post_only: None, + reduce_only: None, + stop_price: None, + trigger_price: None, + take_profit_price: None, + stop_loss_price: None, + trailing_delta: None, + trailing_percent: None, + activation_price: None, + callback_rate: None, + working_type: None, + trades: None, + info: HashMap::new(), + }) +} + +/// Parse a WebSocket order fill into a Trade (my trade). +/// +/// Extracts the latest fill information from an order update. +pub fn parse_ws_my_trade(data: &Value) -> Result { + let inst_id = data + .get("instId") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + let symbol = format_unified_symbol(inst_id); + + let trade_id = data + .get("tradeId") + .and_then(|v| v.as_str()) + .map(ToString::to_string); + + let order_id = data + .get("ordId") + .and_then(|v| v.as_str()) + .map(ToString::to_string); + + let side = match data.get("side").and_then(|v| v.as_str()) { + Some("buy") => OrderSide::Buy, + _ => OrderSide::Sell, + }; + + let fill_px = data + .get("fillPx") + .or_else(|| data.get("price")) + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let fill_sz = data + .get("fillSz") + .or_else(|| data.get("baseVolume")) + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let timestamp = data + .get("fillTime") + .or_else(|| data.get("uTime")) + .or_else(|| data.get("cTime")) + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0); + + let fee = { + let fee_amount = data + .get("fillFee") + .or_else(|| data.get("fee")) + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()); + let fee_currency = data + .get("fillFeeCcy") + .or_else(|| data.get("feeCcy")) + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + fee_amount.map(|f| Fee::new(fee_currency, f.abs())) + }; + + Ok(Trade { + id: trade_id, + order: order_id, + symbol, + trade_type: None, + side, + taker_or_maker: None, + price: Price(fill_px), + amount: Amount(fill_sz), + cost: Some(Cost(fill_px * fill_sz)), + fee, + timestamp, + datetime: None, + info: HashMap::new(), + }) +} + #[cfg(test)] mod tests { use super::*; @@ -1356,4 +1916,287 @@ mod tests { // Unknown quote currency returns as-is assert_eq!(format_unified_symbol("BTCXYZ"), "BTCXYZ"); } + + // ==================== Private Message Type Detection Tests ==================== + + #[test] + fn test_is_account_message_true() { + let msg: Value = serde_json::from_str( + r#"{ + "arg": {"instType": "SPOT", "channel": "account", "coin": "default"}, + "data": [{}] + }"#, + ) + .unwrap(); + assert!(is_account_message(&msg)); + } + + #[test] + fn test_is_account_message_wrong_channel() { + let msg: Value = serde_json::from_str( + r#"{ + "arg": {"instType": "SPOT", "channel": "orders"}, + "data": [{}] + }"#, + ) + .unwrap(); + assert!(!is_account_message(&msg)); + } + + #[test] + fn test_is_orders_message_true() { + let msg: Value = serde_json::from_str( + r#"{ + "arg": {"instType": "SPOT", "channel": "orders", "instId": "default"}, + "data": [{}] + }"#, + ) + .unwrap(); + assert!(is_orders_message(&msg)); + } + + #[test] + fn test_is_orders_message_wrong_channel() { + let msg: Value = serde_json::from_str( + r#"{ + "arg": {"instType": "SPOT", "channel": "account"}, + "data": [{}] + }"#, + ) + .unwrap(); + assert!(!is_orders_message(&msg)); + } + + // ==================== Private Balance Parsing Tests ==================== + + #[test] + fn test_parse_ws_balance_single_coin() { + let msg: Value = serde_json::from_str( + r#"{ + "arg": {"instType": "SPOT", "channel": "account", "coin": "default"}, + "data": [{ + "coin": "USDT", + "available": "50000.50", + "frozen": "10000.25", + "uTime": "1700000000000" + }] + }"#, + ) + .unwrap(); + + let balance = parse_ws_balance(&msg).unwrap(); + assert_eq!(balance.balances.len(), 1); + + let usdt = balance.balances.get("USDT").unwrap(); + assert_eq!(usdt.free, dec!(50000.50)); + assert_eq!(usdt.used, dec!(10000.25)); + assert_eq!(usdt.total, dec!(60000.75)); + } + + #[test] + fn test_parse_ws_balance_multiple_coins() { + let msg: Value = serde_json::from_str( + r#"{ + "arg": {"instType": "SPOT", "channel": "account"}, + "data": [ + {"coin": "USDT", "available": "50000", "frozen": "10000"}, + {"coin": "BTC", "available": "1.5", "frozen": "0.5"} + ] + }"#, + ) + .unwrap(); + + let balance = parse_ws_balance(&msg).unwrap(); + assert_eq!(balance.balances.len(), 2); + + let usdt = balance.balances.get("USDT").unwrap(); + assert_eq!(usdt.free, dec!(50000)); + assert_eq!(usdt.total, dec!(60000)); + + let btc = balance.balances.get("BTC").unwrap(); + assert_eq!(btc.free, dec!(1.5)); + assert_eq!(btc.total, dec!(2.0)); + } + + #[test] + fn test_parse_ws_balance_missing_data() { + let msg: Value = serde_json::from_str( + r#"{ + "arg": {"instType": "SPOT", "channel": "account"} + }"#, + ) + .unwrap(); + + let result = parse_ws_balance(&msg); + assert!(result.is_err()); + } + + // ==================== Private Order Parsing Tests ==================== + + #[test] + fn test_parse_ws_order_limit_buy() { + let data: Value = serde_json::from_str( + r#"{ + "instId": "BTCUSDT", + "ordId": "123456789", + "clOrdId": "client123", + "side": "buy", + "ordType": "limit", + "px": "50000", + "sz": "0.1", + "accFillSz": "0.05", + "avgPx": "49999", + "status": "partially_filled", + "fee": "-0.5", + "feeCcy": "USDT", + "cTime": "1700000000000" + }"#, + ) + .unwrap(); + + let order = parse_ws_order(&data).unwrap(); + assert_eq!(order.symbol, "BTC/USDT"); + assert_eq!(order.id, "123456789"); + assert_eq!(order.client_order_id, Some("client123".to_string())); + assert_eq!(order.side, OrderSide::Buy); + assert_eq!(order.order_type, OrderType::Limit); + assert_eq!(order.price, Some(dec!(50000))); + assert_eq!(order.amount, dec!(0.1)); + assert_eq!(order.filled, Some(dec!(0.05))); + assert_eq!(order.average, Some(dec!(49999))); + assert_eq!(order.status, OrderStatus::Open); // partially_filled is still open + assert_eq!(order.timestamp, Some(1700000000000)); + assert!(order.fee.is_some()); + let fee = order.fee.unwrap(); + assert_eq!(fee.cost, dec!(0.5)); // abs value + } + + #[test] + fn test_parse_ws_order_market_sell() { + let data: Value = serde_json::from_str( + r#"{ + "instId": "ETHUSDT", + "ordId": "987654321", + "side": "sell", + "ordType": "market", + "sz": "1.0", + "accFillSz": "1.0", + "avgPx": "2000", + "status": "filled", + "fee": "-2.0", + "feeCcy": "USDT", + "cTime": "1700000000000" + }"#, + ) + .unwrap(); + + let order = parse_ws_order(&data).unwrap(); + assert_eq!(order.symbol, "ETH/USDT"); + assert_eq!(order.side, OrderSide::Sell); + assert_eq!(order.order_type, OrderType::Market); + assert_eq!(order.status, OrderStatus::Closed); + assert_eq!(order.cost, Some(dec!(2000))); // 1.0 * 2000 + } + + #[test] + fn test_parse_ws_order_cancelled() { + let data: Value = serde_json::from_str( + r#"{ + "instId": "BTCUSDT", + "ordId": "111222333", + "side": "buy", + "ordType": "limit", + "px": "45000", + "sz": "0.5", + "status": "cancelled", + "cTime": "1700000000000" + }"#, + ) + .unwrap(); + + let order = parse_ws_order(&data).unwrap(); + assert_eq!(order.status, OrderStatus::Cancelled); + } + + // ==================== Private Trade (My Trade) Parsing Tests ==================== + + #[test] + fn test_parse_ws_my_trade_basic() { + let data: Value = serde_json::from_str( + r#"{ + "instId": "BTCUSDT", + "ordId": "123456789", + "tradeId": "trade001", + "side": "buy", + "fillPx": "50000", + "fillSz": "0.1", + "fillTime": "1700000000000", + "fillFee": "-5.0", + "fillFeeCcy": "USDT" + }"#, + ) + .unwrap(); + + let trade = parse_ws_my_trade(&data).unwrap(); + assert_eq!(trade.symbol, "BTC/USDT"); + assert_eq!(trade.id, Some("trade001".to_string())); + assert_eq!(trade.order, Some("123456789".to_string())); + assert_eq!(trade.side, OrderSide::Buy); + assert_eq!(trade.price, Price::new(dec!(50000))); + assert_eq!(trade.amount, Amount::new(dec!(0.1))); + assert_eq!(trade.cost, Some(Cost::new(dec!(5000)))); // 50000 * 0.1 + assert_eq!(trade.timestamp, 1700000000000); + assert!(trade.fee.is_some()); + let fee = trade.fee.unwrap(); + assert_eq!(fee.cost, dec!(5.0)); // abs value + assert_eq!(fee.currency, "USDT"); + } + + #[test] + fn test_parse_ws_my_trade_sell() { + let data: Value = serde_json::from_str( + r#"{ + "instId": "ETHUSDT", + "ordId": "987654321", + "tradeId": "trade002", + "side": "sell", + "fillPx": "2000", + "fillSz": "0.5", + "fillTime": "1700000000001" + }"#, + ) + .unwrap(); + + let trade = parse_ws_my_trade(&data).unwrap(); + assert_eq!(trade.symbol, "ETH/USDT"); + assert_eq!(trade.side, OrderSide::Sell); + assert_eq!(trade.price, Price::new(dec!(2000))); + assert_eq!(trade.amount, Amount::new(dec!(0.5))); + assert_eq!(trade.cost, Some(Cost::new(dec!(1000)))); // 2000 * 0.5 + assert!(trade.fee.is_none()); // No fee fields + } + + #[test] + fn test_parse_ws_my_trade_fallback_fields() { + // Test fallback field names (price instead of fillPx, etc.) + let data: Value = serde_json::from_str( + r#"{ + "instId": "BTCUSDT", + "ordId": "111222333", + "side": "buy", + "price": "48000", + "baseVolume": "0.2", + "uTime": "1700000000002", + "fee": "-1.5", + "feeCcy": "USDT" + }"#, + ) + .unwrap(); + + let trade = parse_ws_my_trade(&data).unwrap(); + assert_eq!(trade.price, Price::new(dec!(48000))); + assert_eq!(trade.amount, Amount::new(dec!(0.2))); + assert_eq!(trade.timestamp, 1700000000002); + assert!(trade.fee.is_some()); + } } diff --git a/ccxt-exchanges/src/bitget/ws_exchange_impl.rs b/ccxt-exchanges/src/bitget/ws_exchange_impl.rs index 9438416..5f90a1b 100644 --- a/ccxt-exchanges/src/bitget/ws_exchange_impl.rs +++ b/ccxt-exchanges/src/bitget/ws_exchange_impl.rs @@ -126,21 +126,24 @@ impl WsExchange for Bitget { // ==================== Private Data Streams ==================== async fn watch_balance(&self) -> Result> { - Err(Error::not_implemented( - "watch_balance not yet implemented for Bitget", - )) + let ws = self.create_private_ws(); + let auth = self.get_auth()?; + ws.login(&auth).await?; + ws.watch_balance().await } - async fn watch_orders(&self, _symbol: Option<&str>) -> Result> { - Err(Error::not_implemented( - "watch_orders not yet implemented for Bitget", - )) + async fn watch_orders(&self, symbol: Option<&str>) -> Result> { + let ws = self.create_private_ws(); + let auth = self.get_auth()?; + ws.login(&auth).await?; + ws.watch_orders(symbol.map(ToString::to_string)).await } - async fn watch_my_trades(&self, _symbol: Option<&str>) -> Result> { - Err(Error::not_implemented( - "watch_my_trades not yet implemented for Bitget", - )) + async fn watch_my_trades(&self, symbol: Option<&str>) -> Result> { + let ws = self.create_private_ws(); + let auth = self.get_auth()?; + ws.login(&auth).await?; + ws.watch_my_trades(symbol.map(ToString::to_string)).await } // ==================== Subscription Management ==================== diff --git a/ccxt-exchanges/src/okx/margin_impl.rs b/ccxt-exchanges/src/okx/margin_impl.rs new file mode 100644 index 0000000..8830540 --- /dev/null +++ b/ccxt-exchanges/src/okx/margin_impl.rs @@ -0,0 +1,192 @@ +//! Margin trait implementation for OKX. +//! +//! Implements the `Margin` trait from `ccxt-core` for OKX, providing +//! position management, leverage configuration, margin mode, and funding rate operations. + +use async_trait::async_trait; +use ccxt_core::{ + Result, + capability::ExchangeCapabilities, + traits::{Margin, PublicExchange}, + types::{ + FundingRate, FundingRateHistory, Position, Timeframe, + params::{LeverageParams, MarginMode}, + }, +}; + +use super::Okx; + +// ============================================================================ +// PublicExchange Implementation +// ============================================================================ + +impl PublicExchange for Okx { + fn id(&self) -> &'static str { + "okx" + } + + fn name(&self) -> &'static str { + "OKX" + } + + fn version(&self) -> &'static str { + "v5" + } + + fn certified(&self) -> bool { + false + } + + fn capabilities(&self) -> ExchangeCapabilities { + use ccxt_core::exchange::Capability; + + ExchangeCapabilities::builder() + // Market Data + .market_data() + .without_capability(Capability::FetchCurrencies) + .without_capability(Capability::FetchStatus) + .without_capability(Capability::FetchTime) + // Trading + .trading() + .without_capability(Capability::CancelAllOrders) + .without_capability(Capability::EditOrder) + .without_capability(Capability::FetchOrders) + .without_capability(Capability::FetchCanceledOrders) + // Account + .capability(Capability::FetchBalance) + .capability(Capability::FetchMyTrades) + // Margin / Futures + .capability(Capability::FetchPositions) + .capability(Capability::SetLeverage) + .capability(Capability::SetMarginMode) + .capability(Capability::FetchFundingRate) + .capability(Capability::FetchFundingRates) + // WebSocket + .capability(Capability::Websocket) + .capability(Capability::WatchTicker) + .capability(Capability::WatchOrderBook) + .capability(Capability::WatchTrades) + .capability(Capability::WatchBalance) + .capability(Capability::WatchOrders) + .capability(Capability::WatchMyTrades) + .build() + } + + fn timeframes(&self) -> Vec { + vec![ + Timeframe::M1, + Timeframe::M3, + Timeframe::M5, + Timeframe::M15, + Timeframe::M30, + Timeframe::H1, + Timeframe::H2, + Timeframe::H4, + Timeframe::H6, + Timeframe::H12, + Timeframe::D1, + Timeframe::W1, + Timeframe::Mon1, + ] + } + + fn rate_limit(&self) -> u32 { + 20 + } + + fn has_websocket(&self) -> bool { + true + } +} + +// ============================================================================ +// Margin Trait Implementation +// ============================================================================ + +#[async_trait] +impl Margin for Okx { + async fn fetch_positions_for(&self, symbols: &[&str]) -> Result> { + self.fetch_positions_impl(symbols).await + } + + async fn fetch_position(&self, symbol: &str) -> Result { + self.fetch_position_impl(symbol).await + } + + async fn set_leverage_with_params(&self, params: LeverageParams) -> Result<()> { + let margin_mode = params.margin_mode.map(|m| match m { + MarginMode::Isolated => "isolated", + MarginMode::Cross => "cross", + }); + self.set_leverage_impl(¶ms.symbol, params.leverage, margin_mode) + .await + } + + async fn get_leverage(&self, symbol: &str) -> Result { + self.get_leverage_impl(symbol).await + } + + async fn set_margin_mode(&self, symbol: &str, mode: MarginMode) -> Result<()> { + let mode_str = match mode { + MarginMode::Isolated => "isolated", + MarginMode::Cross => "cross", + }; + self.set_margin_mode_impl(symbol, mode_str).await + } + + async fn fetch_funding_rate(&self, symbol: &str) -> Result { + self.fetch_funding_rate_impl(symbol).await + } + + async fn fetch_funding_rates(&self, symbols: &[&str]) -> Result> { + self.fetch_funding_rates_impl(symbols).await + } + + async fn fetch_funding_rate_history( + &self, + symbol: &str, + since: Option, + limit: Option, + ) -> Result> { + self.fetch_funding_rate_history_impl(symbol, since, limit) + .await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use ccxt_core::ExchangeConfig; + + #[test] + fn test_okx_public_exchange_impl() { + let okx = Okx::new(ExchangeConfig::default()).unwrap(); + let pe: &dyn PublicExchange = &okx; + + assert_eq!(pe.id(), "okx"); + assert_eq!(pe.name(), "OKX"); + assert_eq!(pe.version(), "v5"); + assert!(!pe.certified()); + assert!(pe.has_websocket()); + assert_eq!(pe.rate_limit(), 20); + } + + #[test] + fn test_okx_capabilities_include_margin() { + let okx = Okx::new(ExchangeConfig::default()).unwrap(); + let caps = PublicExchange::capabilities(&okx); + + assert!(caps.has("fetchPositions")); + assert!(caps.has("setLeverage")); + assert!(caps.has("setMarginMode")); + assert!(caps.has("fetchFundingRate")); + assert!(caps.has("fetchFundingRates")); + } + + #[test] + fn test_okx_margin_trait_object_safety() { + let okx = Okx::new(ExchangeConfig::default()).unwrap(); + // Verify trait is object-safe by creating a trait object + let _margin: Box = Box::new(okx); + } +} diff --git a/ccxt-exchanges/src/okx/mod.rs b/ccxt-exchanges/src/okx/mod.rs index f7d4b23..634c60b 100644 --- a/ccxt-exchanges/src/okx/mod.rs +++ b/ccxt-exchanges/src/okx/mod.rs @@ -13,6 +13,7 @@ pub mod builder; pub mod endpoint_router; pub mod error; pub mod exchange_impl; +pub mod margin_impl; pub mod parser; pub mod rest; pub mod signed_request; @@ -332,6 +333,16 @@ impl Okx { ws::OkxWs::new(urls.ws_public) } + /// Creates a new private WebSocket client for OKX. + /// + /// Returns an `OkxWs` instance configured with the private WebSocket URL. + /// The caller must call `login()` on the returned client before subscribing + /// to private channels. + pub fn create_private_ws(&self) -> ws::OkxWs { + let urls = self.urls(); + ws::OkxWs::new(urls.ws_private) + } + /// Creates a new signed request builder for authenticated API requests. /// /// This builder encapsulates the common signing workflow for OKX API requests, diff --git a/ccxt-exchanges/src/okx/parser.rs b/ccxt-exchanges/src/okx/parser.rs index 495fe8c..494218c 100644 --- a/ccxt-exchanges/src/okx/parser.rs +++ b/ccxt-exchanges/src/okx/parser.rs @@ -646,6 +646,223 @@ fn parse_balance_entry(data: &Value, balances: &mut HashMap Result { + use ccxt_core::types::position::PositionSide; + + let pos_side_str = data["posSide"].as_str().unwrap_or("net"); + let position_side = match pos_side_str.to_lowercase().as_str() { + "long" => PositionSide::Long, + "short" => PositionSide::Short, + _ => PositionSide::Both, + }; + + let pos = parse_f64_field(data, "pos").unwrap_or(0.0); + let avg_px = parse_f64_field(data, "avgPx"); + let mark_px = parse_f64_field(data, "markPx"); + let upl = parse_f64_field(data, "upl"); + let lever = parse_f64_field(data, "lever"); + let liq_px = parse_f64_field(data, "liqPx"); + let imr = parse_f64_field(data, "imr"); + let mmr = parse_f64_field(data, "mmr"); + let notional_usd = parse_f64_field(data, "notionalUsd"); + let margin = parse_f64_field(data, "margin"); + let realized_pnl = parse_f64_field(data, "realizedPnl"); + + let mgn_mode = data["mgnMode"].as_str().unwrap_or("cross"); + let margin_mode = Some(mgn_mode.to_string()); + + let timestamp = parse_timestamp(data, "uTime").or_else(|| parse_timestamp(data, "cTime")); + let datetime = timestamp.and_then(timestamp_to_datetime); + + // Determine side from position quantity or posSide + let side = match position_side { + PositionSide::Long => Some("long".to_string()), + PositionSide::Short => Some("short".to_string()), + PositionSide::Both => { + if pos > 0.0 { + Some("long".to_string()) + } else if pos < 0.0 { + Some("short".to_string()) + } else { + None + } + } + }; + + let contracts = Some(pos.abs()); + + // Calculate initial margin percentage from leverage + let initial_margin_percentage = lever.map(|l| if l > 0.0 { 1.0 / l } else { 0.0 }); + + // Calculate notional value + let notional = notional_usd.or(match (avg_px, contracts) { + (Some(price), Some(qty)) => Some(price * qty), + _ => None, + }); + + // Calculate percentage PnL + let percentage = match (upl, margin.or(imr)) { + (Some(pnl), Some(m)) if m > 0.0 => Some((pnl / m) * 100.0), + _ => None, + }; + + let hedged = match position_side { + PositionSide::Both => Some(false), + _ => Some(true), + }; + + Ok(ccxt_core::types::Position { + info: data.clone(), + id: data["posId"].as_str().map(ToString::to_string), + symbol: symbol.to_string(), + side, + position_side: Some(position_side), + dual_side_position: hedged, + contracts, + contract_size: parse_f64_field(data, "ctVal"), + entry_price: avg_px, + mark_price: mark_px, + notional, + leverage: lever, + collateral: margin, + initial_margin: imr, + initial_margin_percentage, + maintenance_margin: mmr, + maintenance_margin_percentage: None, + unrealized_pnl: upl, + realized_pnl, + liquidation_price: liq_px, + margin_ratio: None, + margin_mode, + hedged, + percentage, + timestamp, + datetime, + }) +} + +/// Parse funding rate data from OKX public funding-rate response. +/// +/// OKX funding rate fields: +/// - instId: instrument ID +/// - fundingRate: current funding rate +/// - fundingTime: next funding time +/// - nextFundingRate: estimated next funding rate +/// - nextFundingTime: next funding settlement time +/// +/// # Arguments +/// +/// * `data` - OKX funding rate data JSON object +/// * `symbol` - Unified symbol string +/// +/// # Returns +/// +/// Returns a CCXT [`FundingRate`] structure. +pub fn parse_funding_rate(data: &Value, symbol: &str) -> Result { + let funding_rate = parse_f64_field(data, "fundingRate"); + let funding_time = parse_timestamp(data, "fundingTime"); + let next_funding_time = parse_timestamp(data, "nextFundingTime"); + + let timestamp = funding_time.or_else(|| parse_timestamp(data, "ts")); + let datetime = timestamp.and_then(timestamp_to_datetime); + + let funding_datetime = next_funding_time.and_then(timestamp_to_datetime); + + Ok(ccxt_core::types::FundingRate { + info: data.clone(), + symbol: symbol.to_string(), + mark_price: parse_f64_field(data, "markPx"), + index_price: parse_f64_field(data, "idxPx"), + interest_rate: None, + estimated_settle_price: None, + funding_rate, + funding_timestamp: next_funding_time, + funding_datetime, + previous_funding_rate: None, + previous_funding_timestamp: None, + previous_funding_datetime: None, + timestamp, + datetime, + }) +} + +/// Parse funding rate history data from OKX public funding-rate-history response. +/// +/// OKX funding rate history fields: +/// - instId: instrument ID +/// - fundingRate: historical funding rate +/// - fundingTime: funding settlement time +/// - realizedRate: realized funding rate +/// +/// # Arguments +/// +/// * `data` - OKX funding rate history data JSON object +/// * `symbol` - Unified symbol string +/// +/// # Returns +/// +/// Returns a CCXT [`FundingRateHistory`] structure. +pub fn parse_funding_rate_history( + data: &Value, + symbol: &str, +) -> Result { + let funding_rate = + parse_f64_field(data, "fundingRate").or_else(|| parse_f64_field(data, "realizedRate")); + let timestamp = parse_timestamp(data, "fundingTime"); + let datetime = timestamp.and_then(timestamp_to_datetime); + + Ok(ccxt_core::types::FundingRateHistory { + info: data.clone(), + symbol: symbol.to_string(), + funding_rate, + timestamp, + datetime, + }) +} + +/// Helper to parse a string field as f64. +fn parse_f64_field(data: &Value, field: &str) -> Option { + data[field] + .as_str() + .and_then(|s| { + if s.is_empty() { + None + } else { + s.parse::().ok() + } + }) + .or_else(|| data[field].as_f64()) +} + // ============================================================================ // Unit Tests // ============================================================================ @@ -828,4 +1045,154 @@ mod tests { let dt = timestamp_to_datetime(ts).unwrap(); assert!(dt.contains("2023-11-14")); } + + // ======================================================================== + // Position Parser Tests + // ======================================================================== + + #[test] + fn test_parse_position_long() { + let data = json!({ + "instId": "BTC-USDT-SWAP", + "posId": "12345", + "posSide": "long", + "pos": "1.5", + "avgPx": "50000.00", + "markPx": "51000.00", + "upl": "1500.00", + "lever": "10", + "liqPx": "45000.00", + "mgnMode": "cross", + "imr": "5000.00", + "mmr": "500.00", + "notionalUsd": "76500.00", + "margin": "5000.00", + "uTime": "1700000000000" + }); + + let position = parse_position(&data, "BTC/USDT:USDT").unwrap(); + assert_eq!(position.symbol, "BTC/USDT:USDT"); + assert_eq!(position.side, Some("long".to_string())); + assert_eq!(position.contracts, Some(1.5)); + assert_eq!(position.entry_price, Some(50000.00)); + assert_eq!(position.mark_price, Some(51000.00)); + assert_eq!(position.unrealized_pnl, Some(1500.00)); + assert_eq!(position.leverage, Some(10.0)); + assert_eq!(position.liquidation_price, Some(45000.00)); + assert_eq!(position.margin_mode, Some("cross".to_string())); + assert_eq!(position.initial_margin, Some(5000.00)); + assert_eq!(position.maintenance_margin, Some(500.00)); + assert_eq!(position.hedged, Some(true)); + } + + #[test] + fn test_parse_position_short() { + let data = json!({ + "instId": "ETH-USDT-SWAP", + "posSide": "short", + "pos": "-10", + "avgPx": "3000.00", + "markPx": "2900.00", + "upl": "1000.00", + "lever": "5", + "mgnMode": "isolated", + "uTime": "1700000000000" + }); + + let position = parse_position(&data, "ETH/USDT:USDT").unwrap(); + assert_eq!(position.side, Some("short".to_string())); + assert_eq!(position.contracts, Some(10.0)); + assert_eq!(position.margin_mode, Some("isolated".to_string())); + assert_eq!(position.hedged, Some(true)); + } + + #[test] + fn test_parse_position_net_mode() { + let data = json!({ + "instId": "BTC-USDT-SWAP", + "posSide": "net", + "pos": "2", + "avgPx": "50000.00", + "lever": "10", + "mgnMode": "cross", + "uTime": "1700000000000" + }); + + let position = parse_position(&data, "BTC/USDT:USDT").unwrap(); + assert_eq!(position.side, Some("long".to_string())); + assert_eq!(position.contracts, Some(2.0)); + assert_eq!(position.hedged, Some(false)); + } + + #[test] + fn test_parse_position_empty_fields() { + let data = json!({ + "instId": "BTC-USDT-SWAP", + "posSide": "net", + "pos": "0", + "avgPx": "", + "markPx": "", + "upl": "", + "lever": "10", + "mgnMode": "cross" + }); + + let position = parse_position(&data, "BTC/USDT:USDT").unwrap(); + assert_eq!(position.contracts, Some(0.0)); + assert_eq!(position.entry_price, None); + assert_eq!(position.mark_price, None); + assert_eq!(position.unrealized_pnl, None); + } + + // ======================================================================== + // Funding Rate Parser Tests + // ======================================================================== + + #[test] + fn test_parse_funding_rate() { + let data = json!({ + "instId": "BTC-USDT-SWAP", + "fundingRate": "0.0001", + "fundingTime": "1700000000000", + "nextFundingRate": "0.00015", + "nextFundingTime": "1700028800000" + }); + + let rate = parse_funding_rate(&data, "BTC/USDT:USDT").unwrap(); + assert_eq!(rate.symbol, "BTC/USDT:USDT"); + assert_eq!(rate.funding_rate, Some(0.0001)); + assert_eq!(rate.funding_timestamp, Some(1700028800000)); + assert_eq!(rate.timestamp, Some(1700000000000)); + } + + #[test] + fn test_parse_funding_rate_history() { + let data = json!({ + "instId": "BTC-USDT-SWAP", + "fundingRate": "0.0001", + "fundingTime": "1700000000000", + "realizedRate": "0.00009" + }); + + let history = parse_funding_rate_history(&data, "BTC/USDT:USDT").unwrap(); + assert_eq!(history.symbol, "BTC/USDT:USDT"); + assert_eq!(history.funding_rate, Some(0.0001)); + assert_eq!(history.timestamp, Some(1700000000000)); + } + + #[test] + fn test_parse_f64_field() { + let data = json!({ + "a": "123.45", + "b": "", + "c": 67.89, + "d": null + }); + + assert_eq!(parse_f64_field(&data, "a"), Some(123.45)); + assert_eq!(parse_f64_field(&data, "b"), None); + assert_eq!(parse_f64_field(&data, "c"), Some(67.89)); + assert_eq!(parse_f64_field(&data, "d"), None); + assert_eq!(parse_f64_field(&data, "missing"), None); + } } diff --git a/ccxt-exchanges/src/okx/rest/futures/funding.rs b/ccxt-exchanges/src/okx/rest/futures/funding.rs new file mode 100644 index 0000000..4828d30 --- /dev/null +++ b/ccxt-exchanges/src/okx/rest/futures/funding.rs @@ -0,0 +1,97 @@ +//! Funding rate operations for OKX futures/swap. + +use super::super::super::{Okx, parser}; +use ccxt_core::{ + Error, ParseError, Result, + types::{FundingRate, FundingRateHistory}, +}; +use std::collections::HashMap; +use tracing::warn; + +impl Okx { + /// Fetch current funding rate for a symbol. + /// + /// Uses OKX GET `/api/v5/public/funding-rate` (public endpoint). + pub async fn fetch_funding_rate_impl(&self, symbol: &str) -> Result { + let inst_id = Self::symbol_to_inst_id(symbol); + + let mut params = HashMap::new(); + params.insert("instId".to_string(), inst_id); + + let data = self + .public_request("GET", "/api/v5/public/funding-rate", Some(¶ms)) + .await?; + + let rates_array = data["data"].as_array().ok_or_else(|| { + Error::from(ParseError::invalid_format("data", "Expected data array")) + })?; + + if rates_array.is_empty() { + return Err(Error::from(ParseError::missing_field_owned(format!( + "No funding rate found for symbol: {}", + symbol + )))); + } + + parser::parse_funding_rate(&rates_array[0], symbol) + } + + /// Fetch funding rates for multiple symbols. + /// + /// Calls `fetch_funding_rate_impl` for each symbol. + pub async fn fetch_funding_rates_impl(&self, symbols: &[&str]) -> Result> { + let mut rates = Vec::new(); + for symbol in symbols { + match self.fetch_funding_rate_impl(symbol).await { + Ok(rate) => rates.push(rate), + Err(e) => { + warn!(error = %e, symbol = %symbol, "Failed to fetch funding rate"); + } + } + } + Ok(rates) + } + + /// Fetch funding rate history for a symbol. + /// + /// Uses OKX GET `/api/v5/public/funding-rate-history` (public endpoint). + pub async fn fetch_funding_rate_history_impl( + &self, + symbol: &str, + since: Option, + limit: Option, + ) -> Result> { + let inst_id = Self::symbol_to_inst_id(symbol); + + let mut params = HashMap::new(); + params.insert("instId".to_string(), inst_id); + + if let Some(s) = since { + params.insert("before".to_string(), s.to_string()); + } + + if let Some(l) = limit { + params.insert("limit".to_string(), l.to_string()); + } + + let data = self + .public_request("GET", "/api/v5/public/funding-rate-history", Some(¶ms)) + .await?; + + let history_array = data["data"].as_array().ok_or_else(|| { + Error::from(ParseError::invalid_format("data", "Expected data array")) + })?; + + let mut history = Vec::new(); + for item in history_array { + match parser::parse_funding_rate_history(item, symbol) { + Ok(record) => history.push(record), + Err(e) => { + warn!(error = %e, "Failed to parse OKX funding rate history"); + } + } + } + + Ok(history) + } +} diff --git a/ccxt-exchanges/src/okx/rest/futures/leverage.rs b/ccxt-exchanges/src/okx/rest/futures/leverage.rs new file mode 100644 index 0000000..eedcd30 --- /dev/null +++ b/ccxt-exchanges/src/okx/rest/futures/leverage.rs @@ -0,0 +1,105 @@ +//! Leverage operations for OKX futures/swap. + +use super::super::super::{Okx, signed_request::HttpMethod}; +use ccxt_core::{Error, ParseError, Result}; + +impl Okx { + /// Set leverage for a symbol. + /// + /// Uses OKX POST `/api/v5/account/set-leverage` endpoint. + /// + /// # Arguments + /// + /// * `symbol` - Unified symbol (e.g., "BTC/USDT:USDT") + /// * `leverage` - Leverage multiplier + /// * `margin_mode` - "cross" or "isolated" + pub async fn set_leverage_impl( + &self, + symbol: &str, + leverage: u32, + margin_mode: Option<&str>, + ) -> Result<()> { + let inst_id = Self::symbol_to_inst_id(symbol); + let mgn_mode = margin_mode.unwrap_or("cross"); + + let mut builder = self + .signed_request("/api/v5/account/set-leverage") + .method(HttpMethod::Post) + .param("instId", &inst_id) + .param("lever", leverage.to_string()) + .param("mgnMode", mgn_mode); + + // For isolated mode, we need to specify posSide + if mgn_mode == "isolated" { + // Set for both long and short sides + builder = builder.param("posSide", "net"); + } + + let data = builder.execute().await?; + + // Check for nested error in data array + if let Some(arr) = data["data"].as_array() { + if let Some(first) = arr.first() { + if let Some(code) = first["sCode"].as_str() { + if code != "0" { + let msg = first["sMsg"].as_str().unwrap_or("Unknown error"); + return Err(Error::exchange(code, msg)); + } + } + } + } + + Ok(()) + } + + /// Get current leverage for a symbol. + /// + /// Uses OKX GET `/api/v5/account/leverage-info` endpoint. + pub async fn get_leverage_impl(&self, symbol: &str) -> Result { + let inst_id = Self::symbol_to_inst_id(symbol); + + let data = self + .signed_request("/api/v5/account/leverage-info") + .param("instId", &inst_id) + .param("mgnMode", "cross") + .execute() + .await?; + + let leverage_array = data["data"].as_array().ok_or_else(|| { + Error::from(ParseError::invalid_format("data", "Expected data array")) + })?; + + if leverage_array.is_empty() { + return Err(Error::from(ParseError::missing_field_owned(format!( + "No leverage info found for symbol: {}", + symbol + )))); + } + + let lever_str = leverage_array[0]["lever"] + .as_str() + .ok_or_else(|| Error::from(ParseError::missing_field("lever")))?; + + lever_str.parse::().map(|v| v as u32).map_err(|_| { + Error::from(ParseError::invalid_value( + "lever", + format!("Cannot parse leverage: {}", lever_str), + )) + }) + } +} + +#[cfg(test)] +mod tests { + #[test] + fn test_leverage_parse() { + // Test that leverage string parsing works + let lever_str = "10"; + let leverage: u32 = lever_str.parse::().unwrap() as u32; + assert_eq!(leverage, 10); + + let lever_str = "5.00"; + let leverage: u32 = lever_str.parse::().unwrap() as u32; + assert_eq!(leverage, 5); + } +} diff --git a/ccxt-exchanges/src/okx/rest/futures/margin.rs b/ccxt-exchanges/src/okx/rest/futures/margin.rs new file mode 100644 index 0000000..fb23c22 --- /dev/null +++ b/ccxt-exchanges/src/okx/rest/futures/margin.rs @@ -0,0 +1,59 @@ +//! Margin mode operations for OKX futures/swap. + +use super::super::super::{Okx, signed_request::HttpMethod}; +use ccxt_core::{Error, Result}; + +impl Okx { + /// Set margin mode (cross or isolated) for a symbol. + /// + /// Uses OKX POST `/api/v5/account/set-position-mode` for position mode, + /// and the leverage endpoint implicitly sets margin mode. + /// + /// Note: OKX sets margin mode per-instrument via the set-leverage endpoint's + /// `mgnMode` parameter. This method provides a dedicated interface. + pub async fn set_margin_mode_impl(&self, symbol: &str, mode: &str) -> Result<()> { + let inst_id = Self::symbol_to_inst_id(symbol); + + let mgn_mode = match mode.to_lowercase().as_str() { + "isolated" => "isolated", + "cross" | "crossed" => "cross", + _ => { + return Err(Error::invalid_request(format!( + "Invalid margin mode: {}. Must be 'isolated' or 'cross'", + mode + ))); + } + }; + + // OKX sets margin mode through the set-leverage endpoint + // We need to get current leverage first, then re-set it with the new margin mode + let current_leverage = self.get_leverage_impl(symbol).await.unwrap_or(10); + + let mut builder = self + .signed_request("/api/v5/account/set-leverage") + .method(HttpMethod::Post) + .param("instId", &inst_id) + .param("lever", current_leverage.to_string()) + .param("mgnMode", mgn_mode); + + if mgn_mode == "isolated" { + builder = builder.param("posSide", "net"); + } + + let data = builder.execute().await?; + + // Check for nested error + if let Some(arr) = data["data"].as_array() { + if let Some(first) = arr.first() { + if let Some(code) = first["sCode"].as_str() { + if code != "0" { + let msg = first["sMsg"].as_str().unwrap_or("Unknown error"); + return Err(Error::exchange(code, msg)); + } + } + } + } + + Ok(()) + } +} diff --git a/ccxt-exchanges/src/okx/rest/futures/mod.rs b/ccxt-exchanges/src/okx/rest/futures/mod.rs new file mode 100644 index 0000000..b649a5d --- /dev/null +++ b/ccxt-exchanges/src/okx/rest/futures/mod.rs @@ -0,0 +1,9 @@ +//! OKX futures/swap operations. +//! +//! This module contains methods for managing positions, leverage, +//! funding rates, and margin mode for OKX perpetual swaps and futures contracts. + +mod funding; +mod leverage; +mod margin; +mod positions; diff --git a/ccxt-exchanges/src/okx/rest/futures/positions.rs b/ccxt-exchanges/src/okx/rest/futures/positions.rs new file mode 100644 index 0000000..af4fa22 --- /dev/null +++ b/ccxt-exchanges/src/okx/rest/futures/positions.rs @@ -0,0 +1,213 @@ +//! Position management methods for OKX futures/swap. + +use super::super::super::{Okx, parser}; +use ccxt_core::{Error, ParseError, Result, types::Position}; +use tracing::warn; + +impl Okx { + /// Map a unified symbol to OKX instType parameter. + /// + /// Returns the appropriate instType for the given symbol: + /// - Symbols with `:` and expiry date → "FUTURES" + /// - Symbols with `:` (perpetual) → "SWAP" + /// - Otherwise → "SWAP" (default for contract queries) + pub(crate) fn inst_type_from_symbol(symbol: &str) -> &'static str { + if symbol.contains(':') { + // Check if it has an expiry date (e.g., BTC/USDT:USDT-241231) + if symbol.contains('-') + && symbol + .rsplit('-') + .next() + .is_some_and(|s| s.len() == 6 && s.chars().all(|c| c.is_ascii_digit())) + { + "FUTURES" + } else { + "SWAP" + } + } else { + "SWAP" + } + } + + /// Convert a unified symbol to OKX instId. + /// + /// Examples: + /// - "BTC/USDT:USDT" → "BTC-USDT-SWAP" + /// - "BTC/USD:BTC" → "BTC-USD-SWAP" + /// - "BTC/USDT" → "BTC-USDT" + pub(crate) fn symbol_to_inst_id(symbol: &str) -> String { + if let Some(pos) = symbol.find(':') { + let base_quote = &symbol[..pos]; + let settle_part = &symbol[pos + 1..]; + let base_quote_dash = base_quote.replace('/', "-"); + + if let Some(dash_pos) = settle_part.find('-') { + // Futures with expiry: BTC/USDT:USDT-241231 → BTC-USDT-241231 + let expiry = &settle_part[dash_pos + 1..]; + format!("{}-{}", base_quote_dash, expiry) + } else { + // Perpetual swap: BTC/USDT:USDT → BTC-USDT-SWAP + format!("{}-SWAP", base_quote_dash) + } + } else { + // Spot: BTC/USDT → BTC-USDT + symbol.replace('/', "-") + } + } + + /// Fetch a single position for a symbol. + /// + /// Uses OKX GET `/api/v5/account/positions` endpoint. + pub async fn fetch_position_impl(&self, symbol: &str) -> Result { + let inst_id = Self::symbol_to_inst_id(symbol); + let inst_type = Self::inst_type_from_symbol(symbol); + + let data = self + .signed_request("/api/v5/account/positions") + .param("instId", &inst_id) + .param("instType", inst_type) + .execute() + .await?; + + let positions_array = data["data"].as_array().ok_or_else(|| { + Error::from(ParseError::invalid_format("data", "Expected data array")) + })?; + + if positions_array.is_empty() { + return Err(Error::from(ParseError::missing_field_owned(format!( + "No position found for symbol: {}", + symbol + )))); + } + + parser::parse_position(&positions_array[0], symbol) + } + + /// Fetch positions for specific symbols, or all positions if empty. + /// + /// Uses OKX GET `/api/v5/account/positions` endpoint. + pub async fn fetch_positions_impl(&self, symbols: &[&str]) -> Result> { + let inst_type = if symbols.is_empty() { + // Fetch all — use the configured default type + self.get_inst_type() + } else { + Self::inst_type_from_symbol(symbols[0]) + }; + + let mut builder = self + .signed_request("/api/v5/account/positions") + .param("instType", inst_type); + + // OKX supports filtering by instId (comma-separated for up to 10) + if !symbols.is_empty() && symbols.len() <= 10 { + let inst_ids: Vec = + symbols.iter().map(|s| Self::symbol_to_inst_id(s)).collect(); + builder = builder.param("instId", inst_ids.join(",")); + } + + let data = builder.execute().await?; + + let positions_array = data["data"].as_array().ok_or_else(|| { + Error::from(ParseError::invalid_format("data", "Expected data array")) + })?; + + let mut positions = Vec::new(); + for position_data in positions_array { + let inst_id = position_data["instId"].as_str().unwrap_or_default(); + // Derive symbol from instId for parsing + let symbol_hint = Self::inst_id_to_symbol_hint(inst_id); + + match parser::parse_position(position_data, &symbol_hint) { + Ok(position) => { + // Filter by requested symbols if specified + if symbols.is_empty() || symbols.contains(&position.symbol.as_str()) { + positions.push(position); + } + } + Err(e) => { + warn!(error = %e, inst_id = %inst_id, "Failed to parse OKX position"); + } + } + } + + Ok(positions) + } + + /// Convert an OKX instId back to a rough unified symbol hint. + /// + /// This is a best-effort conversion used when market data isn't loaded. + /// - "BTC-USDT-SWAP" → "BTC/USDT:USDT" + /// - "BTC-USD-SWAP" → "BTC/USD:BTC" + /// - "BTC-USDT-241231" → "BTC/USDT:USDT-241231" + /// - "BTC-USDT" → "BTC/USDT" + fn inst_id_to_symbol_hint(inst_id: &str) -> String { + let parts: Vec<&str> = inst_id.split('-').collect(); + match parts.len() { + 2 => { + // Spot: BTC-USDT → BTC/USDT + format!("{}/{}", parts[0], parts[1]) + } + 3 if parts[2] == "SWAP" => { + // Perpetual: BTC-USDT-SWAP → BTC/USDT:USDT or BTC/USD:BTC + let settle = if parts[1] == "USD" { + parts[0] + } else { + parts[1] + }; + format!("{}/{}:{}", parts[0], parts[1], settle) + } + 3 => { + // Futures with expiry: BTC-USDT-241231 → BTC/USDT:USDT-241231 + let settle = if parts[1] == "USD" { + parts[0] + } else { + parts[1] + }; + format!("{}/{}:{}-{}", parts[0], parts[1], settle, parts[2]) + } + _ => inst_id.to_string(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_inst_type_from_symbol() { + assert_eq!(Okx::inst_type_from_symbol("BTC/USDT:USDT"), "SWAP"); + assert_eq!(Okx::inst_type_from_symbol("BTC/USD:BTC"), "SWAP"); + assert_eq!( + Okx::inst_type_from_symbol("BTC/USDT:USDT-241231"), + "FUTURES" + ); + assert_eq!(Okx::inst_type_from_symbol("BTC/USDT"), "SWAP"); + } + + #[test] + fn test_symbol_to_inst_id() { + assert_eq!(Okx::symbol_to_inst_id("BTC/USDT:USDT"), "BTC-USDT-SWAP"); + assert_eq!(Okx::symbol_to_inst_id("BTC/USD:BTC"), "BTC-USD-SWAP"); + assert_eq!( + Okx::symbol_to_inst_id("BTC/USDT:USDT-241231"), + "BTC-USDT-241231" + ); + assert_eq!(Okx::symbol_to_inst_id("BTC/USDT"), "BTC-USDT"); + assert_eq!(Okx::symbol_to_inst_id("ETH/USD:ETH"), "ETH-USD-SWAP"); + } + + #[test] + fn test_inst_id_to_symbol_hint() { + assert_eq!( + Okx::inst_id_to_symbol_hint("BTC-USDT-SWAP"), + "BTC/USDT:USDT" + ); + assert_eq!(Okx::inst_id_to_symbol_hint("BTC-USD-SWAP"), "BTC/USD:BTC"); + assert_eq!( + Okx::inst_id_to_symbol_hint("BTC-USDT-241231"), + "BTC/USDT:USDT-241231" + ); + assert_eq!(Okx::inst_id_to_symbol_hint("BTC-USDT"), "BTC/USDT"); + } +} diff --git a/ccxt-exchanges/src/okx/rest/mod.rs b/ccxt-exchanges/src/okx/rest/mod.rs index 25bcde1..5c64aa7 100644 --- a/ccxt-exchanges/src/okx/rest/mod.rs +++ b/ccxt-exchanges/src/okx/rest/mod.rs @@ -3,6 +3,7 @@ //! Implements all REST API endpoint operations for the OKX exchange. mod account; +pub mod futures; mod market_data; mod trading; diff --git a/ccxt-exchanges/src/okx/ws.rs b/ccxt-exchanges/src/okx/ws.rs index c9b6ec3..6f3f8b1 100644 --- a/ccxt-exchanges/src/okx/ws.rs +++ b/ccxt-exchanges/src/okx/ws.rs @@ -4,17 +4,26 @@ //! Supports public streams (ticker, orderbook, trades) and private streams //! (balance, orders) with automatic reconnection. +use crate::okx::auth::OkxAuth; use crate::okx::parser::{parse_orderbook, parse_ticker, parse_trade}; use ccxt_core::error::{Error, Result}; -use ccxt_core::types::{Market, OrderBook, Ticker, Trade}; +use ccxt_core::types::{ + Balance, BalanceEntry, Fee, Market, Order, OrderBook, OrderSide, OrderStatus, OrderType, + Ticker, Trade, + financial::{Amount, Cost, Price}, +}; use ccxt_core::ws_client::{WsClient, WsConfig, WsConnectionState}; use ccxt_core::ws_exchange::MessageStream; use futures::Stream; +use rust_decimal::Decimal; +use rust_decimal::prelude::FromStr; use serde_json::Value; +use std::collections::HashMap; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; use tokio::sync::{RwLock, mpsc}; +use tracing::{debug, warn}; /// Default ping interval for OKX WebSocket (25 seconds). /// OKX requires ping every 30 seconds, we use 25 for safety margin. @@ -509,6 +518,271 @@ impl OkxWs { Ok(Box::pin(ReceiverStream::new(rx))) } + + // ======================================================================== + // Private WebSocket Methods + // ======================================================================== + + /// Sends a login message to authenticate the private WebSocket connection. + /// + /// OKX WebSocket login format: + /// ```json + /// { + /// "op": "login", + /// "args": [{ + /// "apiKey": "...", + /// "passphrase": "...", + /// "timestamp": "...", + /// "sign": "..." + /// }] + /// } + /// ``` + /// + /// The signature is: HMAC-SHA256(timestamp + "GET" + "/users/self/verify", secret) + pub async fn login(&self, auth: &OkxAuth) -> Result<()> { + let timestamp = (chrono::Utc::now().timestamp_millis() / 1000).to_string(); + let sign = auth.sign(×tamp, "GET", "/users/self/verify", ""); + + #[allow(clippy::disallowed_methods)] + let msg = serde_json::json!({ + "op": "login", + "args": [{ + "apiKey": auth.api_key(), + "passphrase": auth.passphrase(), + "timestamp": timestamp, + "sign": sign + }] + }); + + self.client.send_json(&msg).await?; + + // Wait for login response + let timeout = tokio::time::Duration::from_secs(10); + let start = tokio::time::Instant::now(); + + while start.elapsed() < timeout { + if let Some(resp) = self.client.receive().await { + if let Some(event) = resp.get("event").and_then(|e| e.as_str()) { + if event == "login" { + let code = resp.get("code").and_then(|c| c.as_str()).unwrap_or("1"); + if code == "0" { + debug!("OKX WebSocket login successful"); + return Ok(()); + } + let msg_text = resp + .get("msg") + .and_then(|m| m.as_str()) + .unwrap_or("Unknown error"); + return Err(Error::authentication(format!( + "OKX WebSocket login failed: {} (code: {})", + msg_text, code + ))); + } + } + } + } + + Err(Error::authentication( + "OKX WebSocket login timed out waiting for response", + )) + } + + /// Subscribes to the orders channel (private). + /// + /// Receives real-time order updates for all instrument types. + pub async fn subscribe_orders(&self, inst_type: Option<&str>) -> Result<()> { + let mut arg = serde_json::Map::new(); + arg.insert("channel".to_string(), Value::String("orders".to_string())); + arg.insert( + "instType".to_string(), + Value::String(inst_type.unwrap_or("ANY").to_string()), + ); + + #[allow(clippy::disallowed_methods)] + let msg = serde_json::json!({ + "op": "subscribe", + "args": [Value::Object(arg)] + }); + + self.client.send_json(&msg).await?; + + self.subscriptions.write().await.push("orders".to_string()); + + Ok(()) + } + + /// Subscribes to the account channel (private). + /// + /// Receives real-time balance/account updates. + pub async fn subscribe_account(&self) -> Result<()> { + #[allow(clippy::disallowed_methods)] + let msg = serde_json::json!({ + "op": "subscribe", + "args": [{ + "channel": "account" + }] + }); + + self.client.send_json(&msg).await?; + + self.subscriptions.write().await.push("account".to_string()); + + Ok(()) + } + + /// Watches balance updates via the private WebSocket. + /// + /// Requires prior authentication via `login()`. + pub async fn watch_balance(&self) -> Result> { + if !self.is_connected() { + self.connect().await?; + } + + self.subscribe_account().await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let client = Arc::clone(&self.client); + + tokio::spawn(async move { + while let Some(msg) = client.receive().await { + if is_account_message(&msg) { + match parse_ws_balance(&msg) { + Ok(balance) => { + if tx.send(Ok(balance)).is_err() { + break; + } + } + Err(e) => { + warn!(error = %e, "Failed to parse OKX balance update"); + if tx.send(Err(e)).is_err() { + break; + } + } + } + } + } + }); + + Ok(Box::pin(ReceiverStream::new(rx))) + } + + /// Watches order updates via the private WebSocket. + /// + /// Requires prior authentication via `login()`. + pub async fn watch_orders( + &self, + symbol_filter: Option, + ) -> Result> { + if !self.is_connected() { + self.connect().await?; + } + + self.subscribe_orders(None).await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let client = Arc::clone(&self.client); + + tokio::spawn(async move { + while let Some(msg) = client.receive().await { + if is_orders_message(&msg) { + if let Some(data_array) = msg.get("data").and_then(|d| d.as_array()) { + for data in data_array { + // Apply symbol filter if specified + if let Some(ref filter) = symbol_filter { + let inst_id = + data.get("instId").and_then(|i| i.as_str()).unwrap_or(""); + let unified = inst_id.replace('-', "/"); + if unified != *filter && inst_id != filter.as_str() { + continue; + } + } + + match parse_ws_order(data) { + Ok(order) => { + if tx.send(Ok(order)).is_err() { + return; + } + } + Err(e) => { + warn!(error = %e, "Failed to parse OKX order update"); + if tx.send(Err(e)).is_err() { + return; + } + } + } + } + } + } + } + }); + + Ok(Box::pin(ReceiverStream::new(rx))) + } + + /// Watches trade fills via the private WebSocket (orders channel). + /// + /// Extracts trade information from order execution reports. + /// Requires prior authentication via `login()`. + pub async fn watch_my_trades( + &self, + symbol_filter: Option, + ) -> Result> { + if !self.is_connected() { + self.connect().await?; + } + + self.subscribe_orders(None).await?; + + let (tx, rx) = mpsc::unbounded_channel::>(); + let client = Arc::clone(&self.client); + + tokio::spawn(async move { + while let Some(msg) = client.receive().await { + if is_orders_message(&msg) { + if let Some(data_array) = msg.get("data").and_then(|d| d.as_array()) { + for data in data_array { + // Only emit trades for filled/partially filled orders + let fill_sz = data + .get("fillSz") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0.0); + + if fill_sz <= 0.0 { + continue; + } + + // Apply symbol filter + if let Some(ref filter) = symbol_filter { + let inst_id = + data.get("instId").and_then(|i| i.as_str()).unwrap_or(""); + let unified = inst_id.replace('-', "/"); + if unified != *filter && inst_id != filter.as_str() { + continue; + } + } + + match parse_ws_my_trade(data) { + Ok(trade) => { + if tx.send(Ok(trade)).is_err() { + return; + } + } + Err(e) => { + warn!(error = %e, "Failed to parse OKX trade update"); + if tx.send(Err(e)).is_err() { + return; + } + } + } + } + } + } + } + }); + + Ok(Box::pin(ReceiverStream::new(rx))) + } } // ============================================================================ @@ -647,6 +921,306 @@ pub fn parse_ws_trades(msg: &Value, market: Option<&Market>) -> Result bool { + if let Some(arg) = msg.get("arg") { + arg.get("channel").and_then(|c| c.as_str()) == Some("account") + } else { + false + } +} + +/// Check if a WebSocket message is an orders update. +fn is_orders_message(msg: &Value) -> bool { + if let Some(arg) = msg.get("arg") { + arg.get("channel").and_then(|c| c.as_str()) == Some("orders") + } else { + false + } +} + +// ============================================================================ +// Private WebSocket Message Parsers +// ============================================================================ + +/// Parse a WebSocket account/balance message into a Balance. +/// +/// OKX account update format: +/// ```json +/// { +/// "arg": {"channel": "account"}, +/// "data": [{ +/// "uTime": "1700000000000", +/// "totalEq": "100000.00", +/// "details": [ +/// {"ccy": "USDT", "availBal": "50000", "frozenBal": "10000", "eq": "60000"}, +/// {"ccy": "BTC", "availBal": "1.5", "frozenBal": "0.5", "eq": "2.0"} +/// ] +/// }] +/// } +/// ``` +pub fn parse_ws_balance(msg: &Value) -> Result { + let data = msg + .get("data") + .and_then(|d| d.as_array()) + .and_then(|arr| arr.first()) + .ok_or_else(|| Error::invalid_request("Missing data in account message"))?; + + // uTime is available but Balance struct doesn't have a timestamp field + let mut info = HashMap::new(); + + if let Some(details) = data.get("details").and_then(|d| d.as_array()) { + for detail in details { + let currency = detail + .get("ccy") + .and_then(|c| c.as_str()) + .unwrap_or_default() + .to_string(); + + if currency.is_empty() { + continue; + } + + let free = detail + .get("availBal") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let used = detail + .get("frozenBal") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let total = free + used; + + info.insert(currency, BalanceEntry { free, used, total }); + } + } + + Ok(Balance { + balances: info, + info: HashMap::new(), + }) +} + +/// Parse a WebSocket order update message into an Order. +/// +/// OKX order update format: +/// ```json +/// { +/// "instId": "BTC-USDT", +/// "ordId": "123456", +/// "clOrdId": "client123", +/// "side": "buy", +/// "ordType": "limit", +/// "px": "50000", +/// "sz": "0.1", +/// "fillPx": "49999", +/// "fillSz": "0.05", +/// "avgPx": "49999", +/// "state": "partially_filled", +/// "fee": "-0.5", +/// "feeCcy": "USDT", +/// "uTime": "1700000000000", +/// "cTime": "1700000000000" +/// } +/// ``` +pub fn parse_ws_order(data: &Value) -> Result { + let inst_id = data + .get("instId") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + let symbol = inst_id.replace('-', "/"); + + let id = data + .get("ordId") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + + let client_order_id = data + .get("clOrdId") + .and_then(|v| v.as_str()) + .map(ToString::to_string); + + let side = match data.get("side").and_then(|v| v.as_str()) { + Some("sell") => OrderSide::Sell, + _ => OrderSide::Buy, + }; + + let order_type = match data.get("ordType").and_then(|v| v.as_str()) { + Some("market") => OrderType::Market, + _ => OrderType::Limit, + }; + + let price = data + .get("px") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()); + + let amount = data + .get("sz") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let filled = data + .get("accFillSz") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()); + + let average = data + .get("avgPx") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .filter(|d| !d.is_zero()); + + let cost = match (filled, average) { + (Some(f), Some(a)) => Some(f * a), + _ => None, + }; + + let status = match data.get("state").and_then(|v| v.as_str()) { + Some("filled") => OrderStatus::Closed, + Some("canceled" | "cancelled") => OrderStatus::Cancelled, + _ => OrderStatus::Open, + }; + + let fee = { + let fee_amount = data + .get("fee") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()); + let fee_currency = data + .get("feeCcy") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + fee_amount.map(|f| Fee::new(fee_currency, f.abs())) + }; + + let timestamp = data + .get("cTime") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()); + + Ok(Order { + id, + client_order_id, + symbol, + order_type, + side, + price, + amount, + filled, + remaining: None, + cost, + average, + status, + fee, + fees: None, + timestamp, + datetime: None, + last_trade_timestamp: None, + time_in_force: None, + post_only: None, + reduce_only: None, + stop_price: None, + trigger_price: None, + take_profit_price: None, + stop_loss_price: None, + trailing_delta: None, + trailing_percent: None, + activation_price: None, + callback_rate: None, + working_type: None, + trades: None, + info: HashMap::new(), + }) +} + +/// Parse a WebSocket order fill into a Trade (my trade). +/// +/// Extracts the latest fill information from an order update. +pub fn parse_ws_my_trade(data: &Value) -> Result { + let inst_id = data + .get("instId") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + let symbol = inst_id.replace('-', "/"); + + let trade_id = data + .get("tradeId") + .and_then(|v| v.as_str()) + .map(ToString::to_string); + + let order_id = data + .get("ordId") + .and_then(|v| v.as_str()) + .map(ToString::to_string); + + let side = match data.get("side").and_then(|v| v.as_str()) { + Some("buy") => OrderSide::Buy, + _ => OrderSide::Sell, + }; + + let fill_px = data + .get("fillPx") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let fill_sz = data + .get("fillSz") + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()) + .unwrap_or_default(); + + let timestamp = data + .get("fillTime") + .or_else(|| data.get("uTime")) + .and_then(|v| v.as_str()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(0); + + let fee = { + let fee_amount = data + .get("fillFee") + .or_else(|| data.get("fee")) + .and_then(|v| v.as_str()) + .and_then(|s| Decimal::from_str(s).ok()); + let fee_currency = data + .get("fillFeeCcy") + .or_else(|| data.get("feeCcy")) + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + fee_amount.map(|f| Fee::new(fee_currency, f.abs())) + }; + + Ok(Trade { + id: trade_id, + order: order_id, + symbol, + trade_type: None, + side, + taker_or_maker: None, + price: Price(fill_px), + amount: Amount(fill_sz), + cost: Some(Cost(fill_px * fill_sz)), + fee, + timestamp, + datetime: None, + info: HashMap::new(), + }) +} + #[cfg(test)] mod tests { use super::*; diff --git a/ccxt-exchanges/src/okx/ws_exchange_impl.rs b/ccxt-exchanges/src/okx/ws_exchange_impl.rs index 774b415..0f2c918 100644 --- a/ccxt-exchanges/src/okx/ws_exchange_impl.rs +++ b/ccxt-exchanges/src/okx/ws_exchange_impl.rs @@ -100,21 +100,24 @@ impl WsExchange for Okx { // ==================== Private Data Streams ==================== async fn watch_balance(&self) -> Result> { - Err(Error::not_implemented( - "watch_balance not yet implemented for OKX", - )) + let ws = self.create_private_ws(); + let auth = self.get_auth()?; + ws.login(&auth).await?; + ws.watch_balance().await } - async fn watch_orders(&self, _symbol: Option<&str>) -> Result> { - Err(Error::not_implemented( - "watch_orders not yet implemented for OKX", - )) + async fn watch_orders(&self, symbol: Option<&str>) -> Result> { + let ws = self.create_private_ws(); + let auth = self.get_auth()?; + ws.login(&auth).await?; + ws.watch_orders(symbol.map(ToString::to_string)).await } - async fn watch_my_trades(&self, _symbol: Option<&str>) -> Result> { - Err(Error::not_implemented( - "watch_my_trades not yet implemented for OKX", - )) + async fn watch_my_trades(&self, symbol: Option<&str>) -> Result> { + let ws = self.create_private_ws(); + let auth = self.get_auth()?; + ws.login(&auth).await?; + ws.watch_my_trades(symbol.map(ToString::to_string)).await } // ==================== Subscription Management ====================