From d97220d6076d79ef34b96d4b704f9e7da0e607c3 Mon Sep 17 00:00:00 2001 From: Praying Date: Fri, 13 Feb 2026 21:21:06 +0800 Subject: [PATCH] fix(okx): harden market parsing and websocket stream handling Body Ensure OKX symbols resolve even when base/quote are missing and skip empty WS payloads that caused errors. Add OKX examples to validate usage across market data, futures, account, and WebSocket flows. --- ccxt-exchanges/src/okx/parser.rs | 44 +++++- ccxt-exchanges/src/okx/ws.rs | 29 +++- examples/okx_account_example.rs | 105 +++++++++++++ examples/okx_example.rs | 6 +- examples/okx_futures_example.rs | 99 +++++++++++++ examples/okx_market_data_example.rs | 137 +++++++++++++++++ examples/okx_ws_example.rs | 221 ++++++++++++++++++++++++++++ 7 files changed, 630 insertions(+), 11 deletions(-) create mode 100644 examples/okx_account_example.rs create mode 100644 examples/okx_futures_example.rs create mode 100644 examples/okx_market_data_example.rs create mode 100644 examples/okx_ws_example.rs diff --git a/ccxt-exchanges/src/okx/parser.rs b/ccxt-exchanges/src/okx/parser.rs index 494218c..78d27a4 100644 --- a/ccxt-exchanges/src/okx/parser.rs +++ b/ccxt-exchanges/src/okx/parser.rs @@ -42,16 +42,28 @@ pub fn parse_market(data: &Value) -> Result { .ok_or_else(|| Error::from(ParseError::missing_field("instId")))? .to_string(); + // Parse instId parts for fallback + let parts: Vec<&str> = id.split('-').collect(); + let (base_from_id, quote_from_id) = if parts.len() >= 2 { + (Some(parts[0]), Some(parts[1])) + } else { + (None, None) + }; + // Base and quote currencies let base = data["baseCcy"] .as_str() - .ok_or_else(|| Error::from(ParseError::missing_field("baseCcy")))? - .to_string(); + .filter(|s| !s.is_empty()) + .map(ToString::to_string) + .or_else(|| base_from_id.map(ToString::to_string)) + .ok_or_else(|| Error::from(ParseError::missing_field("baseCcy")))?; let quote = data["quoteCcy"] .as_str() - .ok_or_else(|| Error::from(ParseError::missing_field("quoteCcy")))? - .to_string(); + .filter(|s| !s.is_empty()) + .map(ToString::to_string) + .or_else(|| quote_from_id.map(ToString::to_string)) + .ok_or_else(|| Error::from(ParseError::missing_field("quoteCcy")))?; // Instrument type let inst_type = data["instType"].as_str().unwrap_or("SPOT"); @@ -873,6 +885,30 @@ mod tests { use rust_decimal_macros::dec; use serde_json::json; + #[test] + fn test_parse_market_swap_empty_base_quote() { + let data = json!({ + "instId": "BTC-USDT-SWAP", + "instType": "SWAP", + "baseCcy": "", + "quoteCcy": "", + "settleCcy": "USDT", + "state": "live", + "tickSz": "0.1", + "lotSz": "1", + "minSz": "1", + "ctVal": "100" + }); + + // This should now correctly infer base/quote from instId + let market = parse_market(&data).unwrap(); + + // Assertions for FIXED behavior + assert_eq!(market.base, "BTC"); + assert_eq!(market.quote, "USDT"); + assert_eq!(market.symbol, "BTC/USDT:USDT"); + } + #[test] fn test_parse_market() { let data = json!({ diff --git a/ccxt-exchanges/src/okx/ws.rs b/ccxt-exchanges/src/okx/ws.rs index 6f3f8b1..df71f0a 100644 --- a/ccxt-exchanges/src/okx/ws.rs +++ b/ccxt-exchanges/src/okx/ws.rs @@ -192,7 +192,12 @@ impl OkxWs { if channel == Some("tickers") { if let Some(id) = inst_id { - if symbols_owned.iter().any(|s| s == id) { + let has_data = msg + .get("data") + .and_then(|d| d.as_array()) + .and_then(|arr| arr.first()) + .is_some(); + if has_data && symbols_owned.iter().any(|s| s == id) { match parse_ws_ticker(&msg, None) { Ok(ticker) => { if tx.send(Ok(vec![ticker])).is_err() { @@ -819,7 +824,12 @@ fn is_ticker_message(msg: &Value, symbol: &str) -> bool { if let Some(arg) = msg.get("arg") { let channel = arg.get("channel").and_then(|c| c.as_str()); let inst_id = arg.get("instId").and_then(|i| i.as_str()); - channel == Some("tickers") && inst_id == Some(symbol) + let has_data = msg + .get("data") + .and_then(|d| d.as_array()) + .and_then(|arr| arr.first()) + .is_some(); + channel == Some("tickers") && inst_id == Some(symbol) && has_data } else { false } @@ -832,9 +842,13 @@ fn is_orderbook_message(msg: &Value, symbol: &str) -> bool { if let Some(arg) = msg.get("arg") { let channel = arg.get("channel").and_then(|c| c.as_str()); let inst_id = arg.get("instId").and_then(|i| i.as_str()); - // Check if channel starts with "books" and instId matches + let has_data = msg + .get("data") + .and_then(|d| d.as_array()) + .and_then(|arr| arr.first()) + .is_some(); if let (Some(ch), Some(id)) = (channel, inst_id) { - ch.starts_with("books") && id == symbol + ch.starts_with("books") && id == symbol && has_data } else { false } @@ -850,7 +864,12 @@ fn is_trade_message(msg: &Value, symbol: &str) -> bool { if let Some(arg) = msg.get("arg") { let channel = arg.get("channel").and_then(|c| c.as_str()); let inst_id = arg.get("instId").and_then(|i| i.as_str()); - channel == Some("trades") && inst_id == Some(symbol) + let has_data = msg + .get("data") + .and_then(|d| d.as_array()) + .and_then(|arr| arr.first()) + .is_some(); + channel == Some("trades") && inst_id == Some(symbol) && has_data } else { false } diff --git a/examples/okx_account_example.rs b/examples/okx_account_example.rs new file mode 100644 index 0000000..f2fe84c --- /dev/null +++ b/examples/okx_account_example.rs @@ -0,0 +1,105 @@ +//! OKX Account Management Example +//! +//! Demonstrates account management operations using ccxt-rust: +//! 1. Query account balance +//! 2. Query trade history +//! +//! # Usage +//! +//! ```bash +//! # Set environment variables +//! export OKX_API_KEY="your_api_key" +//! export OKX_SECRET="your_secret" +//! export OKX_PASSPHRASE="your_passphrase" +//! +//! # Run the example +//! cargo run --example okx_account_example +//! ``` + +// Allow clippy warnings for example code +#![allow(clippy::disallowed_methods)] + +use ccxt_exchanges::okx::OkxBuilder; +use dotenvy::dotenv; +use std::env; + +#[tokio::main] +async fn main() -> std::result::Result<(), Box> { + dotenv().ok(); + println!("=== OKX Account Management Example ===\n"); + + let api_key = env::var("OKX_API_KEY").unwrap_or_default(); + let secret = env::var("OKX_SECRET").unwrap_or_default(); + let passphrase = env::var("OKX_PASSPHRASE").unwrap_or_default(); + + if api_key.is_empty() || secret.is_empty() || passphrase.is_empty() { + println!("⚠️ Skipping example: OKX_API_KEY, OKX_SECRET, or OKX_PASSPHRASE not set"); + println!(" Please set these environment variables to run this example."); + return Ok(()); + } + + // Initialize authenticated OKX exchange + let exchange = OkxBuilder::new() + .api_key(&api_key) + .secret(&secret) + .passphrase(&passphrase) + .build()?; + + // 1. Fetch Account Balance + println!("1. 【fetch_balance】Get account balance"); + match exchange.fetch_balance().await { + Ok(balance) => { + println!(" ✓ Successfully fetched balance"); + + println!(" Non-zero balances:"); + let mut found_any = false; + for (currency, entry) in balance.balances.iter() { + if entry.total > rust_decimal::Decimal::ZERO { + println!( + " - {}: Total={}, Free={}, Used={}", + currency, entry.total, entry.free, entry.used + ); + found_any = true; + } + } + if !found_any { + println!(" (No non-zero balances found)"); + } + } + Err(e) => println!(" ✗ Error: {}", e), + } + println!(); + + // 2. Fetch My Recent Trades + println!("2. 【fetch_my_trades】Get my recent trades (BTC/USDT)"); + // Note: We need to load markets first for symbol resolution + if let Err(e) = exchange.load_markets(false).await { + println!(" ⚠️ Failed to load markets: {}", e); + } else { + match exchange.fetch_my_trades("BTC/USDT", None, Some(5)).await { + Ok(trades) => { + if trades.is_empty() { + println!(" ℹ No recent trades found"); + } else { + println!(" ✓ Found {} recent trades:", trades.len()); + for (i, trade) in trades.iter().enumerate() { + println!( + " {}. {:?} {} @ {} ({})", + i + 1, + trade.side, + trade.amount, + trade.price, + trade.datetime.as_deref().unwrap_or("N/A") + ); + } + } + } + Err(e) => println!(" ✗ Error: {}", e), + } + } + println!(); + + println!("=== Example Complete ==="); + + Ok(()) +} diff --git a/examples/okx_example.rs b/examples/okx_example.rs index ca8f314..309e8a9 100644 --- a/examples/okx_example.rs +++ b/examples/okx_example.rs @@ -4,11 +4,12 @@ //! //! Run with: cargo run --example okx_example +use ccxt_core::DefaultType; use ccxt_core::exchange::Exchange; use ccxt_exchanges::okx::OkxBuilder; #[tokio::main] -async fn main() -> Result<(), Box> { +async fn main() -> std::result::Result<(), Box> { // Initialize logging tracing_subscriber::fmt::init(); @@ -16,7 +17,8 @@ async fn main() -> Result<(), Box> { // Create a public-only instance (no authentication) let exchange = OkxBuilder::new() - .sandbox(true) // Use demo environment for safety + .sandbox(false) // Use demo environment for safety + .default_type(DefaultType::Swap) .build()?; println!("Exchange: {} ({})", exchange.name(), exchange.id()); diff --git a/examples/okx_futures_example.rs b/examples/okx_futures_example.rs new file mode 100644 index 0000000..d980d2f --- /dev/null +++ b/examples/okx_futures_example.rs @@ -0,0 +1,99 @@ +//! OKX Futures (Swap) Example +//! +//! Demonstrates how to use OKX for perpetual swaps (futures): +//! 1. Initialize exchange with `DefaultType::Swap` +//! 2. Fetch swap markets +//! 3. Fetch swap ticker +//! 4. Fetch swap order book +//! +//! # Usage +//! +//! ```bash +//! cargo run --example okx_futures_example +//! ``` + +// Allow clippy warnings for example code +#![allow(clippy::disallowed_methods)] + +use ccxt_core::DefaultType; +use ccxt_exchanges::okx::OkxBuilder; + +#[tokio::main] +async fn main() -> std::result::Result<(), Box> { + println!("=== OKX Futures (Swap) Example ===\n"); + + // Initialize OKX exchange for SWAP (Perpetual Futures) + // This configures the exchange to default to 'SWAP' instrument type + let exchange = OkxBuilder::new().default_type(DefaultType::Swap).build()?; + + // 1. Fetch Markets (Swaps) + println!("1. 【fetch_markets】Get all swap markets"); + println!(" Fetching swap markets..."); + let markets = exchange.fetch_markets().await?; + println!(" ✓ Found {} swap markets", markets.len()); + + // Display sample swap markets (looking for USDT margined) + let swaps: Vec<_> = markets + .values() + .filter(|m| m.quote == "USDT" && m.linear.unwrap_or(false)) + .take(5) + .collect(); + + println!(" Sample USDT-margined Swaps:"); + for market in swaps { + println!( + " - {} ({}) Linear: {:?} Settle: {:?}", + market.symbol, market.id, market.linear, market.settle + ); + } + println!(); + + // Load markets for subsequent calls + exchange.load_markets(false).await?; + + // 2. Fetch Swap Ticker + // For OKX swaps, symbols are usually like "BTC/USDT:USDT" (unified) or "BTC-USDT-SWAP" (raw) + // The parser handles "BTC/USDT:USDT". + let symbol = "BTC/USDT:USDT"; + println!("2. 【fetch_ticker】Get {} ticker", symbol); + match exchange.fetch_ticker(symbol).await { + Ok(ticker) => { + println!(" ✓ Successfully fetched ticker"); + println!(" - Symbol: {}", ticker.symbol); + println!(" - Last Price: {:?}", ticker.last); + println!(" - Mark Price: {:?}", ticker.mark_price); // OKX tickers often include mark price + println!(" - Index Price: {:?}", ticker.index_price); + println!(" - Open Interest: {:?}", ticker.open_interest); + } + Err(e) => { + println!(" ✗ Error: {}", e); + // Fallback to simpler symbol if specific swap symbol fails (though load_markets should fix this) + println!(" Trying BTC/USDT..."); + match exchange.fetch_ticker("BTC/USDT").await { + Ok(ticker) => println!(" ✓ Fetched BTC/USDT: {:?}", ticker.last), + Err(e2) => println!(" ✗ Error: {}", e2), + } + } + } + println!(); + + // 3. Fetch Swap Order Book + println!("3. 【fetch_order_book】Get {} order book", symbol); + match exchange.fetch_order_book(symbol, Some(5)).await { + Ok(order_book) => { + println!(" ✓ Successfully fetched order book"); + if let Some(best_bid) = order_book.bids.first() { + println!(" - Best Bid: {} @ {}", best_bid.amount, best_bid.price); + } + if let Some(best_ask) = order_book.asks.first() { + println!(" - Best Ask: {} @ {}", best_ask.amount, best_ask.price); + } + } + Err(e) => println!(" ✗ Error: {}", e), + } + println!(); + + println!("=== Example Complete ==="); + + Ok(()) +} diff --git a/examples/okx_market_data_example.rs b/examples/okx_market_data_example.rs new file mode 100644 index 0000000..6caf05c --- /dev/null +++ b/examples/okx_market_data_example.rs @@ -0,0 +1,137 @@ +//! OKX Market Data API Example +//! +//! Demonstrates how to use OKX exchange market data APIs, including: +//! 1. `fetch_markets` - Get all supported markets +//! 2. `fetch_ticker` - Get specific ticker +//! 3. `fetch_tickers` - Get all tickers +//! 4. `fetch_order_book` - Get order book +//! 5. `fetch_trades` - Get recent trades +//! 6. `fetch_ohlcv` - Get candlestick data +//! +//! # Usage +//! +//! ```bash +//! cargo run --example okx_market_data_example +//! ``` + +// Allow clippy warnings for example code +#![allow(clippy::disallowed_methods)] + +use ccxt_core::ExchangeConfig; +use ccxt_core::types::ohlcv_request::OhlcvRequest; +use ccxt_exchanges::okx::Okx; + +#[tokio::main] +async fn main() -> std::result::Result<(), Box> { + println!("=== OKX Market Data API Example ===\n"); + + // Initialize OKX exchange (public only, no credentials needed for market data) + let config = ExchangeConfig::default(); + let exchange = Okx::new(config)?; + + // 1. Fetch Markets + println!("1. 【fetch_markets】Get all supported markets"); + println!(" Fetching markets..."); + let markets = exchange.fetch_markets().await?; + println!(" ✓ Found {} markets", markets.len()); + + // Display some sample markets + println!(" Sample markets:"); + for market in markets.values().take(5) { + println!( + " - {} ({}) Type: {:?} Active: {}", + market.symbol, market.id, market.market_type, market.active + ); + } + println!(); + + // Load markets for subsequent calls (required for correct symbol resolution) + exchange.load_markets(false).await?; + + // 2. Fetch Ticker + println!("2. 【fetch_ticker】Get BTC/USDT ticker"); + match exchange.fetch_ticker("BTC/USDT").await { + Ok(ticker) => { + println!(" ✓ Successfully fetched BTC/USDT ticker"); + println!(" - Symbol: {}", ticker.symbol); + println!(" - Last Price: {:?}", ticker.last); + println!(" - High (24h): {:?}", ticker.high); + println!(" - Low (24h): {:?}", ticker.low); + println!(" - Volume (24h): {:?}", ticker.base_volume); + println!(" - Timestamp: {}", ticker.timestamp); + } + Err(e) => println!(" ✗ Error: {}", e), + } + println!(); + + // 3. Fetch Order Book + println!("3. 【fetch_order_book】Get BTC/USDT order book"); + match exchange.fetch_order_book("BTC/USDT", Some(5)).await { + Ok(order_book) => { + println!(" ✓ Successfully fetched order book"); + println!(" - Bids: {} levels", order_book.bids.len()); + println!(" - Asks: {} levels", order_book.asks.len()); + + if let Some(best_bid) = order_book.bids.first() { + println!(" - Best Bid: {} @ {}", best_bid.amount, best_bid.price); + } + if let Some(best_ask) = order_book.asks.first() { + println!(" - Best Ask: {} @ {}", best_ask.amount, best_ask.price); + } + } + Err(e) => println!(" ✗ Error: {}", e), + } + println!(); + + // 4. Fetch Recent Trades + println!("4. 【fetch_trades】Get BTC/USDT recent trades"); + match exchange.fetch_trades("BTC/USDT", Some(5)).await { + Ok(trades) => { + println!(" ✓ Successfully fetched {} trades", trades.len()); + for (i, trade) in trades.iter().take(5).enumerate() { + println!( + " {}. {:?} {} @ {} (Time: {})", + i + 1, + trade.side, + trade.amount, + trade.price, + trade.datetime.as_deref().unwrap_or("N/A") + ); + } + } + Err(e) => println!(" ✗ Error: {}", e), + } + println!(); + + // 5. Fetch OHLCV + println!("5. 【fetch_ohlcv】Get BTC/USDT 1h candles"); + let request = OhlcvRequest::builder() + .symbol("BTC/USDT") + .timeframe("1h") + .limit(5) + .build()?; + + match exchange.fetch_ohlcv_v2(request).await { + Ok(ohlcvs) => { + println!(" ✓ Successfully fetched {} candles", ohlcvs.len()); + for (i, ohlcv) in ohlcvs.iter().enumerate() { + println!( + " {}. Time: {}, Open: {}, High: {}, Low: {}, Close: {}, Vol: {}", + i + 1, + ohlcv.timestamp, + ohlcv.open, + ohlcv.high, + ohlcv.low, + ohlcv.close, + ohlcv.volume + ); + } + } + Err(e) => println!(" ✗ Error: {}", e), + } + println!(); + + println!("=== Example Complete ==="); + + Ok(()) +} diff --git a/examples/okx_ws_example.rs b/examples/okx_ws_example.rs new file mode 100644 index 0000000..9e29610 --- /dev/null +++ b/examples/okx_ws_example.rs @@ -0,0 +1,221 @@ +#![allow(clippy::disallowed_methods)] + +use anyhow::Result; +use ccxt_core::logging::{LogConfig, init_logging}; +use ccxt_core::prelude::Price; +use ccxt_exchanges::okx::Okx; +use ccxt_exchanges::prelude::*; +use futures::StreamExt; +use rust_decimal::Decimal; +use std::env; +use tokio::time::Duration; + +#[tokio::main] +async fn main() -> Result<()> { + println!("=== OKX WebSocket Example ===\n"); + let mut config = LogConfig::development(); + config.show_span_events = false; + init_logging(&config); + + let api_key = env::var("OKX_API_KEY").ok(); + let secret = env::var("OKX_SECRET").ok(); + let passphrase = env::var("OKX_PASSPHRASE").ok(); + + let mut builder = Okx::builder(); + if let (Some(k), Some(s), Some(p)) = (&api_key, &secret, &passphrase) { + builder = builder.api_key(k).secret(s).passphrase(p); + } + let exchange = anyhow::Context::context(builder.build(), "Failed to initialize OKX exchange")?; + + println!("Connecting to WebSocket..."); + exchange.ws_connect().await?; + println!("Connected!"); + + println!("1. 【watch_ticker】Monitor BTC/USDT"); + match exchange.watch_ticker("BTC/USDT").await { + Ok(mut stream) => { + println!(" ✓ Subscribed to BTC/USDT ticker"); + let mut count = 0; + while let Some(result) = stream.next().await { + match result { + Ok(ticker) => { + println!( + " Update {}: {} Last: {}", + count + 1, + ticker.symbol, + ticker.last.unwrap_or(Price(Decimal::ZERO)) + ); + count += 1; + if count >= 3 { + break; + } + } + Err(e) => { + eprintln!(" Error: {}", e); + break; + } + } + } + } + Err(e) => eprintln!(" ✗ Error: {}", e), + } + println!(); + + println!("2. 【watch_tickers】Monitor BTC/USDT and ETH/USDT"); + match exchange + .watch_tickers(&["BTC/USDT".to_string(), "ETH/USDT".to_string()]) + .await + { + Ok(mut stream) => { + println!(" ✓ Subscribed to multiple tickers"); + let mut count = 0; + while let Some(result) = stream.next().await { + match result { + Ok(tickers) => { + println!( + " Update {}: Received {} tickers", + count + 1, + tickers.len() + ); + for ticker in tickers { + println!( + " - {}: {}", + ticker.symbol, + ticker.last.unwrap_or(Price(Decimal::ZERO)) + ); + } + count += 1; + if count >= 3 { + break; + } + } + Err(e) => { + eprintln!(" Error: {}", e); + break; + } + } + } + } + Err(e) => eprintln!(" ✗ Error: {}", e), + } + println!(); + + println!("3. 【watch_order_book】Monitor BTC/USDT Order Book"); + match exchange.watch_order_book("BTC/USDT", Some(5)).await { + Ok(mut stream) => { + println!(" ✓ Subscribed to order book"); + let mut count = 0; + while let Some(result) = stream.next().await { + match result { + Ok(book) => { + println!( + " Update {}: Bids={} Asks={} ", + count + 1, + book.bids.len(), + book.asks.len() + ); + if let Some(best_bid) = book.bids.first() { + println!(" Best Bid: {} @ {}", best_bid.amount, best_bid.price); + } + count += 1; + if count >= 3 { + break; + } + } + Err(e) => { + eprintln!(" Error: {}", e); + break; + } + } + } + } + Err(e) => eprintln!(" ✗ Error: {}", e), + } + println!(); + + println!("4. 【watch_trades】Monitor BTC/USDT Trades"); + match exchange.watch_trades("BTC/USDT").await { + Ok(mut stream) => { + println!(" ✓ Subscribed to trades"); + let mut count = 0; + while let Some(result) = stream.next().await { + match result { + Ok(trades) => { + println!( + " Update {}: Received {} trades", + count + 1, + trades.len() + ); + if let Some(trade) = trades.first() { + println!( + " Latest: {:?} {} @ {}", + trade.side, trade.amount, trade.price + ); + } + count += 1; + if count >= 3 { + break; + } + } + Err(e) => { + eprintln!(" Error: {}", e); + break; + } + } + } + } + Err(e) => eprintln!(" ✗ Error: {}", e), + } + println!(); + + if api_key.is_some() { + println!("5. 【Private Channels】Monitor Balance and Orders"); + + println!(" Subscribing to balance updates..."); + match exchange.watch_balance().await { + Ok(mut stream) => { + println!(" ✓ Subscribed to balance"); + let timeout = tokio::time::timeout(Duration::from_secs(5), stream.next()).await; + match timeout { + Ok(Some(Ok(balance))) => { + println!( + " Received balance update. Total currencies: {}", + balance.balances.len() + ); + } + Ok(Some(Err(e))) => eprintln!(" Error: {}", e), + Ok(None) => println!(" Stream ended"), + Err(_) => println!(" Timeout waiting for balance update"), + } + } + Err(e) => eprintln!(" ✗ Error watching balance: {}", e), + } + + println!(" Subscribing to order updates..."); + match exchange.watch_orders(None).await { + Ok(mut stream) => { + println!(" ✓ Subscribed to orders"); + let timeout = tokio::time::timeout(Duration::from_secs(5), stream.next()).await; + match timeout { + Ok(Some(Ok(order))) => { + println!( + " Received order update: {} {:?}", + order.id, order.status + ); + } + Ok(Some(Err(e))) => eprintln!(" Error: {}", e), + Ok(None) => println!(" Stream ended"), + Err(_) => println!(" Timeout waiting for order update"), + } + } + Err(e) => eprintln!(" ✗ Error watching orders: {}", e), + } + } else { + println!("5. 【Private Channels】Skipped (requires API credentials)"); + } + + println!("=== Example Complete ==="); + exchange.ws_disconnect().await?; + + Ok(()) +}