fix(okx): harden market parsing and websocket stream handling

ccxt-exchanges/src/okx/parser.rs

@@ -42,16 +42,28 @@ pub fn parse_market(data: &Value) -> Result<Market> {
         .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!({

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
     }

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<dyn std::error::Error>> {
+    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(())
+}

examples/okx_example.rs

@@ -4,19 +4,21 @@
 //!
 //! 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<dyn std::error::Error>> {
+async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
     // Initialize logging
     tracing_subscriber::fmt::init();
 
     println!("=== OKX Exchange Example ===\n");
 
     // 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());

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<dyn std::error::Error>> {
+    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(())
+}