nautilus_okx/
data.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Live market data client implementation for the OKX adapter.
17
18use std::sync::{
19    Arc,
20    atomic::{AtomicBool, Ordering},
21};
22
23use ahash::{AHashMap, AHashSet};
24use anyhow::Context;
25use futures_util::{StreamExt, pin_mut};
26use nautilus_common::{
27    cache::quote::QuoteCache,
28    clients::DataClient,
29    live::{runner::get_data_event_sender, runtime::get_runtime},
30    messages::{
31        DataEvent,
32        data::{
33            BarsResponse, BookResponse, DataResponse, ForwardPricesResponse, FundingRatesResponse,
34            InstrumentResponse, InstrumentsResponse, RequestBars, RequestBookSnapshot,
35            RequestForwardPrices, RequestFundingRates, RequestInstrument, RequestInstruments,
36            RequestTrades, SubscribeBars, SubscribeBookDeltas, SubscribeFundingRates,
37            SubscribeIndexPrices, SubscribeInstrument, SubscribeInstrumentStatus,
38            SubscribeInstruments, SubscribeMarkPrices, SubscribeOptionGreeks, SubscribeQuotes,
39            SubscribeTrades, TradesResponse, UnsubscribeBars, UnsubscribeBookDeltas,
40            UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrument,
41            UnsubscribeInstrumentStatus, UnsubscribeMarkPrices, UnsubscribeOptionGreeks,
42            UnsubscribeQuotes, UnsubscribeTrades,
43        },
44    },
45};
46use nautilus_core::{
47    AtomicMap, Params, UnixNanos,
48    datetime::datetime_to_unix_nanos,
49    time::{AtomicTime, get_atomic_clock_realtime},
50};
51use nautilus_model::{
52    data::{Data, FundingRateUpdate, InstrumentStatus, OrderBookDeltas_API},
53    enums::{BookType, GreeksConvention, MarketStatusAction},
54    identifiers::{ClientId, InstrumentId, Venue},
55    instruments::{Instrument, InstrumentAny},
56};
57use tokio::{task::JoinHandle, time::Duration};
58use tokio_util::sync::CancellationToken;
59use ustr::Ustr;
60
61use crate::{
62    common::{
63        consts::{
64            OKX_VENUE, OKX_WS_HEARTBEAT_SECS, resolve_book_depth, resolve_instrument_families,
65            should_retry_error_code,
66        },
67        enums::{
68            OKXBookAction, OKXBookChannel, OKXContractType, OKXGreeksType, OKXInstrumentStatus,
69            OKXInstrumentType, OKXVipLevel,
70        },
71        parse::{
72            extract_inst_family, is_okx_spread_symbol, okx_instrument_type_from_symbol,
73            okx_status_to_market_action, parse_base_quote_from_symbol, parse_instrument_any,
74            parse_instrument_id, parse_millisecond_timestamp, parse_price, parse_quantity,
75        },
76    },
77    config::OKXDataClientConfig,
78    http::{
79        client::{OKXHttpClient, OKXInstrumentDefinitionError},
80        query::GetSpreadsParams,
81    },
82    websocket::{
83        client::OKXWebSocketClient,
84        enums::OKXWsChannel,
85        messages::{NautilusWsMessage, OKXBookMsg, OKXOptionSummaryMsg, OKXWsMessage},
86        parse::{
87            extract_fees_from_cached_instrument, parse_book_msg_vec, parse_index_price_msg_vec,
88            parse_option_summary_greeks, parse_ws_message_data,
89        },
90    },
91};
92
93/// Resolves the set of [`OKXGreeksType`] conventions for an option greeks subscription.
94///
95/// Reads the `greeks_convention` key from `params`, accepting either a single
96/// [`GreeksConvention`] string (e.g. `"BLACK_SCHOLES"` or `"PRICE_ADJUSTED"`) or a
97/// JSON array of such strings. Unrecognized entries log a warning and are skipped.
98/// Returns the default set `{Bs, Pa}` when the key is absent, unparsable, or
99/// yields no valid entries so every subscription defaults to both conventions.
100pub(crate) fn parse_greeks_conventions_from_params(
101    params: &Option<Params>,
102) -> AHashSet<OKXGreeksType> {
103    let default_set: AHashSet<OKXGreeksType> =
104        [OKXGreeksType::Bs, OKXGreeksType::Pa].into_iter().collect();
105
106    let Some(value) = params.as_ref().and_then(|p| p.get("greeks_convention")) else {
107        return default_set;
108    };
109
110    let mut out = AHashSet::new();
111    match value {
112        serde_json::Value::String(s) => push_convention_str(&mut out, s),
113        serde_json::Value::Array(items) => {
114            for item in items {
115                if let Some(s) = item.as_str() {
116                    push_convention_str(&mut out, s);
117                } else {
118                    log::warn!("Ignoring non-string greeks_convention entry {item:?}");
119                }
120            }
121        }
122        other => {
123            log::warn!(
124                "Unsupported greeks_convention value {other:?}, defaulting to both conventions"
125            );
126        }
127    }
128
129    if out.is_empty() { default_set } else { out }
130}
131
132fn push_convention_str(out: &mut AHashSet<OKXGreeksType>, raw: &str) {
133    match raw.parse::<GreeksConvention>() {
134        Ok(convention) => {
135            out.insert(convention.into());
136        }
137        Err(_) => log::warn!("Unrecognized greeks_convention {raw:?}, skipping"),
138    }
139}
140
141#[derive(Debug)]
142pub struct OKXDataClient {
143    client_id: ClientId,
144    config: OKXDataClientConfig,
145    http_client: OKXHttpClient,
146    ws_public: Option<OKXWebSocketClient>,
147    ws_business: Option<OKXWebSocketClient>,
148    is_connected: AtomicBool,
149    cancellation_token: CancellationToken,
150    tasks: Vec<JoinHandle<()>>,
151    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
152    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
153    book_channels: Arc<AtomicMap<InstrumentId, OKXBookChannel>>,
154    index_ticker_map: Arc<AtomicMap<Ustr, AHashSet<Ustr>>>,
155    option_greeks_subs: Arc<AtomicMap<InstrumentId, AHashSet<OKXGreeksType>>>,
156    // `Mutex<AHashMap>` so the spawned subscribe task can roll back the
157    // refcount on failure. A bare `AHashMap` would leave the count
158    // permanently incremented and wedge future Greeks subscribes.
159    option_summary_family_subs: Arc<std::sync::Mutex<AHashMap<Ustr, usize>>>,
160    clock: &'static AtomicTime,
161}
162
163impl OKXDataClient {
164    /// Creates a new [`OKXDataClient`] instance.
165    ///
166    /// # Errors
167    ///
168    /// Returns an error if the client fails to initialize.
169    pub fn new(client_id: ClientId, config: OKXDataClientConfig) -> anyhow::Result<Self> {
170        let clock = get_atomic_clock_realtime();
171        let data_sender = get_data_event_sender();
172
173        let http_client = if config.has_api_credentials() {
174            OKXHttpClient::with_credentials(
175                config.api_key.clone(),
176                config.api_secret.clone(),
177                config.api_passphrase.clone(),
178                Some(config.http_base_url()),
179                config.http_timeout_secs,
180                config.max_retries,
181                config.retry_delay_initial_ms,
182                config.retry_delay_max_ms,
183                config.environment,
184                config.proxy_url.clone(),
185            )?
186        } else {
187            OKXHttpClient::new(
188                Some(config.http_base_url()),
189                config.http_timeout_secs,
190                config.max_retries,
191                config.retry_delay_initial_ms,
192                config.retry_delay_max_ms,
193                config.environment,
194                config.proxy_url.clone(),
195            )?
196        };
197
198        let ws_public = OKXWebSocketClient::new(
199            Some(config.ws_public_url()),
200            None,
201            None,
202            None,
203            None,
204            Some(OKX_WS_HEARTBEAT_SECS),
205            None,
206            config.transport_backend,
207            config.proxy_url.clone(),
208        )
209        .context("failed to construct OKX public websocket client")?;
210
211        let ws_business = if config.requires_business_ws() {
212            let ws = OKXWebSocketClient::new(
213                Some(config.ws_business_url()),
214                None, // No auth needed for public business channels
215                None,
216                None,
217                None,
218                Some(OKX_WS_HEARTBEAT_SECS),
219                None,
220                config.transport_backend,
221                config.proxy_url.clone(),
222            )
223            .context("failed to construct OKX business websocket client")?;
224            Some(ws)
225        } else {
226            None
227        };
228
229        if let Some(vip_level) = config.vip_level {
230            ws_public.set_vip_level(vip_level);
231
232            if let Some(ref ws) = ws_business {
233                ws.set_vip_level(vip_level);
234            }
235        }
236
237        Ok(Self {
238            client_id,
239            config,
240            http_client,
241            ws_public: Some(ws_public),
242            ws_business,
243            is_connected: AtomicBool::new(false),
244            cancellation_token: CancellationToken::new(),
245            tasks: Vec::new(),
246            data_sender,
247            instruments: Arc::new(AtomicMap::new()),
248            book_channels: Arc::new(AtomicMap::new()),
249            index_ticker_map: Arc::new(AtomicMap::new()),
250            option_greeks_subs: Arc::new(AtomicMap::new()),
251            option_summary_family_subs: Arc::new(std::sync::Mutex::new(AHashMap::new())),
252            clock,
253        })
254    }
255
256    fn venue(&self) -> Venue {
257        *OKX_VENUE
258    }
259
260    fn vip_level(&self) -> Option<OKXVipLevel> {
261        self.ws_public.as_ref().map(|ws| ws.vip_level())
262    }
263
264    fn public_ws(&self) -> anyhow::Result<&OKXWebSocketClient> {
265        self.ws_public
266            .as_ref()
267            .context("public websocket client not initialized")
268    }
269
270    fn business_ws(&self) -> anyhow::Result<&OKXWebSocketClient> {
271        self.ws_business
272            .as_ref()
273            .context("business websocket client not available (credentials required)")
274    }
275
276    fn send_data(sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>, data: Data) {
277        if let Err(e) = sender.send(DataEvent::Data(data)) {
278            log::error!("Failed to emit data event: {e}");
279        }
280    }
281
282    fn spawn_ws<F>(&self, fut: F, context: &'static str)
283    where
284        F: Future<Output = anyhow::Result<()>> + Send + 'static,
285    {
286        get_runtime().spawn(async move {
287            if let Err(e) = fut.await {
288                log::error!("{context}: {e:?}");
289            }
290        });
291    }
292
293    #[expect(clippy::too_many_arguments)]
294    fn handle_ws_message(
295        message: OKXWsMessage,
296        data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
297        instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
298        instruments_by_symbol: &mut AHashMap<Ustr, InstrumentAny>,
299        quote_cache: &mut QuoteCache,
300        funding_cache: &mut AHashMap<Ustr, (Ustr, u64)>,
301        index_ticker_map: &Arc<AtomicMap<Ustr, AHashSet<Ustr>>>,
302        option_greeks_subs: &Arc<AtomicMap<InstrumentId, AHashSet<OKXGreeksType>>>,
303        clock: &AtomicTime,
304    ) {
305        match message {
306            OKXWsMessage::BookData { arg, action, data } => {
307                let Some(inst_id) = arg.inst_id else {
308                    log::warn!("Book data without inst_id");
309                    return;
310                };
311                let Some(instrument) = instruments_by_symbol.get(&inst_id) else {
312                    log::warn!("No cached instrument for book data: {inst_id}");
313                    return;
314                };
315                let ts_init = clock.get_time_ns();
316
317                match parse_book_msg_vec(
318                    data,
319                    &instrument.id(),
320                    instrument.price_precision(),
321                    instrument.size_precision(),
322                    action,
323                    ts_init,
324                ) {
325                    Ok(data_vec) => {
326                        for data in data_vec {
327                            Self::send_data(data_sender, data);
328                        }
329                    }
330                    Err(e) => log::error!("Failed to parse book data: {e}"),
331                }
332            }
333            OKXWsMessage::ChannelData {
334                channel,
335                inst_id,
336                data,
337            } => {
338                // Option summary subscriptions use instFamily (not instId), so
339                // the arg has inst_id: None. Each element in the data array carries
340                // its own inst_id that we resolve per-message.
341                if matches!(channel, OKXWsChannel::OptionSummary) {
342                    let ts_init = clock.get_time_ns();
343
344                    match serde_json::from_value::<Vec<OKXOptionSummaryMsg>>(data) {
345                        Ok(msgs) => {
346                            let subs = option_greeks_subs.load();
347
348                            for msg in &msgs {
349                                let Some(instrument) = instruments_by_symbol.get(&msg.inst_id)
350                                else {
351                                    continue;
352                                };
353                                let instrument_id = instrument.id();
354                                let Some(conventions) = subs.get(&instrument_id) else {
355                                    continue;
356                                };
357
358                                for greeks_type in conventions {
359                                    match parse_option_summary_greeks(
360                                        msg,
361                                        &instrument_id,
362                                        *greeks_type,
363                                        ts_init,
364                                    ) {
365                                        Ok(greeks) => {
366                                            if let Err(e) =
367                                                data_sender.send(DataEvent::OptionGreeks(greeks))
368                                            {
369                                                log::error!(
370                                                    "Failed to emit option greeks event: {e}"
371                                                );
372                                            }
373                                        }
374                                        Err(e) => {
375                                            log::error!(
376                                                "Failed to parse option summary for {} ({greeks_type:?}): {e}",
377                                                msg.inst_id
378                                            );
379                                        }
380                                    }
381                                }
382                            }
383                        }
384                        Err(e) => {
385                            log::error!("Failed to deserialize option summary data: {e}");
386                        }
387                    }
388                    return;
389                }
390
391                let Some(inst_id) = inst_id else {
392                    log::debug!("Channel data without inst_id: {channel:?}");
393                    return;
394                };
395
396                // Index tickers use base pair format (e.g., "BTC-USDT") but instruments
397                // are keyed by full symbol (e.g., "BTC-USDT-SWAP"). Dispatch index price
398                // updates only to instruments that subscribed via subscribe_index_prices.
399                if matches!(channel, OKXWsChannel::IndexTickers) {
400                    let ts_init = clock.get_time_ns();
401                    let map_guard = index_ticker_map.load();
402                    let Some(subscribed_symbols) = map_guard.get(&inst_id) else {
403                        log::debug!("No subscribed instruments for index ticker: {inst_id}");
404                        return;
405                    };
406                    let symbols: Vec<Ustr> = subscribed_symbols.iter().copied().collect();
407                    drop(map_guard);
408
409                    for sym in &symbols {
410                        let Some(instrument) = instruments_by_symbol.get(sym) else {
411                            log::warn!("No cached instrument for index ticker symbol: {sym}");
412                            continue;
413                        };
414
415                        match parse_index_price_msg_vec(
416                            data.clone(),
417                            &instrument.id(),
418                            instrument.price_precision(),
419                            ts_init,
420                        ) {
421                            Ok(data_vec) => {
422                                for d in data_vec {
423                                    Self::send_data(data_sender, d);
424                                }
425                            }
426                            Err(e) => log::error!("Failed to parse index price data: {e}"),
427                        }
428                    }
429                    return;
430                }
431
432                let Some(instrument) = instruments_by_symbol.get(&inst_id) else {
433                    log::warn!("No cached instrument for {channel:?}: {inst_id}");
434                    return;
435                };
436                let instrument_id = instrument.id();
437                let price_precision = instrument.price_precision();
438                let size_precision = instrument.size_precision();
439                let ts_init = clock.get_time_ns();
440
441                if matches!(channel, OKXWsChannel::SprdBooks5) {
442                    let msgs: Vec<OKXBookMsg> = match serde_json::from_value(data) {
443                        Ok(m) => m,
444                        Err(e) => {
445                            log::error!("Failed to deserialize spread book data: {e}");
446                            return;
447                        }
448                    };
449
450                    // sprd-books5 pushes a full 5-level snapshot each message.
451                    match parse_book_msg_vec(
452                        msgs,
453                        &instrument_id,
454                        price_precision,
455                        size_precision,
456                        OKXBookAction::Snapshot,
457                        ts_init,
458                    ) {
459                        Ok(data_vec) => {
460                            for d in data_vec {
461                                Self::send_data(data_sender, d);
462                            }
463                        }
464                        Err(e) => log::error!("Failed to parse spread book data: {e}"),
465                    }
466
467                    return;
468                }
469
470                if matches!(channel, OKXWsChannel::BboTbt | OKXWsChannel::SprdBboTbt) {
471                    let msgs: Vec<OKXBookMsg> = match serde_json::from_value(data) {
472                        Ok(m) => m,
473                        Err(e) => {
474                            log::error!("Failed to deserialize BboTbt data: {e}");
475                            return;
476                        }
477                    };
478
479                    for msg in &msgs {
480                        let bid = msg.bids.first();
481                        let ask = msg.asks.first();
482                        let bid_price =
483                            bid.and_then(|e| parse_price(&e.price, price_precision).ok());
484                        let bid_size =
485                            bid.and_then(|e| parse_quantity(&e.size, size_precision).ok());
486                        let ask_price =
487                            ask.and_then(|e| parse_price(&e.price, price_precision).ok());
488                        let ask_size =
489                            ask.and_then(|e| parse_quantity(&e.size, size_precision).ok());
490                        let ts_event = parse_millisecond_timestamp(msg.ts);
491
492                        match quote_cache.process(
493                            instrument_id,
494                            bid_price,
495                            ask_price,
496                            bid_size,
497                            ask_size,
498                            ts_event,
499                            ts_init,
500                        ) {
501                            Ok(quote) => Self::send_data(data_sender, Data::Quote(quote)),
502                            Err(e) => {
503                                log::debug!("Skipping partial BboTbt for {instrument_id}: {e}");
504                            }
505                        }
506                    }
507
508                    return;
509                }
510
511                match parse_ws_message_data(
512                    &channel,
513                    data,
514                    &instrument_id,
515                    price_precision,
516                    size_precision,
517                    ts_init,
518                    funding_cache,
519                    instruments_by_symbol,
520                ) {
521                    Ok(Some(ws_msg)) => {
522                        dispatch_parsed_data(
523                            ws_msg,
524                            data_sender,
525                            instruments,
526                            instruments_by_symbol,
527                        );
528                    }
529                    Ok(None) => {}
530                    Err(e) => log::error!("Failed to parse {channel:?} data: {e}"),
531                }
532            }
533            OKXWsMessage::Instruments(okx_instruments) => {
534                let ts_init = clock.get_time_ns();
535
536                for okx_inst in okx_instruments {
537                    let inst_key = Ustr::from(&okx_inst.inst_id);
538                    let (margin_init, margin_maint, maker_fee, taker_fee) =
539                        instruments_by_symbol.get(&inst_key).map_or(
540                            (None, None, None, None),
541                            extract_fees_from_cached_instrument,
542                        );
543                    let status_action = okx_status_to_market_action(okx_inst.state);
544                    let is_live = matches!(okx_inst.state, OKXInstrumentStatus::Live);
545                    match parse_instrument_any(
546                        &okx_inst,
547                        margin_init,
548                        margin_maint,
549                        maker_fee,
550                        taker_fee,
551                        ts_init,
552                    ) {
553                        Ok(Some(inst_any)) => {
554                            let instrument_id = inst_any.id();
555                            instruments_by_symbol
556                                .insert(inst_any.symbol().inner(), inst_any.clone());
557                            upsert_instrument(instruments, inst_any);
558                            emit_instrument_status(
559                                data_sender,
560                                instrument_id,
561                                status_action,
562                                is_live,
563                                ts_init,
564                            );
565                        }
566                        Ok(None) => {
567                            let instrument_id = instruments_by_symbol
568                                .get(&inst_key)
569                                .map_or_else(|| parse_instrument_id(inst_key), |i| i.id());
570                            emit_instrument_status(
571                                data_sender,
572                                instrument_id,
573                                status_action,
574                                is_live,
575                                ts_init,
576                            );
577                        }
578                        Err(e) => {
579                            log::warn!("Failed to parse instrument {}: {e}", okx_inst.inst_id);
580                            let instrument_id = instruments_by_symbol
581                                .get(&inst_key)
582                                .map_or_else(|| parse_instrument_id(inst_key), |i| i.id());
583                            emit_instrument_status(
584                                data_sender,
585                                instrument_id,
586                                status_action,
587                                is_live,
588                                ts_init,
589                            );
590                        }
591                    }
592                }
593            }
594            OKXWsMessage::Orders(_)
595            | OKXWsMessage::SpreadOrders(_)
596            | OKXWsMessage::AlgoOrders(_)
597            | OKXWsMessage::OrderResponse { .. }
598            | OKXWsMessage::Account(_)
599            | OKXWsMessage::Positions(_)
600            | OKXWsMessage::SendFailed { .. } => {
601                log::debug!("Ignoring execution message on data client");
602            }
603            OKXWsMessage::Error(e) => {
604                if should_retry_error_code(&e.code) {
605                    log::warn!("OKX websocket error: {e:?}");
606                } else {
607                    log::error!("OKX websocket error: {e:?}");
608                }
609            }
610            OKXWsMessage::Reconnected => {
611                log::info!("Websocket reconnected");
612            }
613            OKXWsMessage::Authenticated => {
614                log::debug!("Websocket authenticated");
615            }
616        }
617    }
618}
619
620fn dispatch_parsed_data(
621    msg: NautilusWsMessage,
622    data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
623    instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
624    instruments_by_symbol: &mut AHashMap<Ustr, InstrumentAny>,
625) {
626    match msg {
627        NautilusWsMessage::Data(payloads) => {
628            for data in payloads {
629                if let Err(e) = data_sender.send(DataEvent::Data(data)) {
630                    log::error!("Failed to emit data event: {e}");
631                }
632            }
633        }
634        NautilusWsMessage::Deltas(deltas) => {
635            let data = Data::Deltas(OrderBookDeltas_API::new(deltas));
636            if let Err(e) = data_sender.send(DataEvent::Data(data)) {
637                log::error!("Failed to emit data event: {e}");
638            }
639        }
640        NautilusWsMessage::FundingRates(updates) => {
641            emit_funding_rates(data_sender, updates);
642        }
643        NautilusWsMessage::Instrument(instrument, status) => {
644            instruments_by_symbol.insert(instrument.symbol().inner(), (*instrument).clone());
645            upsert_instrument(instruments, *instrument);
646
647            if let Some(status) = status
648                && let Err(e) = data_sender.send(DataEvent::InstrumentStatus(status))
649            {
650                log::error!("Failed to emit instrument status event: {e}");
651            }
652        }
653        NautilusWsMessage::InstrumentStatus(status) => {
654            if let Err(e) = data_sender.send(DataEvent::InstrumentStatus(status)) {
655                log::error!("Failed to emit instrument status event: {e}");
656            }
657        }
658        _ => {}
659    }
660}
661
662fn emit_funding_rates(
663    sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
664    updates: Vec<FundingRateUpdate>,
665) {
666    for update in updates {
667        if let Err(e) = sender.send(DataEvent::FundingRate(update)) {
668            log::error!("Failed to emit funding rate event: {e}");
669        }
670    }
671}
672
673fn emit_instrument_status(
674    sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
675    instrument_id: InstrumentId,
676    status_action: MarketStatusAction,
677    is_live: bool,
678    ts_init: UnixNanos,
679) {
680    let status = InstrumentStatus::new(
681        instrument_id,
682        status_action,
683        ts_init,
684        ts_init,
685        None,
686        None,
687        Some(is_live),
688        None,
689        None,
690    );
691
692    if let Err(e) = sender.send(DataEvent::InstrumentStatus(status)) {
693        log::error!("Failed to emit instrument status event: {e}");
694    }
695}
696
697fn upsert_instrument(
698    cache: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
699    instrument: InstrumentAny,
700) {
701    cache.insert(instrument.id(), instrument);
702}
703
704fn contract_filter_with_config(config: &OKXDataClientConfig, instrument: &InstrumentAny) -> bool {
705    contract_filter_with_config_types(config.contract_types.as_ref(), instrument)
706}
707
708fn contract_filter_with_config_types(
709    contract_types: Option<&Vec<OKXContractType>>,
710    instrument: &InstrumentAny,
711) -> bool {
712    match contract_types {
713        None => true,
714        Some(filter) if filter.is_empty() => true,
715        Some(filter) => {
716            let is_inverse = instrument.is_inverse();
717            (is_inverse && filter.contains(&OKXContractType::Inverse))
718                || (!is_inverse && filter.contains(&OKXContractType::Linear))
719        }
720    }
721}
722
723#[async_trait::async_trait(?Send)]
724impl DataClient for OKXDataClient {
725    fn client_id(&self) -> ClientId {
726        self.client_id
727    }
728
729    fn venue(&self) -> Option<Venue> {
730        Some(self.venue())
731    }
732
733    fn start(&mut self) -> anyhow::Result<()> {
734        log::info!(
735            "Started: client_id={}, vip_level={:?}, instrument_types={:?}, environment={}, proxy_url={:?}",
736            self.client_id,
737            self.vip_level(),
738            self.config.instrument_types,
739            self.config.environment,
740            self.config.proxy_url,
741        );
742        Ok(())
743    }
744
745    fn stop(&mut self) -> anyhow::Result<()> {
746        log::info!("Stopping {id}", id = self.client_id);
747        self.cancellation_token.cancel();
748        self.is_connected.store(false, Ordering::Relaxed);
749        Ok(())
750    }
751
752    fn reset(&mut self) -> anyhow::Result<()> {
753        log::debug!("Resetting {id}", id = self.client_id);
754        self.is_connected.store(false, Ordering::Relaxed);
755        self.cancellation_token = CancellationToken::new();
756        self.tasks.clear();
757        self.book_channels.store(AHashMap::new());
758        self.option_greeks_subs
759            .store(AHashMap::<InstrumentId, AHashSet<OKXGreeksType>>::new());
760        self.option_summary_family_subs
761            .lock()
762            .expect("option_summary_family_subs mutex poisoned")
763            .clear();
764        Ok(())
765    }
766
767    fn dispose(&mut self) -> anyhow::Result<()> {
768        log::debug!("Disposing {id}", id = self.client_id);
769        self.stop()
770    }
771
772    async fn connect(&mut self) -> anyhow::Result<()> {
773        if self.is_connected() {
774            return Ok(());
775        }
776
777        // Create fresh token so tasks from a previous connection cycle are not
778        // immediately cancelled (the old token may already be in cancelled state)
779        self.cancellation_token = CancellationToken::new();
780
781        let instrument_types = if self.config.instrument_types.is_empty() {
782            vec![OKXInstrumentType::Spot]
783        } else {
784            self.config.instrument_types.clone()
785        };
786
787        let mut all_instruments = Vec::new();
788
789        for inst_type in &instrument_types {
790            let Some(families) =
791                resolve_instrument_families(&self.config.instrument_families, *inst_type)
792            else {
793                continue;
794            };
795
796            if families.is_empty() {
797                let (mut fetched, _inst_id_codes) = self
798                    .http_client
799                    .request_instruments(*inst_type, None)
800                    .await
801                    .with_context(|| {
802                        format!("failed to request OKX instruments for {inst_type:?}")
803                    })?;
804
805                fetched.retain(|instrument| contract_filter_with_config(&self.config, instrument));
806                self.http_client.cache_instruments(&fetched);
807
808                self.instruments.rcu(|m| {
809                    for instrument in &fetched {
810                        m.insert(instrument.id(), instrument.clone());
811                    }
812                });
813
814                all_instruments.extend(fetched);
815            } else {
816                for family in &families {
817                    let (mut fetched, _inst_id_codes) = self
818                        .http_client
819                        .request_instruments(*inst_type, Some(family.clone()))
820                        .await
821                        .with_context(|| {
822                            format!(
823                                "failed to request OKX instruments for {inst_type:?} family {family}"
824                            )
825                        })?;
826
827                    fetched
828                        .retain(|instrument| contract_filter_with_config(&self.config, instrument));
829                    self.http_client.cache_instruments(&fetched);
830
831                    self.instruments.rcu(|m| {
832                        for instrument in &fetched {
833                            m.insert(instrument.id(), instrument.clone());
834                        }
835                    });
836
837                    all_instruments.extend(fetched);
838                }
839            }
840        }
841
842        if self.config.load_spreads {
843            match self
844                .http_client
845                .request_spread_instruments(GetSpreadsParams {
846                    state: Some("live".to_string()),
847                    ..Default::default()
848                })
849                .await
850            {
851                Ok(mut fetched) => {
852                    fetched
853                        .retain(|instrument| contract_filter_with_config(&self.config, instrument));
854                    self.http_client.cache_instruments(&fetched);
855
856                    self.instruments.rcu(|m| {
857                        for instrument in &fetched {
858                            m.insert(instrument.id(), instrument.clone());
859                        }
860                    });
861
862                    all_instruments.extend(fetched);
863                }
864                Err(e) => {
865                    log::error!("Failed to fetch OKX spread instruments: {e:?}");
866                }
867            }
868        }
869
870        for instrument in all_instruments {
871            if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
872                log::warn!("Failed to send instrument: {e}");
873            }
874        }
875
876        if let Some(ref mut ws) = self.ws_public {
877            // Cache instruments to websocket before connecting so handler has them
878            let instruments: Vec<_> = self.instruments.load().values().cloned().collect();
879            ws.cache_instruments(&instruments);
880
881            ws.connect()
882                .await
883                .context("failed to connect OKX public websocket")?;
884            ws.wait_until_active(10.0)
885                .await
886                .context("public websocket did not become active")?;
887
888            let stream = ws.stream();
889            let sender = self.data_sender.clone();
890            let insts = self.instruments.clone();
891            let idx_map = self.index_ticker_map.clone();
892            let greeks_subs = self.option_greeks_subs.clone();
893            let cancel = self.cancellation_token.clone();
894            let clock = self.clock;
895
896            let handle = get_runtime().spawn(async move {
897                let mut instruments_by_symbol: AHashMap<Ustr, InstrumentAny> = insts
898                    .load()
899                    .values()
900                    .map(|i| (i.symbol().inner(), i.clone()))
901                    .collect();
902                let mut quote_cache = QuoteCache::new();
903                let mut funding_cache: AHashMap<Ustr, (Ustr, u64)> = AHashMap::new();
904                pin_mut!(stream);
905
906                loop {
907                    tokio::select! {
908                        Some(message) = stream.next() => {
909                            Self::handle_ws_message(
910                                message,
911                                &sender,
912                                &insts,
913                                &mut instruments_by_symbol,
914                                &mut quote_cache,
915                                &mut funding_cache,
916                                &idx_map,
917                                &greeks_subs,
918                                clock,
919                            );
920                        }
921                        () = cancel.cancelled() => {
922                            log::debug!("Public websocket stream task cancelled");
923                            break;
924                        }
925                    }
926                }
927            });
928            self.tasks.push(handle);
929
930            for inst_type in &instrument_types {
931                ws.subscribe_instruments(*inst_type)
932                    .await
933                    .with_context(|| {
934                        format!("failed to subscribe to instrument type {inst_type:?}")
935                    })?;
936            }
937        }
938
939        if let Some(ref mut ws) = self.ws_business {
940            // Cache instruments to websocket before connecting so handler has them
941            let instruments: Vec<_> = self.instruments.load().values().cloned().collect();
942            ws.cache_instruments(&instruments);
943
944            ws.connect()
945                .await
946                .context("failed to connect OKX business websocket")?;
947            ws.wait_until_active(10.0)
948                .await
949                .context("business websocket did not become active")?;
950
951            let stream = ws.stream();
952            let sender = self.data_sender.clone();
953            let insts = self.instruments.clone();
954            let idx_map = self.index_ticker_map.clone();
955            let greeks_subs = self.option_greeks_subs.clone();
956            let cancel = self.cancellation_token.clone();
957            let clock = self.clock;
958
959            let handle = get_runtime().spawn(async move {
960                let mut instruments_by_symbol: AHashMap<Ustr, InstrumentAny> = insts
961                    .load()
962                    .values()
963                    .map(|i| (i.symbol().inner(), i.clone()))
964                    .collect();
965                let mut quote_cache = QuoteCache::new();
966                let mut funding_cache: AHashMap<Ustr, (Ustr, u64)> = AHashMap::new();
967                pin_mut!(stream);
968
969                loop {
970                    tokio::select! {
971                        Some(message) = stream.next() => {
972                            Self::handle_ws_message(
973                                message,
974                                &sender,
975                                &insts,
976                                &mut instruments_by_symbol,
977                                &mut quote_cache,
978                                &mut funding_cache,
979                                &idx_map,
980                                &greeks_subs,
981                                clock,
982                            );
983                        }
984                        () = cancel.cancelled() => {
985                            log::debug!("Business websocket stream task cancelled");
986                            break;
987                        }
988                    }
989                }
990            });
991            self.tasks.push(handle);
992        }
993
994        self.is_connected.store(true, Ordering::Release);
995        log::info!("Connected: client_id={}", self.client_id);
996        Ok(())
997    }
998
999    async fn disconnect(&mut self) -> anyhow::Result<()> {
1000        if self.is_disconnected() {
1001            return Ok(());
1002        }
1003
1004        self.cancellation_token.cancel();
1005
1006        if let Some(ref ws) = self.ws_public
1007            && let Err(e) = ws.unsubscribe_all().await
1008        {
1009            log::warn!("Failed to unsubscribe all from public websocket: {e:?}");
1010        }
1011
1012        if let Some(ref ws) = self.ws_business
1013            && let Err(e) = ws.unsubscribe_all().await
1014        {
1015            log::warn!("Failed to unsubscribe all from business websocket: {e:?}");
1016        }
1017
1018        // Allow time for unsubscribe confirmations
1019        tokio::time::sleep(Duration::from_millis(500)).await;
1020
1021        if let Some(ref mut ws) = self.ws_public {
1022            let _result = ws.close().await;
1023        }
1024
1025        if let Some(ref mut ws) = self.ws_business {
1026            let _result = ws.close().await;
1027        }
1028
1029        let handles: Vec<_> = self.tasks.drain(..).collect();
1030
1031        for handle in handles {
1032            if let Err(e) = handle.await {
1033                log::error!("Error joining websocket task: {e}");
1034            }
1035        }
1036
1037        self.book_channels.store(AHashMap::new());
1038        self.option_greeks_subs
1039            .store(AHashMap::<InstrumentId, AHashSet<OKXGreeksType>>::new());
1040        self.option_summary_family_subs
1041            .lock()
1042            .expect("option_summary_family_subs mutex poisoned")
1043            .clear();
1044        self.is_connected.store(false, Ordering::Release);
1045        log::info!("Disconnected: client_id={}", self.client_id);
1046        Ok(())
1047    }
1048
1049    fn is_connected(&self) -> bool {
1050        self.is_connected.load(Ordering::Relaxed)
1051    }
1052
1053    fn is_disconnected(&self) -> bool {
1054        !self.is_connected()
1055    }
1056
1057    fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
1058        for inst_type in &self.config.instrument_types {
1059            let ws = self.public_ws()?.clone();
1060            let inst_type = *inst_type;
1061
1062            self.spawn_ws(
1063                async move {
1064                    ws.subscribe_instruments(inst_type)
1065                        .await
1066                        .context("instruments subscription")?;
1067                    Ok(())
1068                },
1069                "subscribe_instruments",
1070            );
1071        }
1072        Ok(())
1073    }
1074
1075    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
1076        // OKX instruments channel doesn't support subscribing to individual instruments via instId
1077        // Instead, subscribe to the instrument type if not already subscribed
1078        let instrument_id = cmd.instrument_id;
1079        let ws = self.public_ws()?.clone();
1080
1081        self.spawn_ws(
1082            async move {
1083                ws.subscribe_instrument(instrument_id)
1084                    .await
1085                    .context("instrument type subscription")?;
1086                Ok(())
1087            },
1088            "subscribe_instrument",
1089        );
1090        Ok(())
1091    }
1092
1093    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
1094        if cmd.book_type != BookType::L2_MBP {
1095            anyhow::bail!("OKX only supports L2_MBP order book deltas");
1096        }
1097
1098        if is_okx_spread_symbol(cmd.instrument_id.symbol.as_str()) {
1099            // Spreads have no incremental book channel; sprd-books5 pushes a full
1100            // 5-level snapshot, emitted as F_SNAPSHOT deltas to feed the book.
1101            let instrument_id = cmd.instrument_id;
1102            let ws = self.business_ws()?.clone();
1103            self.spawn_ws(
1104                async move {
1105                    ws.subscribe_spread_book(instrument_id)
1106                        .await
1107                        .context("spread book subscription")
1108                },
1109                "spread book subscription",
1110            );
1111            return Ok(());
1112        }
1113
1114        let raw_depth = cmd.depth.map_or(0, |d| d.get());
1115        let depth = resolve_book_depth(raw_depth);
1116        if depth != raw_depth {
1117            log::debug!("Clamped book depth {raw_depth} to {depth} (OKX supports 50 or 400)");
1118        }
1119
1120        let vip = self.vip_level().unwrap_or(OKXVipLevel::Vip0);
1121        let channel = match depth {
1122            50 => {
1123                if vip < OKXVipLevel::Vip4 {
1124                    log::debug!(
1125                        "VIP level {vip} insufficient for 50-depth channel, falling back to default"
1126                    );
1127                    OKXBookChannel::Book
1128                } else {
1129                    OKXBookChannel::Books50L2Tbt
1130                }
1131            }
1132            0 | 400 => {
1133                if vip >= OKXVipLevel::Vip5 {
1134                    OKXBookChannel::BookL2Tbt
1135                } else {
1136                    OKXBookChannel::Book
1137                }
1138            }
1139            _ => unreachable!(),
1140        };
1141
1142        let instrument_id = cmd.instrument_id;
1143        let ws = self.public_ws()?.clone();
1144        let book_channels = Arc::clone(&self.book_channels);
1145
1146        self.spawn_ws(
1147            async move {
1148                match channel {
1149                    OKXBookChannel::Books50L2Tbt => ws
1150                        .subscribe_book50_l2_tbt(instrument_id)
1151                        .await
1152                        .context("books50-l2-tbt subscription")?,
1153                    OKXBookChannel::BookL2Tbt => ws
1154                        .subscribe_book_l2_tbt(instrument_id)
1155                        .await
1156                        .context("books-l2-tbt subscription")?,
1157                    OKXBookChannel::Book => ws
1158                        .subscribe_books_channel(instrument_id)
1159                        .await
1160                        .context("books subscription")?,
1161                }
1162                book_channels.insert(instrument_id, channel);
1163                Ok(())
1164            },
1165            "order book delta subscription",
1166        );
1167
1168        Ok(())
1169    }
1170
1171    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
1172        let instrument_id = cmd.instrument_id;
1173
1174        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
1175            let ws = self.business_ws()?.clone();
1176            self.spawn_ws(
1177                async move {
1178                    ws.subscribe_spread_quotes(instrument_id)
1179                        .await
1180                        .context("spread quotes subscription")
1181                },
1182                "spread quote subscription",
1183            );
1184            return Ok(());
1185        }
1186
1187        let ws = self.public_ws()?.clone();
1188        self.spawn_ws(
1189            async move {
1190                ws.subscribe_quotes(instrument_id)
1191                    .await
1192                    .context("quotes subscription")
1193            },
1194            "quote subscription",
1195        );
1196        Ok(())
1197    }
1198
1199    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
1200        let instrument_id = cmd.instrument_id;
1201
1202        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
1203            let ws = self.business_ws()?.clone();
1204            self.spawn_ws(
1205                async move {
1206                    ws.subscribe_spread_trades(instrument_id)
1207                        .await
1208                        .context("spread trades subscription")
1209                },
1210                "spread trade subscription",
1211            );
1212            return Ok(());
1213        }
1214
1215        let ws = self.public_ws()?.clone();
1216        self.spawn_ws(
1217            async move {
1218                ws.subscribe_trades(instrument_id, false)
1219                    .await
1220                    .context("trades subscription")
1221            },
1222            "trade subscription",
1223        );
1224        Ok(())
1225    }
1226
1227    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
1228        let ws = self.public_ws()?.clone();
1229        let instrument_id = cmd.instrument_id;
1230
1231        self.spawn_ws(
1232            async move {
1233                ws.subscribe_mark_prices(instrument_id)
1234                    .await
1235                    .context("mark price subscription")
1236            },
1237            "mark price subscription",
1238        );
1239        Ok(())
1240    }
1241
1242    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
1243        let ws = self.public_ws()?.clone();
1244        let instrument_id = cmd.instrument_id;
1245        let symbol = instrument_id.symbol.inner();
1246
1247        let (base, quote) = parse_base_quote_from_symbol(symbol.as_str())?;
1248        let base_pair = Ustr::from(&format!("{base}-{quote}"));
1249        self.index_ticker_map.rcu(|m| {
1250            m.entry(base_pair).or_default().insert(symbol);
1251        });
1252
1253        self.spawn_ws(
1254            async move {
1255                ws.subscribe_index_prices(instrument_id)
1256                    .await
1257                    .context("index price subscription")
1258            },
1259            "index price subscription",
1260        );
1261        Ok(())
1262    }
1263
1264    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
1265        let ws = self.business_ws()?.clone();
1266        let bar_type = cmd.bar_type;
1267
1268        self.spawn_ws(
1269            async move {
1270                ws.subscribe_bars(bar_type)
1271                    .await
1272                    .context("bars subscription")
1273            },
1274            "bar subscription",
1275        );
1276        Ok(())
1277    }
1278
1279    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
1280        let ws = self.public_ws()?.clone();
1281        let instrument_id = cmd.instrument_id;
1282
1283        self.spawn_ws(
1284            async move {
1285                ws.subscribe_funding_rates(instrument_id)
1286                    .await
1287                    .context("funding rate subscription")
1288            },
1289            "funding rate subscription",
1290        );
1291        Ok(())
1292    }
1293
1294    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
1295        let instrument_id = cmd.instrument_id;
1296        let conventions = parse_greeks_conventions_from_params(&cmd.params);
1297        self.option_greeks_subs.insert(instrument_id, conventions);
1298
1299        let family = extract_inst_family(instrument_id.symbol.inner().as_str())?;
1300        let is_first = {
1301            let mut family_subs = self
1302                .option_summary_family_subs
1303                .lock()
1304                .expect("option_summary_family_subs mutex poisoned");
1305            let count = family_subs.entry(family).or_default();
1306            *count += 1;
1307            *count == 1
1308        };
1309
1310        if is_first {
1311            let ws = self.public_ws()?.clone();
1312            let family_subs = self.option_summary_family_subs.clone();
1313            self.spawn_ws(
1314                async move {
1315                    let result = ws
1316                        .subscribe_option_summary(family)
1317                        .await
1318                        .context("opt-summary subscription");
1319
1320                    if result.is_err() {
1321                        // Roll back the refcount so a retry can re-arm the subscribe;
1322                        // otherwise the family wedges and Greeks stay dark.
1323                        let mut subs = family_subs
1324                            .lock()
1325                            .expect("option_summary_family_subs mutex poisoned");
1326
1327                        if let Some(count) = subs.get_mut(&family) {
1328                            *count = count.saturating_sub(1);
1329                            if *count == 0 {
1330                                subs.remove(&family);
1331                            }
1332                        }
1333                    }
1334                    result
1335                },
1336                "option greeks subscription",
1337            );
1338        }
1339        Ok(())
1340    }
1341
1342    fn subscribe_instrument_status(
1343        &mut self,
1344        cmd: SubscribeInstrumentStatus,
1345    ) -> anyhow::Result<()> {
1346        let ws = self.public_ws()?.clone();
1347        let instrument_id = cmd.instrument_id;
1348
1349        self.spawn_ws(
1350            async move {
1351                ws.subscribe_instrument(instrument_id)
1352                    .await
1353                    .context("instrument status subscription")
1354            },
1355            "instrument status subscription",
1356        );
1357        Ok(())
1358    }
1359
1360    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
1361        let instrument_id = cmd.instrument_id;
1362        let ws = self.public_ws()?.clone();
1363
1364        self.spawn_ws(
1365            async move {
1366                ws.unsubscribe_instrument(instrument_id)
1367                    .await
1368                    .context("instrument unsubscribe")?;
1369                Ok(())
1370            },
1371            "unsubscribe_instrument",
1372        );
1373        Ok(())
1374    }
1375
1376    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
1377        let instrument_id = cmd.instrument_id;
1378
1379        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
1380            let ws = self.business_ws()?.clone();
1381            self.spawn_ws(
1382                async move {
1383                    ws.unsubscribe_spread_book(instrument_id)
1384                        .await
1385                        .context("spread book unsubscribe")
1386                },
1387                "spread book unsubscribe",
1388            );
1389            return Ok(());
1390        }
1391
1392        let ws = self.public_ws()?.clone();
1393        let channel = self.book_channels.get_cloned(&instrument_id);
1394        self.book_channels.remove(&instrument_id);
1395
1396        self.spawn_ws(
1397            async move {
1398                match channel {
1399                    Some(OKXBookChannel::Books50L2Tbt) => ws
1400                        .unsubscribe_book50_l2_tbt(instrument_id)
1401                        .await
1402                        .context("books50-l2-tbt unsubscribe")?,
1403                    Some(OKXBookChannel::BookL2Tbt) => ws
1404                        .unsubscribe_book_l2_tbt(instrument_id)
1405                        .await
1406                        .context("books-l2-tbt unsubscribe")?,
1407                    Some(OKXBookChannel::Book) => ws
1408                        .unsubscribe_book(instrument_id)
1409                        .await
1410                        .context("book unsubscribe")?,
1411                    None => {
1412                        log::warn!(
1413                            "Book channel not found for {instrument_id}; unsubscribing fallback channel"
1414                        );
1415                        ws.unsubscribe_book(instrument_id)
1416                            .await
1417                            .context("book fallback unsubscribe")?;
1418                    }
1419                }
1420                Ok(())
1421            },
1422            "order book unsubscribe",
1423        );
1424        Ok(())
1425    }
1426
1427    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
1428        let instrument_id = cmd.instrument_id;
1429
1430        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
1431            let ws = self.business_ws()?.clone();
1432            self.spawn_ws(
1433                async move {
1434                    ws.unsubscribe_spread_quotes(instrument_id)
1435                        .await
1436                        .context("spread quotes unsubscribe")
1437                },
1438                "spread quote unsubscribe",
1439            );
1440            return Ok(());
1441        }
1442
1443        let ws = self.public_ws()?.clone();
1444        self.spawn_ws(
1445            async move {
1446                ws.unsubscribe_quotes(instrument_id)
1447                    .await
1448                    .context("quotes unsubscribe")
1449            },
1450            "quote unsubscribe",
1451        );
1452        Ok(())
1453    }
1454
1455    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
1456        let instrument_id = cmd.instrument_id;
1457
1458        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
1459            let ws = self.business_ws()?.clone();
1460            self.spawn_ws(
1461                async move {
1462                    ws.unsubscribe_spread_trades(instrument_id)
1463                        .await
1464                        .context("spread trades unsubscribe")
1465                },
1466                "spread trade unsubscribe",
1467            );
1468            return Ok(());
1469        }
1470
1471        let ws = self.public_ws()?.clone();
1472        self.spawn_ws(
1473            async move {
1474                ws.unsubscribe_trades(instrument_id, false) // TODO: Aggregated trades?
1475                    .await
1476                    .context("trades unsubscribe")
1477            },
1478            "trade unsubscribe",
1479        );
1480        Ok(())
1481    }
1482
1483    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
1484        let ws = self.public_ws()?.clone();
1485        let instrument_id = cmd.instrument_id;
1486
1487        self.spawn_ws(
1488            async move {
1489                ws.unsubscribe_mark_prices(instrument_id)
1490                    .await
1491                    .context("mark price unsubscribe")
1492            },
1493            "mark price unsubscribe",
1494        );
1495        Ok(())
1496    }
1497
1498    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
1499        let ws = self.public_ws()?.clone();
1500        let instrument_id = cmd.instrument_id;
1501        let symbol = instrument_id.symbol.inner();
1502
1503        // The OKX index-tickers channel is keyed by base pair, so multiple
1504        // instruments on the same pair share one subscription. Per-base-pair
1505        // refcounting lives on the WS client, so we always forward the
1506        // unsubscribe and let the WS layer fire the venue request only when
1507        // it knows the last subscriber dropped. Local routing in
1508        // `index_ticker_map` is still maintained for downstream emit fan-out.
1509        if let Ok((base, quote)) = parse_base_quote_from_symbol(symbol.as_str()) {
1510            let base_pair = Ustr::from(&format!("{base}-{quote}"));
1511            self.index_ticker_map.rcu(|m| {
1512                if let Some(set) = m.get_mut(&base_pair) {
1513                    set.remove(&symbol);
1514                    if set.is_empty() {
1515                        m.remove(&base_pair);
1516                    }
1517                }
1518            });
1519        }
1520
1521        self.spawn_ws(
1522            async move {
1523                ws.unsubscribe_index_prices(instrument_id)
1524                    .await
1525                    .context("index price unsubscribe")
1526            },
1527            "index price unsubscribe",
1528        );
1529        Ok(())
1530    }
1531
1532    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
1533        let ws = self.business_ws()?.clone();
1534        let bar_type = cmd.bar_type;
1535
1536        self.spawn_ws(
1537            async move {
1538                ws.unsubscribe_bars(bar_type)
1539                    .await
1540                    .context("bars unsubscribe")
1541            },
1542            "bar unsubscribe",
1543        );
1544        Ok(())
1545    }
1546
1547    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
1548        let ws = self.public_ws()?.clone();
1549        let instrument_id = cmd.instrument_id;
1550
1551        self.spawn_ws(
1552            async move {
1553                ws.unsubscribe_funding_rates(instrument_id)
1554                    .await
1555                    .context("funding rate unsubscribe")
1556            },
1557            "funding rate unsubscribe",
1558        );
1559        Ok(())
1560    }
1561
1562    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
1563        let instrument_id = cmd.instrument_id;
1564        self.option_greeks_subs.remove(&instrument_id);
1565
1566        let family = extract_inst_family(instrument_id.symbol.inner().as_str())?;
1567        let should_unsubscribe = {
1568            let mut family_subs = self
1569                .option_summary_family_subs
1570                .lock()
1571                .expect("option_summary_family_subs mutex poisoned");
1572
1573            if let Some(count) = family_subs.get_mut(&family) {
1574                *count = count.saturating_sub(1);
1575                if *count == 0 {
1576                    family_subs.remove(&family);
1577                    true
1578                } else {
1579                    false
1580                }
1581            } else {
1582                false
1583            }
1584        };
1585
1586        if should_unsubscribe {
1587            let ws = self.public_ws()?.clone();
1588            self.spawn_ws(
1589                async move {
1590                    ws.unsubscribe_option_summary(family)
1591                        .await
1592                        .context("opt-summary unsubscription")
1593                },
1594                "option greeks unsubscription",
1595            );
1596        }
1597        Ok(())
1598    }
1599
1600    fn unsubscribe_instrument_status(
1601        &mut self,
1602        cmd: &UnsubscribeInstrumentStatus,
1603    ) -> anyhow::Result<()> {
1604        let ws = self.public_ws()?.clone();
1605        let instrument_id = cmd.instrument_id;
1606
1607        self.spawn_ws(
1608            async move {
1609                ws.unsubscribe_instrument(instrument_id)
1610                    .await
1611                    .context("instrument status unsubscription")
1612            },
1613            "instrument status unsubscription",
1614        );
1615        Ok(())
1616    }
1617
1618    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
1619        let http = self.http_client.clone();
1620        let sender = self.data_sender.clone();
1621        let instruments_cache = self.instruments.clone();
1622        let request_id = request.request_id;
1623        let client_id = request.client_id.unwrap_or(self.client_id);
1624        let venue = self.venue();
1625        let start = request.start;
1626        let end = request.end;
1627        let params = request.params;
1628        let clock = self.clock;
1629        let start_nanos = datetime_to_unix_nanos(start);
1630        let end_nanos = datetime_to_unix_nanos(end);
1631        let instrument_types = if self.config.instrument_types.is_empty() {
1632            vec![OKXInstrumentType::Spot]
1633        } else {
1634            self.config.instrument_types.clone()
1635        };
1636        let contract_types = self.config.contract_types.clone();
1637        let instrument_families = self.config.instrument_families.clone();
1638        let load_spreads = self.config.load_spreads;
1639
1640        get_runtime().spawn(async move {
1641            let mut all_instruments = Vec::new();
1642
1643            for inst_type in instrument_types {
1644                let Some(families) =
1645                    resolve_instrument_families(&instrument_families, inst_type)
1646                else {
1647                    continue;
1648                };
1649
1650                if families.is_empty() {
1651                    match http.request_instruments(inst_type, None).await {
1652                        Ok((instruments, _inst_id_codes)) => {
1653                            for instrument in instruments {
1654                                if !contract_filter_with_config_types(
1655                                    contract_types.as_ref(),
1656                                    &instrument,
1657                                ) {
1658                                    continue;
1659                                }
1660
1661                                upsert_instrument(&instruments_cache, instrument.clone());
1662                                all_instruments.push(instrument);
1663                            }
1664                        }
1665                        Err(e) => {
1666                            log::error!("Failed to fetch instruments for {inst_type:?}: {e:?}");
1667                        }
1668                    }
1669                } else {
1670                    for family in families {
1671                        match http
1672                            .request_instruments(inst_type, Some(family.clone()))
1673                            .await
1674                        {
1675                            Ok((instruments, _inst_id_codes)) => {
1676                                for instrument in instruments {
1677                                    if !contract_filter_with_config_types(
1678                                        contract_types.as_ref(),
1679                                        &instrument,
1680                                    ) {
1681                                        continue;
1682                                    }
1683
1684                                    upsert_instrument(&instruments_cache, instrument.clone());
1685                                    all_instruments.push(instrument);
1686                                }
1687                            }
1688                            Err(e) => {
1689                                log::error!(
1690                                    "Failed to fetch instruments for {inst_type:?} family {family}: {e:?}"
1691                                );
1692                            }
1693                        }
1694                    }
1695                }
1696            }
1697
1698            if load_spreads {
1699                match http
1700                    .request_spread_instruments(GetSpreadsParams {
1701                        state: Some("live".to_string()),
1702                        ..Default::default()
1703                    })
1704                    .await
1705                {
1706                    Ok(instruments) => {
1707                        for instrument in instruments {
1708                            if !contract_filter_with_config_types(
1709                                contract_types.as_ref(),
1710                                &instrument,
1711                            ) {
1712                                continue;
1713                            }
1714
1715                            upsert_instrument(&instruments_cache, instrument.clone());
1716                            all_instruments.push(instrument);
1717                        }
1718                    }
1719                    Err(e) => {
1720                        log::error!("Failed to fetch OKX spread instruments: {e:?}");
1721                    }
1722                }
1723            }
1724
1725            let response = DataResponse::Instruments(InstrumentsResponse::new(
1726                request_id,
1727                client_id,
1728                venue,
1729                all_instruments,
1730                start_nanos,
1731                end_nanos,
1732                clock.get_time_ns(),
1733                params,
1734            ));
1735
1736            if let Err(e) = sender.send(DataEvent::Response(response)) {
1737                log::error!("Failed to send instruments response: {e}");
1738            }
1739        });
1740
1741        Ok(())
1742    }
1743
1744    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
1745        let http = self.http_client.clone();
1746        let sender = self.data_sender.clone();
1747        let instruments = self.instruments.clone();
1748        let instrument_id = request.instrument_id;
1749        let request_id = request.request_id;
1750        let client_id = request.client_id.unwrap_or(self.client_id);
1751        let start = request.start;
1752        let end = request.end;
1753        let params = request.params;
1754        let clock = self.clock;
1755        let start_nanos = datetime_to_unix_nanos(start);
1756        let end_nanos = datetime_to_unix_nanos(end);
1757        let instrument_types = if self.config.instrument_types.is_empty() {
1758            vec![OKXInstrumentType::Spot]
1759        } else {
1760            self.config.instrument_types.clone()
1761        };
1762        let contract_types = self.config.contract_types.clone();
1763        let load_spreads = self.config.load_spreads;
1764
1765        get_runtime().spawn(async move {
1766            match http
1767                .request_instrument(instrument_id)
1768                .await
1769                .context("fetch instrument from API")
1770            {
1771                Ok(instrument) => {
1772                    let inst_id = instrument.id();
1773                    let symbol = inst_id.symbol.as_str();
1774                    if is_okx_spread_symbol(symbol) {
1775                        if !load_spreads {
1776                            log::error!(
1777                                "Instrument {instrument_id} is a spread but load_spreads is false"
1778                            );
1779                            return;
1780                        }
1781                    } else {
1782                        let inst_type = okx_instrument_type_from_symbol(symbol);
1783                        if !instrument_types.contains(&inst_type) {
1784                            log::error!(
1785                                "Instrument {instrument_id} type {inst_type:?} not in configured types {instrument_types:?}"
1786                            );
1787                            return;
1788                        }
1789                    }
1790
1791                    if !contract_filter_with_config_types(contract_types.as_ref(), &instrument) {
1792                        log::error!(
1793                            "Instrument {instrument_id} filtered out by contract_types config"
1794                        );
1795                        return;
1796                    }
1797
1798                    upsert_instrument(&instruments, instrument.clone());
1799
1800                    let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
1801                        request_id,
1802                        client_id,
1803                        instrument.id(),
1804                        instrument,
1805                        start_nanos,
1806                        end_nanos,
1807                        clock.get_time_ns(),
1808                        params,
1809                    )));
1810
1811                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1812                        log::error!("Failed to send instrument response: {e}");
1813                    }
1814                }
1815                Err(e) if e.downcast_ref::<OKXInstrumentDefinitionError>().is_some() => {
1816                    log::warn!("Instrument request skipped: {e:?}");
1817                }
1818                Err(e) => log::error!("Instrument request failed: {e:?}"),
1819            }
1820        });
1821
1822        Ok(())
1823    }
1824
1825    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
1826        let http = self.http_client.clone();
1827        let sender = self.data_sender.clone();
1828        let instrument_id = request.instrument_id;
1829        let depth = request.depth.map(|n| n.get() as u32);
1830        let request_id = request.request_id;
1831        let client_id = request.client_id.unwrap_or(self.client_id);
1832        let params = request.params;
1833        let clock = self.clock;
1834
1835        get_runtime().spawn(async move {
1836            match http
1837                .request_book_snapshot(instrument_id, depth)
1838                .await
1839                .context("failed to request book snapshot from OKX")
1840            {
1841                Ok(book) => {
1842                    let response = DataResponse::Book(BookResponse::new(
1843                        request_id,
1844                        client_id,
1845                        instrument_id,
1846                        book,
1847                        None,
1848                        None,
1849                        clock.get_time_ns(),
1850                        params,
1851                    ));
1852
1853                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1854                        log::error!("Failed to send book snapshot response: {e}");
1855                    }
1856                }
1857                Err(e) => log::error!("Book snapshot request failed: {e:?}"),
1858            }
1859        });
1860
1861        Ok(())
1862    }
1863
1864    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
1865        let http = self.http_client.clone();
1866        let sender = self.data_sender.clone();
1867        let instrument_id = request.instrument_id;
1868        let start = request.start;
1869        let end = request.end;
1870        let limit = request.limit.map(|n| n.get() as u32);
1871        let request_id = request.request_id;
1872        let client_id = request.client_id.unwrap_or(self.client_id);
1873        let params = request.params;
1874        let clock = self.clock;
1875        let start_nanos = datetime_to_unix_nanos(start);
1876        let end_nanos = datetime_to_unix_nanos(end);
1877
1878        get_runtime().spawn(async move {
1879            match http
1880                .request_trades(instrument_id, start, end, limit)
1881                .await
1882                .context("failed to request trades from OKX")
1883            {
1884                Ok(trades) => {
1885                    let response = DataResponse::Trades(TradesResponse::new(
1886                        request_id,
1887                        client_id,
1888                        instrument_id,
1889                        trades,
1890                        start_nanos,
1891                        end_nanos,
1892                        clock.get_time_ns(),
1893                        params,
1894                    ));
1895
1896                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1897                        log::error!("Failed to send trades response: {e}");
1898                    }
1899                }
1900                Err(e) => log::error!("Trade request failed: {e:?}"),
1901            }
1902        });
1903
1904        Ok(())
1905    }
1906
1907    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
1908        let http = self.http_client.clone();
1909        let sender = self.data_sender.clone();
1910        let bar_type = request.bar_type;
1911        let start = request.start;
1912        let end = request.end;
1913        let limit = request.limit.map(|n| n.get() as u32);
1914        let request_id = request.request_id;
1915        let client_id = request.client_id.unwrap_or(self.client_id);
1916        let params = request.params;
1917        let clock = self.clock;
1918        let start_nanos = datetime_to_unix_nanos(start);
1919        let end_nanos = datetime_to_unix_nanos(end);
1920
1921        get_runtime().spawn(async move {
1922            match http
1923                .request_bars(bar_type, start, end, limit)
1924                .await
1925                .context("failed to request bars from OKX")
1926            {
1927                Ok(bars) => {
1928                    let response = DataResponse::Bars(BarsResponse::new(
1929                        request_id,
1930                        client_id,
1931                        bar_type,
1932                        bars,
1933                        start_nanos,
1934                        end_nanos,
1935                        clock.get_time_ns(),
1936                        params,
1937                    ));
1938
1939                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1940                        log::error!("Failed to send bars response: {e}");
1941                    }
1942                }
1943                Err(e) => log::error!("Bar request failed: {e:?}"),
1944            }
1945        });
1946
1947        Ok(())
1948    }
1949
1950    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
1951        let http = self.http_client.clone();
1952        let sender = self.data_sender.clone();
1953        let instrument_id = request.instrument_id;
1954        let start = request.start;
1955        let end = request.end;
1956        let limit = request.limit.map(|n| n.get() as u32);
1957        let request_id = request.request_id;
1958        let client_id = request.client_id.unwrap_or(self.client_id);
1959        let params = request.params;
1960        let clock = self.clock;
1961        let start_nanos = datetime_to_unix_nanos(start);
1962        let end_nanos = datetime_to_unix_nanos(end);
1963
1964        get_runtime().spawn(async move {
1965            match http
1966                .request_funding_rates(instrument_id, start, end, limit)
1967                .await
1968                .context("failed to request funding rates from OKX")
1969            {
1970                Ok(funding_rates) => {
1971                    let response = DataResponse::FundingRates(FundingRatesResponse::new(
1972                        request_id,
1973                        client_id,
1974                        instrument_id,
1975                        funding_rates,
1976                        start_nanos,
1977                        end_nanos,
1978                        clock.get_time_ns(),
1979                        params,
1980                    ));
1981
1982                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1983                        log::error!("Failed to send funding rates response: {e}");
1984                    }
1985                }
1986                Err(e) => log::error!("Funding rates request failed: {e:?}"),
1987            }
1988        });
1989
1990        Ok(())
1991    }
1992
1993    fn request_forward_prices(&self, request: RequestForwardPrices) -> anyhow::Result<()> {
1994        let http = self.http_client.clone();
1995        let sender = self.data_sender.clone();
1996        let underlying = request.underlying.to_string();
1997        let instrument_id = request.instrument_id;
1998        let request_id = request.request_id;
1999        let client_id = request.client_id.unwrap_or(self.client_id);
2000        let params = request.params;
2001        let clock = self.clock;
2002        let venue = *OKX_VENUE;
2003
2004        get_runtime().spawn(async move {
2005            match http
2006                .request_forward_prices(&underlying, instrument_id)
2007                .await
2008                .context("failed to request forward prices from OKX")
2009            {
2010                Ok(forward_prices) => {
2011                    let response = DataResponse::ForwardPrices(ForwardPricesResponse::new(
2012                        request_id,
2013                        client_id,
2014                        venue,
2015                        forward_prices,
2016                        clock.get_time_ns(),
2017                        params,
2018                    ));
2019
2020                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2021                        log::error!("Failed to send forward prices response: {e}");
2022                    }
2023                }
2024                Err(e) => {
2025                    log::error!("Forward prices request failed for {underlying}: {e:?}");
2026                    let response = DataResponse::ForwardPrices(ForwardPricesResponse::new(
2027                        request_id,
2028                        client_id,
2029                        venue,
2030                        Vec::new(),
2031                        clock.get_time_ns(),
2032                        params,
2033                    ));
2034
2035                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2036                        log::error!("Failed to send forward prices response: {e}");
2037                    }
2038                }
2039            }
2040        });
2041
2042        Ok(())
2043    }
2044}
2045
2046#[cfg(test)]
2047mod tests {
2048    use std::sync::Arc;
2049
2050    use rstest::rstest;
2051    use serde_json::json;
2052
2053    use super::*;
2054
2055    fn both() -> AHashSet<OKXGreeksType> {
2056        [OKXGreeksType::Bs, OKXGreeksType::Pa].into_iter().collect()
2057    }
2058
2059    fn only(greeks_type: OKXGreeksType) -> AHashSet<OKXGreeksType> {
2060        [greeks_type].into_iter().collect()
2061    }
2062
2063    #[rstest]
2064    fn dispatch_parsed_data_emits_instrument_status() {
2065        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
2066        let instruments = Arc::new(AtomicMap::new());
2067        let mut instruments_by_symbol = AHashMap::new();
2068        let status = InstrumentStatus::new(
2069            InstrumentId::from("USDG-SGD.OKX"),
2070            MarketStatusAction::Trading,
2071            UnixNanos::from(1u64),
2072            UnixNanos::from(2u64),
2073            None,
2074            None,
2075            Some(true),
2076            None,
2077            None,
2078        );
2079
2080        dispatch_parsed_data(
2081            NautilusWsMessage::InstrumentStatus(status),
2082            &sender,
2083            &instruments,
2084            &mut instruments_by_symbol,
2085        );
2086
2087        match receiver.try_recv().expect("instrument status event") {
2088            DataEvent::InstrumentStatus(received) => assert_eq!(received, status),
2089            other => panic!("Expected DataEvent::InstrumentStatus, was {other:?}"),
2090        }
2091        assert!(instruments_by_symbol.is_empty());
2092        assert!(instruments.load().is_empty());
2093    }
2094
2095    #[rstest]
2096    fn parse_conventions_returns_both_when_params_missing() {
2097        let result = parse_greeks_conventions_from_params(&None);
2098        assert_eq!(result, both());
2099    }
2100
2101    #[rstest]
2102    fn parse_conventions_returns_both_when_key_absent() {
2103        let mut params = Params::new();
2104        params.insert("other_key".to_string(), json!("value"));
2105        let result = parse_greeks_conventions_from_params(&Some(params));
2106        assert_eq!(result, both());
2107    }
2108
2109    #[rstest]
2110    #[case("BLACK_SCHOLES", OKXGreeksType::Bs)]
2111    #[case("PRICE_ADJUSTED", OKXGreeksType::Pa)]
2112    #[case("black_scholes", OKXGreeksType::Bs)]
2113    #[case("price_adjusted", OKXGreeksType::Pa)]
2114    fn parse_conventions_accepts_single_string(#[case] raw: &str, #[case] expected: OKXGreeksType) {
2115        let mut params = Params::new();
2116        params.insert("greeks_convention".to_string(), json!(raw));
2117        let result = parse_greeks_conventions_from_params(&Some(params));
2118        assert_eq!(result, only(expected));
2119    }
2120
2121    #[rstest]
2122    fn parse_conventions_accepts_list_of_strings() {
2123        let mut params = Params::new();
2124        params.insert(
2125            "greeks_convention".to_string(),
2126            json!(["BLACK_SCHOLES", "PRICE_ADJUSTED"]),
2127        );
2128        let result = parse_greeks_conventions_from_params(&Some(params));
2129        assert_eq!(result, both());
2130    }
2131
2132    #[rstest]
2133    fn parse_conventions_accepts_single_entry_list() {
2134        let mut params = Params::new();
2135        params.insert("greeks_convention".to_string(), json!(["PRICE_ADJUSTED"]));
2136        let result = parse_greeks_conventions_from_params(&Some(params));
2137        assert_eq!(result, only(OKXGreeksType::Pa));
2138    }
2139
2140    #[rstest]
2141    fn parse_conventions_deduplicates_list_entries() {
2142        let mut params = Params::new();
2143        params.insert(
2144            "greeks_convention".to_string(),
2145            json!(["BLACK_SCHOLES", "black_scholes"]),
2146        );
2147        let result = parse_greeks_conventions_from_params(&Some(params));
2148        assert_eq!(result, only(OKXGreeksType::Bs));
2149    }
2150
2151    #[rstest]
2152    fn parse_conventions_skips_unknown_list_entries() {
2153        let mut params = Params::new();
2154        params.insert(
2155            "greeks_convention".to_string(),
2156            json!(["BOGUS", "PRICE_ADJUSTED"]),
2157        );
2158        let result = parse_greeks_conventions_from_params(&Some(params));
2159        assert_eq!(result, only(OKXGreeksType::Pa));
2160    }
2161
2162    #[rstest]
2163    fn parse_conventions_falls_back_to_both_on_all_unknown() {
2164        let mut params = Params::new();
2165        params.insert("greeks_convention".to_string(), json!(["BOGUS"]));
2166        let result = parse_greeks_conventions_from_params(&Some(params));
2167        assert_eq!(result, both());
2168    }
2169
2170    #[rstest]
2171    #[case(json!(1))]
2172    #[case(json!(null))]
2173    #[case(json!(true))]
2174    #[case(json!({"nested": "object"}))]
2175    fn parse_conventions_falls_back_on_non_string_value(#[case] value: serde_json::Value) {
2176        let mut params = Params::new();
2177        params.insert("greeks_convention".to_string(), value);
2178        let result = parse_greeks_conventions_from_params(&Some(params));
2179        assert_eq!(result, both());
2180    }
2181
2182    #[rstest]
2183    fn parse_conventions_falls_back_on_unknown_single_string() {
2184        let mut params = Params::new();
2185        params.insert("greeks_convention".to_string(), json!("BOGUS"));
2186        let result = parse_greeks_conventions_from_params(&Some(params));
2187        assert_eq!(result, both());
2188    }
2189}
nautilus_okx/data.rs

nautilus_okx/
data.rs
