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::{
19    sync::{
20        Arc,
21        atomic::{AtomicBool, AtomicU64, Ordering},
22    },
23    time::{Duration, Instant},
24};
25
26use ahash::{AHashMap, AHashSet};
27use anyhow::Context;
28use futures_util::{StreamExt, pin_mut};
29use nautilus_common::{
30    cache::quote::QuoteCache,
31    clients::DataClient,
32    live::runner::get_data_event_sender,
33    messages::{
34        DataEvent,
35        data::{
36            BarsResponse, BookResponse, DataResponse, ForwardPricesResponse, FundingRatesResponse,
37            InstrumentResponse, InstrumentsResponse, RequestBars, RequestBookSnapshot,
38            RequestForwardPrices, RequestFundingRates, RequestInstrument, RequestInstruments,
39            RequestTrades, SubscribeBars, SubscribeBookDeltas, SubscribeFundingRates,
40            SubscribeIndexPrices, SubscribeInstrument, SubscribeInstrumentStatus,
41            SubscribeInstruments, SubscribeMarkPrices, SubscribeOptionGreeks, SubscribeQuotes,
42            SubscribeTrades, TradesResponse, UnsubscribeBars, UnsubscribeBookDeltas,
43            UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrument,
44            UnsubscribeInstrumentStatus, UnsubscribeMarkPrices, UnsubscribeOptionGreeks,
45            UnsubscribeQuotes, UnsubscribeTrades,
46        },
47    },
48};
49use nautilus_core::{
50    AtomicMap, Params, UnixNanos,
51    datetime::datetime_to_unix_nanos,
52    time::{AtomicTime, get_atomic_clock_realtime},
53};
54use nautilus_live::{
55    SocketControl,
56    task::{TaskGroup, TaskGroupGuard, TaskSpawner},
57};
58use nautilus_model::{
59    data::{Data, FundingRateUpdate, InstrumentStatus},
60    enums::{BookType, GreeksConvention, MarketStatusAction},
61    identifiers::{ClientId, InstrumentId, Venue},
62    instruments::{Instrument, InstrumentAny},
63};
64use ustr::Ustr;
65
66use crate::{
67    book_sync::{
68        BookChannelScope, BookSequenceOutcome, BookSyncSignal, BookSyncSignalKind, BookSyncTracker,
69    },
70    common::{
71        consts::{
72            OKX_VENUE, OKX_WS_HEARTBEAT_SECS, resolve_book_depth, resolve_instrument_families,
73            select_book_channel, should_retry_error_code,
74        },
75        enums::{
76            OKXBookAction, OKXBookChannel, OKXContractType, OKXGreeksType, OKXInstrumentStatus,
77            OKXInstrumentType, OKXVipLevel,
78        },
79        models::OKXInstrument,
80        parse::{
81            extract_inst_family, is_okx_spread_symbol, okx_instrument_type_from_symbol,
82            okx_status_to_market_action, parse_base_quote_from_symbol, parse_instrument_any,
83            parse_instrument_id, parse_millisecond_timestamp, parse_price, parse_quantity,
84        },
85        task::{spawn_task, terminate_tasks},
86    },
87    config::OKXDataClientConfig,
88    http::{
89        client::{OKXHttpClient, OKXInstrumentDefinitionError},
90        query::GetSpreadsParams,
91    },
92    websocket::{
93        client::OKXWebSocketClient,
94        enums::OKXWsChannel,
95        messages::{NautilusWsMessage, OKXBookMsg, OKXOptionSummaryMsg, OKXWsMessage},
96        parse::{
97            extract_fees_from_cached_instrument, parse_book_msg_vec, parse_index_price_msg_vec,
98            parse_option_summary_greeks, parse_rpi_book_msg_vec, parse_ws_message_data,
99        },
100    },
101};
102
103#[derive(Debug)]
104pub struct OKXDataClient {
105    client_id: ClientId,
106    config: OKXDataClientConfig,
107    http_client: OKXHttpClient,
108    ws_public: Option<OKXWebSocketClient>,
109    ws_business: Option<OKXWebSocketClient>,
110    is_connected: AtomicBool,
111    transports_started: bool,
112    tasks: TaskGroup,
113    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
114    // Shared instrument cache keyed by raw symbol so stream tasks, reconciliation,
115    // and request paths all read and write one source of truth
116    instruments_by_symbol: Arc<AtomicMap<Ustr, InstrumentAny>>,
117    // Serializes instrument diff-update-publish sequences between the stream
118    // tasks and the refresh task; only held for synchronous sections
119    instrument_update_lock: Arc<InstrumentUpdateLock>,
120    book_channels: Arc<AtomicMap<InstrumentId, OKXBookChannel>>,
121    book_sync: BookSyncTracker,
122    index_ticker_map: Arc<AtomicMap<Ustr, AHashSet<Ustr>>>,
123    option_greeks_subs: Arc<AtomicMap<InstrumentId, AHashSet<OKXGreeksType>>>,
124    // `Mutex<AHashMap>` so the spawned subscribe task can roll back the
125    // refcount on failure. A bare `AHashMap` would leave the count
126    // permanently incremented and wedge future Greeks subscribes.
127    option_summary_family_subs: Arc<parking_lot::Mutex<AHashMap<Ustr, usize>>>,
128    clock: &'static AtomicTime,
129}
130
131impl OKXDataClient {
132    /// Creates a new [`OKXDataClient`] instance.
133    ///
134    /// # Errors
135    ///
136    /// Returns an error if the client fails to initialize.
137    pub fn new(client_id: ClientId, config: OKXDataClientConfig) -> anyhow::Result<Self> {
138        let clock = get_atomic_clock_realtime();
139        let data_sender = get_data_event_sender();
140
141        let http_client = if config.has_api_credentials() {
142            OKXHttpClient::with_credentials(
143                config.api_key.clone(),
144                config.api_secret.clone(),
145                config.api_passphrase.clone(),
146                Some(config.http_base_url()),
147                config.http_timeout_secs,
148                config.max_retries,
149                config.retry_delay_initial_ms,
150                config.retry_delay_max_ms,
151                config.environment,
152                config.proxy_url.clone(),
153            )?
154        } else {
155            OKXHttpClient::new(
156                Some(config.http_base_url()),
157                config.http_timeout_secs,
158                config.max_retries,
159                config.retry_delay_initial_ms,
160                config.retry_delay_max_ms,
161                config.environment,
162                config.proxy_url.clone(),
163            )?
164        };
165
166        let ws_public = OKXWebSocketClient::new(
167            Some(config.ws_public_url()),
168            None,
169            None,
170            None,
171            None,
172            Some(OKX_WS_HEARTBEAT_SECS),
173            None,
174            config.transport_backend,
175            config.proxy_url.clone(),
176        )
177        .context("failed to construct OKX public websocket client")?
178        .with_socket_control(SocketControl::new(
179            client_id,
180            Some(*OKX_VENUE),
181            "okx-public-data-streams",
182        ));
183
184        let ws_business = if config.requires_business_ws() {
185            let ws = OKXWebSocketClient::new(
186                Some(config.ws_business_url()),
187                None, // No auth needed for public business channels
188                None,
189                None,
190                None,
191                Some(OKX_WS_HEARTBEAT_SECS),
192                None,
193                config.transport_backend,
194                config.proxy_url.clone(),
195            )
196            .context("failed to construct OKX business websocket client")?
197            .with_socket_control(SocketControl::new(
198                client_id,
199                Some(*OKX_VENUE),
200                "okx-business-data-streams",
201            ));
202            Some(ws)
203        } else {
204            None
205        };
206
207        if let Some(vip_level) = config.vip_level {
208            ws_public.set_vip_level(vip_level);
209
210            if let Some(ref ws) = ws_business {
211                ws.set_vip_level(vip_level);
212            }
213        }
214
215        Ok(Self {
216            client_id,
217            config,
218            http_client,
219            ws_public: Some(ws_public),
220            ws_business,
221            is_connected: AtomicBool::new(false),
222            transports_started: false,
223            tasks: TaskGroup::new(),
224            data_sender,
225            instruments_by_symbol: Arc::new(AtomicMap::new()),
226            instrument_update_lock: Arc::new(InstrumentUpdateLock::default()),
227            book_channels: Arc::new(AtomicMap::new()),
228            book_sync: BookSyncTracker::default(),
229            index_ticker_map: Arc::new(AtomicMap::new()),
230            option_greeks_subs: Arc::new(AtomicMap::new()),
231            option_summary_family_subs: Arc::new(parking_lot::Mutex::new(AHashMap::new())),
232            clock,
233        })
234    }
235
236    fn venue(&self) -> Venue {
237        *OKX_VENUE
238    }
239
240    fn vip_level(&self) -> Option<OKXVipLevel> {
241        self.ws_public.as_ref().map(|ws| ws.vip_level())
242    }
243
244    fn public_ws(&self) -> anyhow::Result<&OKXWebSocketClient> {
245        self.ws_public
246            .as_ref()
247            .context("public websocket client not initialized")
248    }
249
250    fn business_ws(&self) -> anyhow::Result<&OKXWebSocketClient> {
251        self.ws_business
252            .as_ref()
253            .context("business websocket client not available (credentials required)")
254    }
255
256    fn send_data(sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>, data: Data) {
257        if let Err(e) = sender.send(DataEvent::Data(data)) {
258            log::error!("Failed to emit data event: {e}");
259        }
260    }
261
262    fn spawn_ws<F>(&self, fut: F, context: &'static str)
263    where
264        F: Future<Output = anyhow::Result<()>> + Send + 'static,
265    {
266        let fut = async move {
267            if let Err(e) = fut.await {
268                log::error!("{context}: {e:?}");
269            }
270        };
271        self.spawn_task(fut);
272    }
273
274    fn spawn_task<F>(&self, fut: F)
275    where
276        F: Future<Output = ()> + Send + 'static,
277    {
278        match self.tasks.spawner() {
279            Ok(spawner) => spawn_task(&spawner, fut),
280            Err(e) => log::debug!("Skipping task after OKX shutdown began: {e}"),
281        }
282    }
283
284    fn begin_generation_shutdown(&self) {
285        self.tasks.begin_shutdown();
286        self.is_connected.store(false, Ordering::Release);
287
288        if let Some(ws) = self.ws_public.as_ref() {
289            ws.begin_shutdown();
290        }
291
292        if let Some(ws) = self.ws_business.as_ref() {
293            ws.begin_shutdown();
294        }
295    }
296
297    fn register_book_health_monitor(&self) -> anyhow::Result<()> {
298        let interval_duration = Duration::from_secs(self.config.book_stale_check_interval_secs);
299        let threshold = Duration::from_secs(self.config.book_stale_threshold_secs);
300
301        if interval_duration.is_zero() || threshold.is_zero() {
302            return Ok(());
303        }
304
305        let book_sync = self.book_sync.clone();
306        let tasks = self
307            .tasks
308            .spawner()
309            .context("OKX data task admission is closed")?;
310        let cancel = tasks.cancellation_token();
311
312        tasks.spawn(async move {
313            let mut interval = tokio::time::interval(interval_duration);
314
315            loop {
316                tokio::select! {
317                    biased;
318                    () = cancel.cancelled() => {
319                        log::debug!("Book health monitor task cancelled");
320                        break;
321                    }
322                    _ = interval.tick() => {
323                        handle_book_sync_signals(
324                            book_sync.stale_books(threshold, Instant::now())
325                        );
326                    }
327                }
328            }
329        })?;
330        Ok(())
331    }
332
333    #[expect(clippy::too_many_arguments)]
334    fn handle_ws_message(
335        message: OKXWsMessage,
336        data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
337        instruments_by_symbol: &Arc<AtomicMap<Ustr, InstrumentAny>>,
338        http_client: &OKXHttpClient,
339        config: &OKXDataClientConfig,
340        instrument_update_lock: &InstrumentUpdateLock,
341        book_channels: &Arc<AtomicMap<InstrumentId, OKXBookChannel>>,
342        book_sync: &BookSyncTracker,
343        recovery_ws: Option<&OKXWebSocketClient>,
344        business_ws: Option<&OKXWebSocketClient>,
345        quote_cache: &mut QuoteCache,
346        funding_cache: &mut AHashMap<Ustr, (Ustr, u64)>,
347        index_ticker_map: &Arc<AtomicMap<Ustr, AHashSet<Ustr>>>,
348        option_greeks_subs: &Arc<AtomicMap<InstrumentId, AHashSet<OKXGreeksType>>>,
349        book_channel_scope: BookChannelScope,
350        snapshot_timeout: Duration,
351        tasks: &TaskSpawner,
352        clock: &AtomicTime,
353    ) {
354        match message {
355            OKXWsMessage::BookData { arg, action, data } => {
356                let Some(inst_id) = arg.inst_id else {
357                    log::warn!("Book data without inst_id");
358                    return;
359                };
360                let instruments_guard = instruments_by_symbol.load();
361                let Some(instrument) = instruments_guard.get(&inst_id) else {
362                    log::warn!("No cached instrument for book data: {inst_id}");
363                    return;
364                };
365                let ts_init = clock.get_time_ns();
366                let sequences = data
367                    .iter()
368                    .map(|msg| (msg.prev_seq_id, msg.seq_id))
369                    .collect::<Vec<_>>();
370
371                match parse_book_msg_vec(
372                    data,
373                    &instrument.id(),
374                    instrument.price_precision(),
375                    instrument.size_precision(),
376                    action,
377                    ts_init,
378                ) {
379                    Ok(data_vec) => {
380                        let outcome = book_sync.validate_sequence_if_subscribed(
381                            book_channels,
382                            instrument.id(),
383                            action == OKXBookAction::Snapshot,
384                            &sequences,
385                            snapshot_timeout,
386                            Instant::now(),
387                        );
388
389                        if !handle_book_sequence_outcome(
390                            outcome,
391                            instrument.id(),
392                            book_channels,
393                            book_sync,
394                            recovery_ws,
395                            snapshot_timeout,
396                            tasks,
397                        ) {
398                            return;
399                        }
400
401                        for data in data_vec {
402                            Self::send_data(data_sender, data);
403                        }
404                    }
405                    Err(e) => log::error!("Failed to parse book data: {e}"),
406                }
407            }
408            OKXWsMessage::RpiBookData { arg, action, data } => {
409                let Some(inst_id) = arg.inst_id else {
410                    log::warn!("RPI book data without inst_id");
411                    return;
412                };
413                let instruments_guard = instruments_by_symbol.load();
414                let Some(instrument) = instruments_guard.get(&inst_id) else {
415                    log::warn!("No cached instrument for RPI book data: {inst_id}");
416                    return;
417                };
418                let ts_init = clock.get_time_ns();
419                let sequences = data
420                    .iter()
421                    .map(|msg| (Some(msg.prev_seq_id), msg.seq_id))
422                    .collect::<Vec<_>>();
423
424                match parse_rpi_book_msg_vec(
425                    data,
426                    &instrument.id(),
427                    instrument.price_precision(),
428                    instrument.size_precision(),
429                    action,
430                    ts_init,
431                ) {
432                    Ok(data_vec) => {
433                        let outcome = book_sync.validate_sequence_if_subscribed(
434                            book_channels,
435                            instrument.id(),
436                            action == OKXBookAction::Snapshot,
437                            &sequences,
438                            snapshot_timeout,
439                            Instant::now(),
440                        );
441
442                        if !handle_book_sequence_outcome(
443                            outcome,
444                            instrument.id(),
445                            book_channels,
446                            book_sync,
447                            recovery_ws,
448                            snapshot_timeout,
449                            tasks,
450                        ) {
451                            return;
452                        }
453
454                        for data in data_vec {
455                            Self::send_data(data_sender, data);
456                        }
457                    }
458                    Err(e) => log::error!("Failed to parse RPI book data: {e}"),
459                }
460            }
461            OKXWsMessage::ChannelData {
462                channel,
463                inst_id,
464                data,
465            } => {
466                // Option summary subscriptions use instFamily (not instId), so
467                // the arg has inst_id: None. Each element in the data array carries
468                // its own inst_id that we resolve per-message.
469                if matches!(channel, OKXWsChannel::OptionSummary) {
470                    let ts_init = clock.get_time_ns();
471
472                    match serde_json::from_value::<Vec<OKXOptionSummaryMsg>>(data) {
473                        Ok(msgs) => {
474                            let subs = option_greeks_subs.load();
475                            let instruments_guard = instruments_by_symbol.load();
476
477                            for msg in &msgs {
478                                let Some(instrument) = instruments_guard.get(&msg.inst_id) else {
479                                    continue;
480                                };
481                                let instrument_id = instrument.id();
482                                let Some(conventions) = subs.get(&instrument_id) else {
483                                    continue;
484                                };
485
486                                for greeks_type in conventions {
487                                    match parse_option_summary_greeks(
488                                        msg,
489                                        &instrument_id,
490                                        *greeks_type,
491                                        ts_init,
492                                    ) {
493                                        Ok(greeks) => {
494                                            if let Err(e) =
495                                                data_sender.send(DataEvent::OptionGreeks(greeks))
496                                            {
497                                                log::error!(
498                                                    "Failed to emit option greeks event: {e}"
499                                                );
500                                            }
501                                        }
502                                        Err(e) => {
503                                            log::error!(
504                                                "Failed to parse option summary for {} ({greeks_type:?}): {e}",
505                                                msg.inst_id
506                                            );
507                                        }
508                                    }
509                                }
510                            }
511                        }
512                        Err(e) => {
513                            log::error!("Failed to deserialize option summary data: {e}");
514                        }
515                    }
516                    return;
517                }
518
519                let Some(inst_id) = inst_id else {
520                    log::debug!("Channel data without inst_id: {channel:?}");
521                    return;
522                };
523
524                // Index tickers use base pair format (e.g., "BTC-USDT") but instruments
525                // are keyed by full symbol (e.g., "BTC-USDT-SWAP"). Dispatch index price
526                // updates only to instruments that subscribed via subscribe_index_prices.
527                if matches!(channel, OKXWsChannel::IndexTickers) {
528                    let ts_init = clock.get_time_ns();
529                    let map_guard = index_ticker_map.load();
530                    let Some(subscribed_symbols) = map_guard.get(&inst_id) else {
531                        log::debug!("No subscribed instruments for index ticker: {inst_id}");
532                        return;
533                    };
534                    let symbols: Vec<Ustr> = subscribed_symbols.iter().copied().collect();
535                    drop(map_guard);
536
537                    let instruments_guard = instruments_by_symbol.load();
538
539                    for sym in &symbols {
540                        let Some(instrument) = instruments_guard.get(sym) else {
541                            log::warn!("No cached instrument for index ticker symbol: {sym}");
542                            continue;
543                        };
544
545                        match parse_index_price_msg_vec(
546                            data.clone(),
547                            &instrument.id(),
548                            instrument.price_precision(),
549                            ts_init,
550                        ) {
551                            Ok(data_vec) => {
552                                for d in data_vec {
553                                    Self::send_data(data_sender, d);
554                                }
555                            }
556                            Err(e) => log::error!("Failed to parse index price data: {e}"),
557                        }
558                    }
559                    return;
560                }
561
562                let instruments_guard = instruments_by_symbol.load();
563                let Some(instrument) = instruments_guard.get(&inst_id) else {
564                    log::warn!("No cached instrument for {channel:?}: {inst_id}");
565                    return;
566                };
567                let instrument_id = instrument.id();
568                let price_precision = instrument.price_precision();
569                let size_precision = instrument.size_precision();
570                let ts_init = clock.get_time_ns();
571
572                if matches!(channel, OKXWsChannel::SprdBooks5) {
573                    let msgs: Vec<OKXBookMsg> = match serde_json::from_value(data) {
574                        Ok(m) => m,
575                        Err(e) => {
576                            log::error!("Failed to deserialize spread book data: {e}");
577                            return;
578                        }
579                    };
580
581                    // sprd-books5 pushes a full 5-level snapshot each message.
582                    match parse_book_msg_vec(
583                        msgs,
584                        &instrument_id,
585                        price_precision,
586                        size_precision,
587                        OKXBookAction::Snapshot,
588                        ts_init,
589                    ) {
590                        Ok(data_vec) => {
591                            book_sync.record_update_if_subscribed(
592                                book_channels,
593                                instrument_id,
594                                true,
595                                Instant::now(),
596                            );
597
598                            for d in data_vec {
599                                Self::send_data(data_sender, d);
600                            }
601                        }
602                        Err(e) => log::error!("Failed to parse spread book data: {e}"),
603                    }
604
605                    return;
606                }
607
608                if matches!(channel, OKXWsChannel::BboTbt | OKXWsChannel::SprdBboTbt) {
609                    let msgs: Vec<OKXBookMsg> = match serde_json::from_value(data) {
610                        Ok(m) => m,
611                        Err(e) => {
612                            log::error!("Failed to deserialize BboTbt data: {e}");
613                            return;
614                        }
615                    };
616
617                    for msg in &msgs {
618                        let bid = msg.bids.first();
619                        let ask = msg.asks.first();
620                        let bid_price =
621                            bid.and_then(|e| parse_price(&e.price, price_precision).ok());
622                        let bid_size =
623                            bid.and_then(|e| parse_quantity(&e.size, size_precision).ok());
624                        let ask_price =
625                            ask.and_then(|e| parse_price(&e.price, price_precision).ok());
626                        let ask_size =
627                            ask.and_then(|e| parse_quantity(&e.size, size_precision).ok());
628                        let ts_event = parse_millisecond_timestamp(msg.ts);
629
630                        match quote_cache.process(
631                            instrument_id,
632                            bid_price,
633                            ask_price,
634                            bid_size,
635                            ask_size,
636                            ts_event,
637                            ts_init,
638                        ) {
639                            Ok(quote) => Self::send_data(data_sender, Data::Quote(quote)),
640                            Err(e) => {
641                                log::debug!("Skipping partial BboTbt for {instrument_id}: {e}");
642                            }
643                        }
644                    }
645
646                    return;
647                }
648
649                match parse_ws_message_data(
650                    &channel,
651                    data,
652                    &instrument_id,
653                    price_precision,
654                    size_precision,
655                    ts_init,
656                    funding_cache,
657                    &instruments_guard,
658                ) {
659                    Ok(Some(ws_msg)) => {
660                        dispatch_parsed_data(ws_msg, data_sender, instruments_by_symbol);
661                    }
662                    Ok(None) => {}
663                    Err(e) => log::error!("Failed to parse {channel:?} data: {e}"),
664                }
665            }
666            OKXWsMessage::Instruments(okx_instruments) => {
667                let ts_init = clock.get_time_ns();
668                // Hold the instrument lock for the batch so a concurrent
669                // reconciliation cannot interleave diff, cache update, and publish
670                let _update_guard = instrument_update_lock.mutex.lock();
671
672                for okx_inst in okx_instruments {
673                    let inst_key = okx_inst.inst_id;
674                    let cached = instruments_by_symbol.get_cloned(&inst_key);
675                    let (margin_init, margin_maint, maker_fee, taker_fee) = cached
676                        .as_ref()
677                        .map_or((None, None, None, None), |instrument| {
678                            extract_fees_from_cached_instrument(instrument)
679                        });
680                    let status_action = okx_status_to_market_action(okx_inst.state);
681                    let is_live = matches!(okx_inst.state, OKXInstrumentStatus::Live);
682                    match parse_instrument_any(
683                        &okx_inst,
684                        margin_init,
685                        margin_maint,
686                        maker_fee,
687                        taker_fee,
688                        ts_init,
689                    ) {
690                        Ok(Some(inst_any)) => {
691                            let instrument_id = inst_any.id();
692                            let is_new_or_changed = cached.is_none_or(|cached| {
693                                !instrument_definitions_match(&cached, &inst_any)
694                            });
695
696                            if is_new_or_changed
697                                && definition_in_scope(config, &okx_inst, &inst_any)
698                            {
699                                publish_instrument_updates(
700                                    std::slice::from_ref(&inst_any),
701                                    instruments_by_symbol,
702                                    http_client,
703                                    recovery_ws,
704                                    business_ws,
705                                    instrument_update_lock,
706                                    data_sender,
707                                );
708                            }
709
710                            emit_instrument_status(
711                                data_sender,
712                                instrument_id,
713                                status_action,
714                                is_live,
715                                ts_init,
716                            );
717                        }
718                        Ok(None) => {
719                            let instrument_id = instruments_by_symbol
720                                .get_cloned(&inst_key)
721                                .map_or_else(|| parse_instrument_id(inst_key), |i| i.id());
722                            emit_instrument_status(
723                                data_sender,
724                                instrument_id,
725                                status_action,
726                                is_live,
727                                ts_init,
728                            );
729                        }
730                        Err(e) => {
731                            log::warn!("Failed to parse instrument {}: {e}", okx_inst.inst_id);
732                            let instrument_id = instruments_by_symbol
733                                .get_cloned(&inst_key)
734                                .map_or_else(|| parse_instrument_id(inst_key), |i| i.id());
735                            emit_instrument_status(
736                                data_sender,
737                                instrument_id,
738                                status_action,
739                                is_live,
740                                ts_init,
741                            );
742                        }
743                    }
744                }
745            }
746            OKXWsMessage::Orders(_)
747            | OKXWsMessage::SpreadOrders(_)
748            | OKXWsMessage::AlgoOrders(_)
749            | OKXWsMessage::OrderResponse { .. }
750            | OKXWsMessage::Account(_)
751            | OKXWsMessage::Positions(_)
752            | OKXWsMessage::LiquidationWarnings(_)
753            | OKXWsMessage::SendFailed { .. } => {
754                log::debug!("Ignoring execution message on data client");
755            }
756            OKXWsMessage::SubscriptionFailed {
757                channel,
758                inst_id,
759                code,
760                msg,
761            } => {
762                log::error!(
763                    "OKX rejected {channel:?} subscription for {inst_id:?} \
764                     (code={code}, msg={msg}); no data will flow for this subscription"
765                );
766
767                if let Some(inst_id) = inst_id
768                    && channel.is_book()
769                    && let Some(instrument) = instruments_by_symbol.get_cloned(&inst_id)
770                {
771                    let instrument_id = instrument.id();
772                    book_sync.remove(instrument_id);
773                }
774            }
775            OKXWsMessage::Error(e) => {
776                if should_retry_error_code(&e.code) {
777                    log::warn!("OKX websocket error: {e:?}");
778                } else {
779                    log::error!("OKX websocket error: {e:?}");
780                }
781            }
782            OKXWsMessage::Reconnected => {
783                log::info!("Websocket reconnected");
784
785                if book_channel_scope == BookChannelScope::Public {
786                    book_sync.reset_sequences(book_channels, book_channel_scope);
787                }
788
789                if !snapshot_timeout.is_zero() {
790                    let pending_count = book_sync.seed_pending_snapshots(
791                        book_channels,
792                        book_channel_scope,
793                        snapshot_timeout,
794                        Instant::now(),
795                    );
796
797                    if pending_count > 0 {
798                        spawn_snapshot_health_monitor(book_sync.clone(), tasks, snapshot_timeout);
799                    }
800                }
801            }
802            OKXWsMessage::Authenticated => {
803                log::debug!("Websocket authenticated");
804            }
805        }
806    }
807
808    /// Establishes instrument context and both WebSocket transports.
809    ///
810    /// Any failure leaves partially started transports for [`Self::teardown_transports`].
811    async fn connect_session(&mut self) -> anyhow::Result<()> {
812        // Reset leaves the old generation canceled until this async boundary can drain it
813        if self.transports_started
814            || !self.tasks.is_empty()
815            || !self.tasks.is_open()
816            || self
817                .ws_public
818                .as_ref()
819                .is_some_and(OKXWebSocketClient::has_task)
820            || self
821                .ws_business
822                .as_ref()
823                .is_some_and(OKXWebSocketClient::has_task)
824        {
825            self.teardown_transports().await?;
826        }
827
828        if !self.tasks.is_open() {
829            self.tasks
830                .start_generation()
831                .context("failed to start OKX data task generation")?;
832        }
833        self.transports_started = true;
834
835        let all_instruments = fetch_configured_instruments(&self.http_client, &self.config).await?;
836
837        // Diff before updating the cache so reconnects do not republish
838        // unchanged definitions; the writer tasks start after this point,
839        // so no instrument lock is needed here
840        let changed = changed_definitions(&all_instruments, &self.instruments_by_symbol);
841
842        self.instruments_by_symbol.rcu(|m| {
843            for instrument in &all_instruments {
844                m.insert(instrument.symbol().inner(), instrument.clone());
845            }
846        });
847
848        // Cache both websockets before connecting and before publishing, so
849        // every cache holds a definition before it is emitted
850        let instruments: Vec<_> = self
851            .instruments_by_symbol
852            .load()
853            .values()
854            .cloned()
855            .collect();
856
857        if let Some(ref ws) = self.ws_public {
858            ws.cache_instruments(&instruments);
859        }
860
861        if let Some(ref ws) = self.ws_business {
862            ws.cache_instruments(&instruments);
863        }
864
865        publish_instrument_updates(
866            &changed,
867            &self.instruments_by_symbol,
868            &self.http_client,
869            self.ws_public.as_ref(),
870            self.ws_business.as_ref(),
871            &self.instrument_update_lock,
872            &self.data_sender,
873        );
874
875        let instrument_types = configured_instrument_types(&self.config);
876
877        if let Some(ref mut ws) = self.ws_public {
878            ws.connect()
879                .await
880                .context("failed to connect OKX public websocket")?;
881            ws.wait_until_active(10.0)
882                .await
883                .context("public websocket did not become active")?;
884
885            let stream = ws.stream();
886            let sender = self.data_sender.clone();
887            let insts = self.instruments_by_symbol.clone();
888            let http = self.http_client.clone();
889            let config = self.config.clone();
890            let update_lock = self.instrument_update_lock.clone();
891            let book_channels = self.book_channels.clone();
892            let book_sync = self.book_sync.clone();
893            let recovery_ws = ws.clone();
894            let business_ws = self.ws_business.clone();
895            let idx_map = self.index_ticker_map.clone();
896            let greeks_subs = self.option_greeks_subs.clone();
897            let tasks = self
898                .tasks
899                .spawner()
900                .context("OKX data task admission is closed")?;
901            let task_spawner = tasks.clone();
902            let cancel = tasks.cancellation_token();
903            let snapshot_timeout = Duration::from_secs(self.config.book_snapshot_timeout_secs);
904            let clock = self.clock;
905
906            tasks
907                .spawn(async move {
908                    let mut quote_cache = QuoteCache::new();
909                    let mut funding_cache: AHashMap<Ustr, (Ustr, u64)> = AHashMap::new();
910
911                    pin_mut!(stream);
912
913                    loop {
914                        tokio::select! {
915                            biased;
916                            () = cancel.cancelled() => {
917                                log::debug!("Public websocket stream task cancelled");
918                                break;
919                            }
920                            Some(message) = stream.next() => {
921                                Self::handle_ws_message(
922                                    message,
923                                    &sender,
924                                    &insts,
925                                    &http,
926                                    &config,
927                                    &update_lock,
928                                    &book_channels,
929                                    &book_sync,
930                                    Some(&recovery_ws),
931                                    business_ws.as_ref(),
932                                    &mut quote_cache,
933                                    &mut funding_cache,
934                                    &idx_map,
935                                    &greeks_subs,
936                                    BookChannelScope::Public,
937                                    snapshot_timeout,
938                                    &task_spawner,
939                                    clock,
940                                );
941                            }
942                        }
943                    }
944                })
945                .context("failed to register OKX public WebSocket stream task")?;
946
947            for inst_type in &instrument_types {
948                ws.subscribe_instruments(*inst_type)
949                    .await
950                    .with_context(|| {
951                        format!("failed to subscribe to instrument type {inst_type:?}")
952                    })?;
953            }
954        }
955
956        if let Some(ref mut ws) = self.ws_business {
957            ws.connect()
958                .await
959                .context("failed to connect OKX business websocket")?;
960            ws.wait_until_active(10.0)
961                .await
962                .context("business websocket did not become active")?;
963
964            let stream = ws.stream();
965            let sender = self.data_sender.clone();
966            let insts = self.instruments_by_symbol.clone();
967            let http = self.http_client.clone();
968            let config = self.config.clone();
969            let update_lock = self.instrument_update_lock.clone();
970            let book_channels = self.book_channels.clone();
971            let book_sync = self.book_sync.clone();
972            let business_ws = ws.clone();
973            let idx_map = self.index_ticker_map.clone();
974            let greeks_subs = self.option_greeks_subs.clone();
975            let tasks = self
976                .tasks
977                .spawner()
978                .context("OKX data task admission is closed")?;
979            let task_spawner = tasks.clone();
980            let cancel = tasks.cancellation_token();
981            let snapshot_timeout = Duration::from_secs(self.config.book_snapshot_timeout_secs);
982            let clock = self.clock;
983
984            tasks
985                .spawn(async move {
986                    let mut quote_cache = QuoteCache::new();
987                    let mut funding_cache: AHashMap<Ustr, (Ustr, u64)> = AHashMap::new();
988
989                    pin_mut!(stream);
990
991                    loop {
992                        tokio::select! {
993                            biased;
994                            () = cancel.cancelled() => {
995                                log::debug!("Business websocket stream task cancelled");
996                                break;
997                            }
998                            Some(message) = stream.next() => {
999                                Self::handle_ws_message(
1000                                    message,
1001                                    &sender,
1002                                    &insts,
1003                                    &http,
1004                                    &config,
1005                                    &update_lock,
1006                                    &book_channels,
1007                                    &book_sync,
1008                                    None,
1009                                    Some(&business_ws),
1010                                    &mut quote_cache,
1011                                    &mut funding_cache,
1012                                    &idx_map,
1013                                    &greeks_subs,
1014                                    BookChannelScope::Business,
1015                                    snapshot_timeout,
1016                                    &task_spawner,
1017                                    clock,
1018                                );
1019                            }
1020                        }
1021                    }
1022                })
1023                .context("failed to register OKX business WebSocket stream task")?;
1024        }
1025
1026        self.register_book_health_monitor()?;
1027        self.register_instrument_refresh()?;
1028        Ok(())
1029    }
1030
1031    /// Spawns the periodic instrument reconciliation task, disabled when the
1032    /// configured interval is zero. The handle is tracked in `self.tasks`, so
1033    /// `teardown_transports` joins it and the cancellation token stops it on
1034    /// disconnect, failed connect, stop, and dispose.
1035    fn register_instrument_refresh(&self) -> anyhow::Result<()> {
1036        let minutes = self.config.update_instruments_interval_mins;
1037
1038        if minutes == 0 {
1039            log::debug!("Instrument refresh disabled (update_instruments_interval_mins=0)");
1040            return Ok(());
1041        }
1042
1043        let interval = Duration::from_secs(minutes.saturating_mul(60));
1044        let tasks = self
1045            .tasks
1046            .spawner()
1047            .context("OKX data task admission is closed")?;
1048        let cancel = tasks.cancellation_token();
1049        let http_client = self.http_client.clone();
1050        let config = self.config.clone();
1051        let instruments = self.instruments_by_symbol.clone();
1052        let update_lock = self.instrument_update_lock.clone();
1053        let ws_public = self.ws_public.clone();
1054        let ws_business = self.ws_business.clone();
1055        let data_sender = self.data_sender.clone();
1056        let client_id = self.client_id;
1057
1058        tasks.spawn(async move {
1059            loop {
1060                let sleep = tokio::time::sleep(interval);
1061                tokio::pin!(sleep);
1062
1063                tokio::select! {
1064                    biased;
1065                    () = cancel.cancelled() => break,
1066                    () = &mut sleep => {}
1067                }
1068
1069                let result = tokio::select! {
1070                    biased;
1071                    () = cancel.cancelled() => break,
1072                    result = reconcile_instruments(
1073                        &http_client,
1074                        &config,
1075                        &instruments,
1076                        &update_lock,
1077                        ws_public.as_ref(),
1078                        ws_business.as_ref(),
1079                        &data_sender,
1080                    ) => result,
1081                };
1082
1083                match result {
1084                    Ok(summary) => {
1085                        log::debug!(
1086                            "OKX instruments refreshed: client_id={client_id}, fetched={}, changed={}, missing={}",
1087                            summary.fetched,
1088                            summary.changed,
1089                            summary.missing,
1090                        );
1091                    }
1092                    Err(e) => {
1093                        log::warn!(
1094                            "Failed to refresh OKX instruments: client_id={client_id}, error={e:?}"
1095                        );
1096                    }
1097                }
1098            }
1099
1100            log::debug!("Instrument refresh task cancelled");
1101        })?;
1102        Ok(())
1103    }
1104
1105    /// Cancels stream tasks, closes both WebSocket transports, and clears
1106    /// transport-local subscription bookkeeping.
1107    ///
1108    /// Safe to call after a partially failed connect and idempotent.
1109    async fn teardown_transports(&mut self) -> anyhow::Result<()> {
1110        self.transports_started = false;
1111        self.begin_generation_shutdown();
1112
1113        if let Some(ws) = self.ws_public.as_ref() {
1114            ws.request_close().await;
1115        }
1116
1117        if let Some(ws) = self.ws_business.as_ref() {
1118            ws.request_close().await;
1119        }
1120
1121        let task_result = terminate_tasks(&self.tasks, "OKX data client").await;
1122
1123        let public_result = if let Some(ref mut ws) = self.ws_public {
1124            ws.close().await.context("failed to close public websocket")
1125        } else {
1126            Ok(())
1127        };
1128
1129        let business_result = if let Some(ref mut ws) = self.ws_business {
1130            ws.close()
1131                .await
1132                .context("failed to close business websocket")
1133        } else {
1134            Ok(())
1135        };
1136
1137        self.book_channels.store(AHashMap::new());
1138        self.book_sync.clear();
1139        self.option_greeks_subs
1140            .store(AHashMap::<InstrumentId, AHashSet<OKXGreeksType>>::new());
1141        self.option_summary_family_subs.lock().clear();
1142        self.is_connected.store(false, Ordering::Release);
1143
1144        let mut errors = Vec::new();
1145        if let Err(e) = task_result {
1146            errors.push(e.to_string());
1147        }
1148
1149        if let Err(e) = public_result {
1150            errors.push(e.to_string());
1151        }
1152
1153        if let Err(e) = business_result {
1154            errors.push(e.to_string());
1155        }
1156
1157        if errors.is_empty() {
1158            Ok(())
1159        } else {
1160            anyhow::bail!(errors.join("; "))
1161        }
1162    }
1163}
1164
1165fn handle_book_sequence_outcome(
1166    outcome: BookSequenceOutcome,
1167    instrument_id: InstrumentId,
1168    book_channels: &Arc<AtomicMap<InstrumentId, OKXBookChannel>>,
1169    book_sync: &BookSyncTracker,
1170    recovery_ws: Option<&OKXWebSocketClient>,
1171    snapshot_timeout: Duration,
1172    tasks: &TaskSpawner,
1173) -> bool {
1174    match outcome {
1175        BookSequenceOutcome::Accept => true,
1176        BookSequenceOutcome::Suppress => false,
1177        BookSequenceOutcome::Recover {
1178            last_seq_id,
1179            prev_seq_id,
1180            seq_id,
1181        } => {
1182            log::warn!(
1183                "Book sequence gap for {instrument_id}: last_seq_id={last_seq_id:?}, \
1184                 prev_seq_id={prev_seq_id:?}, seq_id={seq_id}; requesting a fresh snapshot"
1185            );
1186
1187            let Some(channel) = book_channels.get_cloned(&instrument_id) else {
1188                log::warn!("Cannot recover book sequence for unsubscribed {instrument_id}");
1189                return false;
1190            };
1191            let Some(ws) = recovery_ws.cloned() else {
1192                log::error!("No public websocket available to recover book for {instrument_id}");
1193                return false;
1194            };
1195            let channels = Arc::clone(book_channels);
1196            let recovery_cancel = tasks.cancellation_token();
1197
1198            spawn_task(tasks, async move {
1199                if recovery_cancel.is_cancelled()
1200                    || channels.get_cloned(&instrument_id) != Some(channel)
1201                {
1202                    return;
1203                }
1204
1205                if let Err(e) = ws.resubscribe_book_channel(instrument_id, channel).await {
1206                    log::error!("Failed to recover book sequence for {instrument_id}: {e}");
1207                }
1208            });
1209
1210            if !snapshot_timeout.is_zero() {
1211                spawn_snapshot_health_monitor(book_sync.clone(), tasks, snapshot_timeout);
1212            }
1213            false
1214        }
1215    }
1216}
1217
1218/// Guards instrument definitions: serializes diff-update-publish sequences
1219/// between writer tasks, and counts completed update batches so a pass can
1220/// detect a write that raced its fetch and skip publishing a stale snapshot.
1221#[derive(Debug, Default)]
1222struct InstrumentUpdateLock {
1223    mutex: parking_lot::Mutex<()>,
1224    write_seq: AtomicU64,
1225}
1226
1227fn dispatch_parsed_data(
1228    msg: NautilusWsMessage,
1229    data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
1230    instruments_by_symbol: &Arc<AtomicMap<Ustr, InstrumentAny>>,
1231) {
1232    match msg {
1233        NautilusWsMessage::Data(payloads) => {
1234            for data in payloads {
1235                if let Err(e) = data_sender.send(DataEvent::Data(data)) {
1236                    log::error!("Failed to emit data event: {e}");
1237                }
1238            }
1239        }
1240        NautilusWsMessage::Deltas(deltas) => {
1241            let data = Data::Deltas(Box::new(deltas));
1242            if let Err(e) = data_sender.send(DataEvent::Data(data)) {
1243                log::error!("Failed to emit data event: {e}");
1244            }
1245        }
1246        NautilusWsMessage::FundingRates(updates) => {
1247            emit_funding_rates(data_sender, updates);
1248        }
1249        NautilusWsMessage::Instrument(instrument, status) => {
1250            instruments_by_symbol.insert(instrument.symbol().inner(), *instrument);
1251
1252            if let Some(status) = status
1253                && let Err(e) = data_sender.send(DataEvent::InstrumentStatus(status))
1254            {
1255                log::error!("Failed to emit instrument status event: {e}");
1256            }
1257        }
1258        NautilusWsMessage::InstrumentStatus(status) => {
1259            if let Err(e) = data_sender.send(DataEvent::InstrumentStatus(status)) {
1260                log::error!("Failed to emit instrument status event: {e}");
1261            }
1262        }
1263        _ => {}
1264    }
1265}
1266
1267fn emit_funding_rates(
1268    sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
1269    updates: Vec<FundingRateUpdate>,
1270) {
1271    for update in updates {
1272        if let Err(e) = sender.send(DataEvent::FundingRate(update)) {
1273            log::error!("Failed to emit funding rate event: {e}");
1274        }
1275    }
1276}
1277
1278fn emit_instrument_status(
1279    sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
1280    instrument_id: InstrumentId,
1281    status_action: MarketStatusAction,
1282    is_live: bool,
1283    ts_init: UnixNanos,
1284) {
1285    let status = InstrumentStatus::new(
1286        instrument_id,
1287        status_action,
1288        ts_init,
1289        ts_init,
1290        None,
1291        None,
1292        Some(is_live),
1293        None,
1294        None,
1295    );
1296
1297    if let Err(e) = sender.send(DataEvent::InstrumentStatus(status)) {
1298        log::error!("Failed to emit instrument status event: {e}");
1299    }
1300}
1301
1302fn spawn_snapshot_health_monitor(
1303    book_sync: BookSyncTracker,
1304    tasks: &TaskSpawner,
1305    timeout: Duration,
1306) {
1307    let task_cancel = tasks.cancellation_token();
1308    spawn_task(tasks, async move {
1309        tokio::select! {
1310            biased;
1311            () = task_cancel.cancelled() => {}
1312            () = tokio::time::sleep(timeout) => {
1313                handle_book_sync_signals(book_sync.expired_pending_snapshots(Instant::now()));
1314            }
1315        }
1316    });
1317}
1318
1319fn handle_book_sync_signals(signals: Vec<BookSyncSignal>) {
1320    for signal in signals {
1321        match signal.kind {
1322            BookSyncSignalKind::Stale { elapsed } => {
1323                log::warn!(
1324                    "Book feed stale for {}: no update for {:.3}s",
1325                    signal.instrument_id,
1326                    elapsed.as_secs_f64()
1327                );
1328            }
1329            BookSyncSignalKind::SnapshotMissing => {
1330                log::warn!(
1331                    "Book snapshot not received for {} after recovery request",
1332                    signal.instrument_id
1333                );
1334            }
1335        }
1336    }
1337}
1338
1339fn changed_definitions(
1340    fetched: &[InstrumentAny],
1341    instruments_by_symbol: &Arc<AtomicMap<Ustr, InstrumentAny>>,
1342) -> Vec<InstrumentAny> {
1343    fetched
1344        .iter()
1345        .filter(|instrument| {
1346            instruments_by_symbol
1347                .get_cloned(&instrument.symbol().inner())
1348                .is_none_or(|cached| !instrument_definitions_match(&cached, instrument))
1349        })
1350        .cloned()
1351        .collect()
1352}
1353
1354/// Updates the data client cache, HTTP client cache, and both WebSocket
1355/// caches, then bumps the update sequence. Callers serialize diff-update
1356/// sequences through the instrument update lock.
1357fn cache_instrument_updates(
1358    changed: &[InstrumentAny],
1359    instruments_by_symbol: &Arc<AtomicMap<Ustr, InstrumentAny>>,
1360    http_client: &OKXHttpClient,
1361    ws_public: Option<&OKXWebSocketClient>,
1362    ws_business: Option<&OKXWebSocketClient>,
1363    instrument_update_lock: &InstrumentUpdateLock,
1364) {
1365    if changed.is_empty() {
1366        return;
1367    }
1368
1369    instruments_by_symbol.rcu(|m| {
1370        for instrument in changed {
1371            m.insert(instrument.symbol().inner(), instrument.clone());
1372        }
1373    });
1374    http_client.cache_instruments(changed);
1375
1376    if let Some(ws) = ws_public {
1377        ws.cache_instruments(changed);
1378    }
1379
1380    if let Some(ws) = ws_business {
1381        ws.cache_instruments(changed);
1382    }
1383
1384    instrument_update_lock
1385        .write_seq
1386        .fetch_add(1, Ordering::SeqCst);
1387}
1388
1389/// Publishes new or changed definitions as [`DataEvent::Instrument`] after
1390/// updating every cache, so consumers never observe a definition the caches
1391/// do not yet hold. Callers serialize diff-update-publish sequences through
1392/// the instrument update lock.
1393fn publish_instrument_updates(
1394    changed: &[InstrumentAny],
1395    instruments_by_symbol: &Arc<AtomicMap<Ustr, InstrumentAny>>,
1396    http_client: &OKXHttpClient,
1397    ws_public: Option<&OKXWebSocketClient>,
1398    ws_business: Option<&OKXWebSocketClient>,
1399    instrument_update_lock: &InstrumentUpdateLock,
1400    data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
1401) {
1402    cache_instrument_updates(
1403        changed,
1404        instruments_by_symbol,
1405        http_client,
1406        ws_public,
1407        ws_business,
1408        instrument_update_lock,
1409    );
1410
1411    for instrument in changed {
1412        if let Err(e) = data_sender.send(DataEvent::Instrument(instrument.clone())) {
1413            log::error!("Failed to emit instrument event: {e}");
1414        }
1415    }
1416}
1417
1418fn contract_filter_with_config(config: &OKXDataClientConfig, instrument: &InstrumentAny) -> bool {
1419    contract_filter_with_config_types(config.contract_types.as_ref(), instrument)
1420}
1421
1422/// Returns `true` when a venue definition belongs to the configured scope:
1423/// the contract type filter, plus configured families for derivative types.
1424/// The instruments channel pushes the whole type, so updates outside the
1425/// configured families must not enter the cache or publish downstream.
1426fn definition_in_scope(
1427    config: &OKXDataClientConfig,
1428    okx_inst: &OKXInstrument,
1429    instrument: &InstrumentAny,
1430) -> bool {
1431    if !contract_filter_with_config(config, instrument) {
1432        return false;
1433    }
1434
1435    let Some(families) = &config.instrument_families else {
1436        return true;
1437    };
1438
1439    if families.is_empty()
1440        || !matches!(
1441            okx_inst.inst_type,
1442            OKXInstrumentType::Option
1443                | OKXInstrumentType::Futures
1444                | OKXInstrumentType::Swap
1445                | OKXInstrumentType::Events
1446        )
1447    {
1448        return true;
1449    }
1450
1451    let family_key = if matches!(okx_inst.inst_type, OKXInstrumentType::Events) {
1452        // Events carry their family as the series ID, matching the REST path
1453        // which passes configured families as series_id
1454        okx_inst.series_id.map(|series| series.as_str())
1455    } else {
1456        Some(okx_inst.inst_family.as_str())
1457    };
1458
1459    let Some(family_key) = family_key else {
1460        return false;
1461    };
1462
1463    families.iter().any(|family| family.as_str() == family_key)
1464}
1465
1466fn contract_filter_with_config_types(
1467    contract_types: Option<&Vec<OKXContractType>>,
1468    instrument: &InstrumentAny,
1469) -> bool {
1470    match contract_types {
1471        None => true,
1472        Some(filter) if filter.is_empty() => true,
1473        Some(filter) => {
1474            let is_inverse = instrument.is_inverse();
1475            (is_inverse && filter.contains(&OKXContractType::Inverse))
1476                || (!is_inverse && filter.contains(&OKXContractType::Linear))
1477        }
1478    }
1479}
1480
1481fn configured_instrument_types(config: &OKXDataClientConfig) -> Vec<OKXInstrumentType> {
1482    if config.instrument_types.is_empty() {
1483        vec![OKXInstrumentType::Spot]
1484    } else {
1485        // A type configured twice must not fetch or publish its instruments twice
1486        let mut seen = AHashSet::new();
1487        config
1488            .instrument_types
1489            .iter()
1490            .filter(|inst_type| seen.insert(**inst_type))
1491            .copied()
1492            .collect()
1493    }
1494}
1495
1496/// Fetches every instrument covered by the configuration, applying the
1497/// contract type filter. Fails on the first type or family error; a spread
1498/// endpoint failure is logged and skipped because spread instruments are
1499/// supplemental. Does not touch any cache: a caller may discard a stale
1500/// snapshot, so cache updates happen only in guarded publish sections.
1501async fn fetch_configured_instruments(
1502    http_client: &OKXHttpClient,
1503    config: &OKXDataClientConfig,
1504) -> anyhow::Result<Vec<InstrumentAny>> {
1505    let instrument_types = configured_instrument_types(config);
1506    let mut all_instruments = Vec::new();
1507
1508    for inst_type in &instrument_types {
1509        let Some(mut families) =
1510            resolve_instrument_families(&config.instrument_families, *inst_type)
1511        else {
1512            continue;
1513        };
1514
1515        // A family configured twice must not fetch or publish its instruments twice
1516        let mut seen = AHashSet::new();
1517        families.retain(|family| seen.insert(family.clone()));
1518
1519        if families.is_empty() {
1520            let (mut fetched, _inst_id_codes) = http_client
1521                .request_instruments(*inst_type, None)
1522                .await
1523                .with_context(|| format!("failed to request OKX instruments for {inst_type:?}"))?;
1524
1525            fetched.retain(|instrument| contract_filter_with_config(config, instrument));
1526            all_instruments.extend(fetched);
1527        } else {
1528            for family in &families {
1529                let (mut fetched, _inst_id_codes) = http_client
1530                    .request_instruments(*inst_type, Some(family.clone()))
1531                    .await
1532                    .with_context(|| {
1533                        format!(
1534                            "failed to request OKX instruments for {inst_type:?} family {family}"
1535                        )
1536                    })?;
1537
1538                fetched.retain(|instrument| contract_filter_with_config(config, instrument));
1539                all_instruments.extend(fetched);
1540            }
1541        }
1542    }
1543
1544    if config.load_spreads {
1545        match http_client
1546            .request_spread_instruments(GetSpreadsParams {
1547                state: Some("live".to_string()),
1548                ..Default::default()
1549            })
1550            .await
1551        {
1552            Ok(mut fetched) => {
1553                fetched.retain(|instrument| contract_filter_with_config(config, instrument));
1554                all_instruments.extend(fetched);
1555            }
1556            Err(e) => {
1557                log::error!("Failed to fetch OKX spread instruments: {e:?}");
1558            }
1559        }
1560    }
1561
1562    Ok(all_instruments)
1563}
1564
1565/// Returns `true` when two instruments carry the same tradable definition,
1566/// ignoring event timestamps.
1567///
1568/// Comparison runs on the serialized form so every venue field, including
1569/// the free-form `info` metadata, participates without listing each field.
1570fn instrument_definitions_match(a: &InstrumentAny, b: &InstrumentAny) -> bool {
1571    fn normalized(instrument: &InstrumentAny) -> Option<serde_json::Value> {
1572        let mut value = serde_json::to_value(instrument).ok()?;
1573
1574        if let Some(definition) = value
1575            .as_object_mut()
1576            .and_then(|obj| obj.values_mut().next())
1577            .and_then(serde_json::Value::as_object_mut)
1578        {
1579            definition.remove("ts_event");
1580            definition.remove("ts_init");
1581        }
1582
1583        Some(value)
1584    }
1585
1586    // A serialization failure compares as changed so updates are never suppressed
1587    match (normalized(a), normalized(b)) {
1588        (Some(a), Some(b)) => a == b,
1589        _ => false,
1590    }
1591}
1592
1593/// Summary of a single instrument reconciliation pass.
1594#[derive(Debug)]
1595struct InstrumentReconciliation {
1596    /// Instruments returned by the REST API after filtering.
1597    fetched: usize,
1598    /// New or materially changed definitions published downstream.
1599    changed: usize,
1600    /// Cached instruments absent from the REST response, retained in place.
1601    missing: usize,
1602}
1603
1604/// Reconciles the instrument cache against the REST API.
1605///
1606/// Fetches every configured instrument type and family, plus spread instruments
1607/// when `load_spreads` is set, then updates the data client, HTTP client, and
1608/// WebSocket caches with new or materially changed definitions before
1609/// publishing them as [`DataEvent::Instrument`]. Unchanged definitions are
1610/// not republished. Cached instruments missing from the response are retained
1611/// because they may still back open subscriptions; the instruments WebSocket
1612/// channel communicates state changes such as suspension or delisting.
1613async fn reconcile_instruments(
1614    http_client: &OKXHttpClient,
1615    config: &OKXDataClientConfig,
1616    instruments_by_symbol: &Arc<AtomicMap<Ustr, InstrumentAny>>,
1617    instrument_update_lock: &InstrumentUpdateLock,
1618    ws_public: Option<&OKXWebSocketClient>,
1619    ws_business: Option<&OKXWebSocketClient>,
1620    data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
1621) -> anyhow::Result<InstrumentReconciliation> {
1622    let seq_before = instrument_update_lock.write_seq.load(Ordering::SeqCst);
1623    let fetched = fetch_configured_instruments(http_client, config).await?;
1624
1625    // Hold the instrument lock from the diff through publication so a concurrent
1626    // instruments channel update cannot interleave with this pass
1627    let _update_guard = instrument_update_lock.mutex.lock();
1628
1629    // A write during the fetch means the snapshot is stale relative to the
1630    // instrument cache; skip publishing it and let the next pass reconcile fully
1631    let changed = if instrument_update_lock.write_seq.load(Ordering::SeqCst) == seq_before {
1632        changed_definitions(&fetched, instruments_by_symbol)
1633    } else {
1634        log::debug!("OKX instrument cache changed during refresh fetch, skipping publish");
1635        Vec::new()
1636    };
1637
1638    if !changed.is_empty() {
1639        publish_instrument_updates(
1640            &changed,
1641            instruments_by_symbol,
1642            http_client,
1643            ws_public,
1644            ws_business,
1645            instrument_update_lock,
1646            data_sender,
1647        );
1648    }
1649
1650    let fetched_symbols: AHashSet<Ustr> = fetched
1651        .iter()
1652        .map(|instrument| instrument.symbol().inner())
1653        .collect();
1654    let missing = instruments_by_symbol
1655        .load()
1656        .keys()
1657        .filter(|symbol| !fetched_symbols.contains(*symbol))
1658        .count();
1659
1660    if missing > 0 {
1661        log::debug!(
1662            "{missing} cached instruments absent from OKX REST response, retaining cached definitions"
1663        );
1664    }
1665
1666    Ok(InstrumentReconciliation {
1667        fetched: fetched.len(),
1668        changed: changed.len(),
1669        missing,
1670    })
1671}
1672
1673#[async_trait::async_trait(?Send)]
1674impl DataClient for OKXDataClient {
1675    fn client_id(&self) -> ClientId {
1676        self.client_id
1677    }
1678
1679    fn venue(&self) -> Option<Venue> {
1680        Some(self.venue())
1681    }
1682
1683    fn start(&mut self) -> anyhow::Result<()> {
1684        log::info!(
1685            "Started: client_id={}, vip_level={:?}, instrument_types={:?}, environment={}, proxy_url={:?}",
1686            self.client_id,
1687            self.vip_level(),
1688            self.config.instrument_types,
1689            self.config.environment,
1690            self.config.proxy_url,
1691        );
1692        Ok(())
1693    }
1694
1695    fn stop(&mut self) -> anyhow::Result<()> {
1696        log::info!("Stopping {id}", id = self.client_id);
1697        self.begin_generation_shutdown();
1698        Ok(())
1699    }
1700
1701    fn reset(&mut self) -> anyhow::Result<()> {
1702        log::debug!("Resetting {id}", id = self.client_id);
1703        self.begin_generation_shutdown();
1704        self.book_channels.store(AHashMap::new());
1705        self.book_sync.clear();
1706        self.option_greeks_subs
1707            .store(AHashMap::<InstrumentId, AHashSet<OKXGreeksType>>::new());
1708        self.option_summary_family_subs.lock().clear();
1709        Ok(())
1710    }
1711
1712    fn dispose(&mut self) -> anyhow::Result<()> {
1713        log::debug!("Disposing {id}", id = self.client_id);
1714        self.begin_generation_shutdown();
1715        Ok(())
1716    }
1717
1718    async fn connect(&mut self) -> anyhow::Result<()> {
1719        if self.is_connected() && self.tasks.is_open() {
1720            return Ok(());
1721        }
1722
1723        let ws_public = self.ws_public.clone();
1724        let ws_business = self.ws_business.clone();
1725        let setup_guard = TaskGroupGuard::new(&[&self.tasks], move || {
1726            if let Some(ws) = ws_public {
1727                ws.begin_shutdown();
1728            }
1729
1730            if let Some(ws) = ws_business {
1731                ws.begin_shutdown();
1732            }
1733        });
1734
1735        if let Err(e) = self.connect_session().await {
1736            if let Err(teardown_error) = self.teardown_transports().await {
1737                return Err(e.context(format!(
1738                    "OKX data startup teardown failed: {teardown_error}"
1739                )));
1740            }
1741            return Err(e);
1742        }
1743
1744        self.is_connected.store(true, Ordering::Release);
1745        setup_guard.disarm();
1746        log::info!("Connected: client_id={}", self.client_id);
1747        Ok(())
1748    }
1749
1750    async fn disconnect(&mut self) -> anyhow::Result<()> {
1751        if self.is_disconnected()
1752            && !self.transports_started
1753            && self.tasks.is_empty()
1754            && self.ws_public.as_ref().is_none_or(|ws| !ws.has_task())
1755            && self.ws_business.as_ref().is_none_or(|ws| !ws.has_task())
1756        {
1757            return Ok(());
1758        }
1759
1760        if !self.is_disconnected() {
1761            if let Some(ref ws) = self.ws_public
1762                && let Err(e) = ws.unsubscribe_all().await
1763            {
1764                log::warn!("Failed to unsubscribe all from public websocket: {e:?}");
1765            }
1766
1767            if let Some(ref ws) = self.ws_business
1768                && let Err(e) = ws.unsubscribe_all().await
1769            {
1770                log::warn!("Failed to unsubscribe all from business websocket: {e:?}");
1771            }
1772
1773            // Allow time for unsubscribe confirmations
1774            tokio::time::sleep(Duration::from_millis(500)).await;
1775        }
1776
1777        self.begin_generation_shutdown();
1778        self.teardown_transports().await?;
1779        log::info!("Disconnected: client_id={}", self.client_id);
1780        Ok(())
1781    }
1782
1783    fn is_connected(&self) -> bool {
1784        self.is_connected.load(Ordering::Relaxed)
1785    }
1786
1787    fn is_disconnected(&self) -> bool {
1788        !self.is_connected()
1789    }
1790
1791    fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
1792        for inst_type in &self.config.instrument_types {
1793            let ws = self.public_ws()?.clone();
1794            let inst_type = *inst_type;
1795
1796            self.spawn_ws(
1797                async move {
1798                    ws.subscribe_instruments(inst_type)
1799                        .await
1800                        .context("instruments subscription")?;
1801                    Ok(())
1802                },
1803                "subscribe_instruments",
1804            );
1805        }
1806        Ok(())
1807    }
1808
1809    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
1810        // OKX instruments channel doesn't support subscribing to individual instruments via instId
1811        // Instead, subscribe to the instrument type if not already subscribed
1812        let instrument_id = cmd.instrument_id;
1813        let ws = self.public_ws()?.clone();
1814
1815        self.spawn_ws(
1816            async move {
1817                ws.subscribe_instrument(instrument_id)
1818                    .await
1819                    .context("instrument type subscription")?;
1820                Ok(())
1821            },
1822            "subscribe_instrument",
1823        );
1824        Ok(())
1825    }
1826
1827    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
1828        if cmd.book_type != BookType::L2_MBP {
1829            anyhow::bail!("OKX only supports L2_MBP order book deltas");
1830        }
1831
1832        if is_okx_spread_symbol(cmd.instrument_id.symbol.as_str()) {
1833            // Spreads have no incremental book channel; sprd-books5 pushes a full
1834            // 5-level snapshot, emitted as F_SNAPSHOT deltas to feed the book.
1835            let instrument_id = cmd.instrument_id;
1836            let ws = self.business_ws()?.clone();
1837            let book_channels = Arc::clone(&self.book_channels);
1838            let book_sync = self.book_sync.clone();
1839            self.spawn_ws(
1840                async move {
1841                    ws.subscribe_spread_book(instrument_id)
1842                        .await
1843                        .context("spread book subscription")?;
1844                    book_channels.insert(instrument_id, OKXBookChannel::SprdBooks5);
1845                    book_sync.record_subscription(instrument_id, Instant::now());
1846                    Ok(())
1847                },
1848                "spread book subscription",
1849            );
1850            return Ok(());
1851        }
1852
1853        let raw_depth = cmd.depth.map_or(0, |d| d.get());
1854        let depth = resolve_book_depth(raw_depth);
1855        if depth != raw_depth {
1856            log::debug!("Clamped book depth {raw_depth} to {depth} (OKX supports 50 or 400)");
1857        }
1858
1859        let rpi = cmd
1860            .params
1861            .as_ref()
1862            .and_then(|params| params.get_bool("rpi"))
1863            .unwrap_or(false);
1864        let vip = self.vip_level().unwrap_or(OKXVipLevel::Vip0);
1865        let channel = if rpi {
1866            OKXBookChannel::BooksRpi
1867        } else {
1868            let channel = select_book_channel(depth, vip);
1869            if depth == 50 && channel == OKXBookChannel::Book {
1870                log::debug!(
1871                    "VIP level {vip} insufficient for 50-depth channel, falling back to default"
1872                );
1873            }
1874            channel
1875        };
1876
1877        let instrument_id = cmd.instrument_id;
1878        let ws = self.public_ws()?.clone();
1879        let book_channels = Arc::clone(&self.book_channels);
1880        let book_sync = self.book_sync.clone();
1881
1882        self.spawn_ws(
1883            async move {
1884                match channel {
1885                    OKXBookChannel::Books50L2Tbt => ws
1886                        .subscribe_book50_l2_tbt(instrument_id)
1887                        .await
1888                        .context("books50-l2-tbt subscription")?,
1889                    OKXBookChannel::BookL2Tbt => ws
1890                        .subscribe_book_l2_tbt(instrument_id)
1891                        .await
1892                        .context("books-l2-tbt subscription")?,
1893                    OKXBookChannel::Book => ws
1894                        .subscribe_books_channel(instrument_id)
1895                        .await
1896                        .context("books subscription")?,
1897                    OKXBookChannel::BooksRpi => ws
1898                        .subscribe_book_rpi(instrument_id)
1899                        .await
1900                        .context("books-rpi subscription")?,
1901                    OKXBookChannel::SprdBooks5 => unreachable!(),
1902                }
1903                book_channels.insert(instrument_id, channel);
1904                book_sync.record_subscription(instrument_id, Instant::now());
1905                Ok(())
1906            },
1907            "order book delta subscription",
1908        );
1909
1910        Ok(())
1911    }
1912
1913    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
1914        let instrument_id = cmd.instrument_id;
1915
1916        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
1917            let ws = self.business_ws()?.clone();
1918            self.spawn_ws(
1919                async move {
1920                    ws.subscribe_spread_quotes(instrument_id)
1921                        .await
1922                        .context("spread quotes subscription")
1923                },
1924                "spread quote subscription",
1925            );
1926            return Ok(());
1927        }
1928
1929        let ws = self.public_ws()?.clone();
1930        self.spawn_ws(
1931            async move {
1932                ws.subscribe_quotes(instrument_id)
1933                    .await
1934                    .context("quotes subscription")
1935            },
1936            "quote subscription",
1937        );
1938        Ok(())
1939    }
1940
1941    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
1942        let instrument_id = cmd.instrument_id;
1943
1944        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
1945            let ws = self.business_ws()?.clone();
1946            self.spawn_ws(
1947                async move {
1948                    ws.subscribe_spread_trades(instrument_id)
1949                        .await
1950                        .context("spread trades subscription")
1951                },
1952                "spread trade subscription",
1953            );
1954            return Ok(());
1955        }
1956
1957        let ws = self.public_ws()?.clone();
1958        self.spawn_ws(
1959            async move {
1960                ws.subscribe_trades(instrument_id, false)
1961                    .await
1962                    .context("trades subscription")
1963            },
1964            "trade subscription",
1965        );
1966        Ok(())
1967    }
1968
1969    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
1970        let ws = self.public_ws()?.clone();
1971        let instrument_id = cmd.instrument_id;
1972
1973        self.spawn_ws(
1974            async move {
1975                ws.subscribe_mark_prices(instrument_id)
1976                    .await
1977                    .context("mark price subscription")
1978            },
1979            "mark price subscription",
1980        );
1981        Ok(())
1982    }
1983
1984    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
1985        let ws = self.public_ws()?.clone();
1986        let instrument_id = cmd.instrument_id;
1987        let symbol = instrument_id.symbol.inner();
1988
1989        let (base, quote) = parse_base_quote_from_symbol(symbol.as_str())?;
1990        let base_pair = Ustr::from(&format!("{base}-{quote}"));
1991        self.index_ticker_map.rcu(|m| {
1992            m.entry(base_pair).or_default().insert(symbol);
1993        });
1994
1995        self.spawn_ws(
1996            async move {
1997                ws.subscribe_index_prices(instrument_id)
1998                    .await
1999                    .context("index price subscription")
2000            },
2001            "index price subscription",
2002        );
2003        Ok(())
2004    }
2005
2006    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
2007        let ws = self.business_ws()?.clone();
2008        let bar_type = cmd.bar_type;
2009
2010        self.spawn_ws(
2011            async move {
2012                ws.subscribe_bars(bar_type)
2013                    .await
2014                    .context("bars subscription")
2015            },
2016            "bar subscription",
2017        );
2018        Ok(())
2019    }
2020
2021    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
2022        let ws = self.public_ws()?.clone();
2023        let instrument_id = cmd.instrument_id;
2024
2025        self.spawn_ws(
2026            async move {
2027                ws.subscribe_funding_rates(instrument_id)
2028                    .await
2029                    .context("funding rate subscription")
2030            },
2031            "funding rate subscription",
2032        );
2033        Ok(())
2034    }
2035
2036    fn subscribe_option_greeks(&mut self, cmd: SubscribeOptionGreeks) -> anyhow::Result<()> {
2037        let instrument_id = cmd.instrument_id;
2038        let conventions = parse_greeks_conventions_from_params(&cmd.params);
2039        self.option_greeks_subs.insert(instrument_id, conventions);
2040
2041        let family = extract_inst_family(instrument_id.symbol.inner().as_str())?;
2042        let is_first = {
2043            let mut family_subs = self.option_summary_family_subs.lock();
2044            let count = family_subs.entry(family).or_default();
2045            *count += 1;
2046            *count == 1
2047        };
2048
2049        if is_first {
2050            let ws = self.public_ws()?.clone();
2051            let family_subs = self.option_summary_family_subs.clone();
2052            self.spawn_ws(
2053                async move {
2054                    let result = ws
2055                        .subscribe_option_summary(family)
2056                        .await
2057                        .context("opt-summary subscription");
2058
2059                    if result.is_err() {
2060                        // Roll back the refcount so a retry can re-arm the subscribe;
2061                        // otherwise the family wedges and Greeks stay dark.
2062                        let mut subs = family_subs.lock();
2063
2064                        if let Some(count) = subs.get_mut(&family) {
2065                            *count = count.saturating_sub(1);
2066                            if *count == 0 {
2067                                subs.remove(&family);
2068                            }
2069                        }
2070                    }
2071                    result
2072                },
2073                "option greeks subscription",
2074            );
2075        }
2076        Ok(())
2077    }
2078
2079    fn subscribe_instrument_status(
2080        &mut self,
2081        cmd: SubscribeInstrumentStatus,
2082    ) -> anyhow::Result<()> {
2083        let ws = self.public_ws()?.clone();
2084        let instrument_id = cmd.instrument_id;
2085
2086        self.spawn_ws(
2087            async move {
2088                ws.subscribe_instrument(instrument_id)
2089                    .await
2090                    .context("instrument status subscription")
2091            },
2092            "instrument status subscription",
2093        );
2094        Ok(())
2095    }
2096
2097    fn unsubscribe_instrument(&mut self, cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
2098        let instrument_id = cmd.instrument_id;
2099        let ws = self.public_ws()?.clone();
2100
2101        self.spawn_ws(
2102            async move {
2103                ws.unsubscribe_instrument(instrument_id)
2104                    .await
2105                    .context("instrument unsubscribe")?;
2106                Ok(())
2107            },
2108            "unsubscribe_instrument",
2109        );
2110        Ok(())
2111    }
2112
2113    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
2114        let instrument_id = cmd.instrument_id;
2115
2116        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
2117            let ws = self.business_ws()?.clone();
2118            self.book_channels.remove(&instrument_id);
2119            self.book_sync.remove(instrument_id);
2120            self.spawn_ws(
2121                async move {
2122                    ws.unsubscribe_spread_book(instrument_id)
2123                        .await
2124                        .context("spread book unsubscribe")
2125                },
2126                "spread book unsubscribe",
2127            );
2128            return Ok(());
2129        }
2130
2131        let ws = self.public_ws()?.clone();
2132        let channel = self.book_channels.get_cloned(&instrument_id);
2133        self.book_channels.remove(&instrument_id);
2134        self.book_sync.remove(instrument_id);
2135
2136        self.spawn_ws(
2137            async move {
2138                match channel {
2139                    Some(OKXBookChannel::Books50L2Tbt) => ws
2140                        .unsubscribe_book50_l2_tbt(instrument_id)
2141                        .await
2142                        .context("books50-l2-tbt unsubscribe")?,
2143                    Some(OKXBookChannel::BookL2Tbt) => ws
2144                        .unsubscribe_book_l2_tbt(instrument_id)
2145                        .await
2146                        .context("books-l2-tbt unsubscribe")?,
2147                    Some(OKXBookChannel::Book) => ws
2148                        .unsubscribe_book(instrument_id)
2149                        .await
2150                        .context("book unsubscribe")?,
2151                    Some(OKXBookChannel::BooksRpi) => ws
2152                        .unsubscribe_book_rpi(instrument_id)
2153                        .await
2154                        .context("books-rpi unsubscribe")?,
2155                    Some(OKXBookChannel::SprdBooks5) => ws
2156                        .unsubscribe_book(instrument_id)
2157                        .await
2158                        .context("book unsubscribe")?,
2159                    None => {
2160                        log::warn!(
2161                            "Book channel not found for {instrument_id}; unsubscribing fallback channel"
2162                        );
2163                        ws.unsubscribe_book(instrument_id)
2164                            .await
2165                            .context("book fallback unsubscribe")?;
2166                    }
2167                }
2168                Ok(())
2169            },
2170            "order book unsubscribe",
2171        );
2172        Ok(())
2173    }
2174
2175    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
2176        let instrument_id = cmd.instrument_id;
2177
2178        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
2179            let ws = self.business_ws()?.clone();
2180            self.spawn_ws(
2181                async move {
2182                    ws.unsubscribe_spread_quotes(instrument_id)
2183                        .await
2184                        .context("spread quotes unsubscribe")
2185                },
2186                "spread quote unsubscribe",
2187            );
2188            return Ok(());
2189        }
2190
2191        let ws = self.public_ws()?.clone();
2192        self.spawn_ws(
2193            async move {
2194                ws.unsubscribe_quotes(instrument_id)
2195                    .await
2196                    .context("quotes unsubscribe")
2197            },
2198            "quote unsubscribe",
2199        );
2200        Ok(())
2201    }
2202
2203    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
2204        let instrument_id = cmd.instrument_id;
2205
2206        if is_okx_spread_symbol(instrument_id.symbol.as_str()) {
2207            let ws = self.business_ws()?.clone();
2208            self.spawn_ws(
2209                async move {
2210                    ws.unsubscribe_spread_trades(instrument_id)
2211                        .await
2212                        .context("spread trades unsubscribe")
2213                },
2214                "spread trade unsubscribe",
2215            );
2216            return Ok(());
2217        }
2218
2219        let ws = self.public_ws()?.clone();
2220        self.spawn_ws(
2221            async move {
2222                ws.unsubscribe_trades(instrument_id, false) // TODO: Aggregated trades?
2223                    .await
2224                    .context("trades unsubscribe")
2225            },
2226            "trade unsubscribe",
2227        );
2228        Ok(())
2229    }
2230
2231    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
2232        let ws = self.public_ws()?.clone();
2233        let instrument_id = cmd.instrument_id;
2234
2235        self.spawn_ws(
2236            async move {
2237                ws.unsubscribe_mark_prices(instrument_id)
2238                    .await
2239                    .context("mark price unsubscribe")
2240            },
2241            "mark price unsubscribe",
2242        );
2243        Ok(())
2244    }
2245
2246    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
2247        let ws = self.public_ws()?.clone();
2248        let instrument_id = cmd.instrument_id;
2249        let symbol = instrument_id.symbol.inner();
2250
2251        // The OKX index-tickers channel is keyed by base pair, so multiple
2252        // instruments on the same pair share one subscription. Per-base-pair
2253        // refcounting lives on the WS client, so we always forward the
2254        // unsubscribe and let the WS layer fire the venue request only when
2255        // it knows the last subscriber dropped. Local routing in
2256        // `index_ticker_map` is still maintained for downstream emit fan-out.
2257        if let Ok((base, quote)) = parse_base_quote_from_symbol(symbol.as_str()) {
2258            let base_pair = Ustr::from(&format!("{base}-{quote}"));
2259            self.index_ticker_map.rcu(|m| {
2260                if let Some(set) = m.get_mut(&base_pair) {
2261                    set.remove(&symbol);
2262                    if set.is_empty() {
2263                        m.remove(&base_pair);
2264                    }
2265                }
2266            });
2267        }
2268
2269        self.spawn_ws(
2270            async move {
2271                ws.unsubscribe_index_prices(instrument_id)
2272                    .await
2273                    .context("index price unsubscribe")
2274            },
2275            "index price unsubscribe",
2276        );
2277        Ok(())
2278    }
2279
2280    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
2281        let ws = self.business_ws()?.clone();
2282        let bar_type = cmd.bar_type;
2283
2284        self.spawn_ws(
2285            async move {
2286                ws.unsubscribe_bars(bar_type)
2287                    .await
2288                    .context("bars unsubscribe")
2289            },
2290            "bar unsubscribe",
2291        );
2292        Ok(())
2293    }
2294
2295    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
2296        let ws = self.public_ws()?.clone();
2297        let instrument_id = cmd.instrument_id;
2298
2299        self.spawn_ws(
2300            async move {
2301                ws.unsubscribe_funding_rates(instrument_id)
2302                    .await
2303                    .context("funding rate unsubscribe")
2304            },
2305            "funding rate unsubscribe",
2306        );
2307        Ok(())
2308    }
2309
2310    fn unsubscribe_option_greeks(&mut self, cmd: &UnsubscribeOptionGreeks) -> anyhow::Result<()> {
2311        let instrument_id = cmd.instrument_id;
2312        self.option_greeks_subs.remove(&instrument_id);
2313
2314        let family = extract_inst_family(instrument_id.symbol.inner().as_str())?;
2315        let should_unsubscribe = {
2316            let mut family_subs = self.option_summary_family_subs.lock();
2317
2318            if let Some(count) = family_subs.get_mut(&family) {
2319                *count = count.saturating_sub(1);
2320                if *count == 0 {
2321                    family_subs.remove(&family);
2322                    true
2323                } else {
2324                    false
2325                }
2326            } else {
2327                false
2328            }
2329        };
2330
2331        if should_unsubscribe {
2332            let ws = self.public_ws()?.clone();
2333            self.spawn_ws(
2334                async move {
2335                    ws.unsubscribe_option_summary(family)
2336                        .await
2337                        .context("opt-summary unsubscription")
2338                },
2339                "option greeks unsubscription",
2340            );
2341        }
2342        Ok(())
2343    }
2344
2345    fn unsubscribe_instrument_status(
2346        &mut self,
2347        cmd: &UnsubscribeInstrumentStatus,
2348    ) -> anyhow::Result<()> {
2349        let ws = self.public_ws()?.clone();
2350        let instrument_id = cmd.instrument_id;
2351
2352        self.spawn_ws(
2353            async move {
2354                ws.unsubscribe_instrument(instrument_id)
2355                    .await
2356                    .context("instrument status unsubscription")
2357            },
2358            "instrument status unsubscription",
2359        );
2360        Ok(())
2361    }
2362
2363    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
2364        let http = self.http_client.clone();
2365        let sender = self.data_sender.clone();
2366        let instruments_cache = self.instruments_by_symbol.clone();
2367        let update_lock = self.instrument_update_lock.clone();
2368        let ws_public = self.ws_public.clone();
2369        let ws_business = self.ws_business.clone();
2370        let request_id = request.request_id;
2371        let client_id = request.client_id.unwrap_or(self.client_id);
2372        let venue = self.venue();
2373        let start = request.start;
2374        let end = request.end;
2375        let params = request.params;
2376        let clock = self.clock;
2377        let start_nanos = datetime_to_unix_nanos(start);
2378        let end_nanos = datetime_to_unix_nanos(end);
2379        let instrument_types = configured_instrument_types(&self.config);
2380        let contract_types = self.config.contract_types.clone();
2381        let instrument_families = self.config.instrument_families.clone();
2382        let load_spreads = self.config.load_spreads;
2383
2384        self.spawn_task(async move {
2385            let seq_before = update_lock.write_seq.load(Ordering::SeqCst);
2386            let mut all_instruments = Vec::new();
2387
2388            for inst_type in instrument_types {
2389                let Some(families) = resolve_instrument_families(&instrument_families, inst_type)
2390                else {
2391                    continue;
2392                };
2393
2394                if families.is_empty() {
2395                    match http.request_instruments(inst_type, None).await {
2396                        Ok((instruments, _inst_id_codes)) => {
2397                            for instrument in instruments {
2398                                if !contract_filter_with_config_types(
2399                                    contract_types.as_ref(),
2400                                    &instrument,
2401                                ) {
2402                                    continue;
2403                                }
2404
2405                                all_instruments.push(instrument);
2406                            }
2407                        }
2408                        Err(e) => {
2409                            log::error!("Failed to fetch instruments for {inst_type:?}: {e:?}");
2410                        }
2411                    }
2412                } else {
2413                    for family in families {
2414                        match http
2415                            .request_instruments(inst_type, Some(family.clone()))
2416                            .await
2417                        {
2418                            Ok((instruments, _inst_id_codes)) => {
2419                                for instrument in instruments {
2420                                    if !contract_filter_with_config_types(
2421                                        contract_types.as_ref(),
2422                                        &instrument,
2423                                    ) {
2424                                        continue;
2425                                    }
2426
2427                                    all_instruments.push(instrument);
2428                                }
2429                            }
2430                            Err(e) => {
2431                                log::error!(
2432                                    "Failed to fetch instruments for {inst_type:?} family {family}: {e:?}"
2433                                );
2434                            }
2435                        }
2436                    }
2437                }
2438            }
2439
2440            if load_spreads {
2441                match http
2442                    .request_spread_instruments(GetSpreadsParams {
2443                        state: Some("live".to_string()),
2444                        ..Default::default()
2445                    })
2446                    .await
2447                {
2448                    Ok(instruments) => {
2449                        for instrument in instruments {
2450                            if !contract_filter_with_config_types(
2451                                contract_types.as_ref(),
2452                                &instrument,
2453                            ) {
2454                                continue;
2455                            }
2456
2457                            all_instruments.push(instrument);
2458                        }
2459                    }
2460                    Err(e) => {
2461                        log::error!("Failed to fetch OKX spread instruments: {e:?}");
2462                    }
2463                }
2464            }
2465
2466            {
2467                let _update_guard = update_lock
2468                    .mutex
2469                    .lock();
2470
2471                if update_lock.write_seq.load(Ordering::SeqCst) == seq_before {
2472                    cache_instrument_updates(
2473                        &all_instruments,
2474                        &instruments_cache,
2475                        &http,
2476                        ws_public.as_ref(),
2477                        ws_business.as_ref(),
2478                        &update_lock,
2479                    );
2480                } else {
2481                    log::debug!(
2482                        "OKX instrument cache changed during request fetch, skipping cache update"
2483                    );
2484                }
2485            }
2486
2487            let response = DataResponse::Instruments(InstrumentsResponse::new(
2488                request_id,
2489                client_id,
2490                venue,
2491                all_instruments,
2492                start_nanos,
2493                end_nanos,
2494                clock.get_time_ns(),
2495                params,
2496            ));
2497
2498            if let Err(e) = sender.send(DataEvent::Response(response)) {
2499                log::error!("Failed to send instruments response: {e}");
2500            }
2501        });
2502
2503        Ok(())
2504    }
2505
2506    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
2507        let http = self.http_client.clone();
2508        let sender = self.data_sender.clone();
2509        let instruments = self.instruments_by_symbol.clone();
2510        let update_lock = self.instrument_update_lock.clone();
2511        let ws_public = self.ws_public.clone();
2512        let ws_business = self.ws_business.clone();
2513        let instrument_id = request.instrument_id;
2514        let request_id = request.request_id;
2515        let client_id = request.client_id.unwrap_or(self.client_id);
2516        let start = request.start;
2517        let end = request.end;
2518        let params = request.params;
2519        let clock = self.clock;
2520        let start_nanos = datetime_to_unix_nanos(start);
2521        let end_nanos = datetime_to_unix_nanos(end);
2522        let instrument_types = configured_instrument_types(&self.config);
2523        let contract_types = self.config.contract_types.clone();
2524        let load_spreads = self.config.load_spreads;
2525
2526        self.spawn_task(async move {
2527            let seq_before = update_lock.write_seq.load(Ordering::SeqCst);
2528
2529            match http
2530                .request_instrument(instrument_id)
2531                .await
2532                .context("fetch instrument from API")
2533            {
2534                Ok(instrument) => {
2535                    let inst_id = instrument.id();
2536                    let symbol = inst_id.symbol.as_str();
2537                    if is_okx_spread_symbol(symbol) {
2538                        if !load_spreads {
2539                            log::error!(
2540                                "Instrument {instrument_id} is a spread but load_spreads is false"
2541                            );
2542                            return;
2543                        }
2544                    } else {
2545                        let inst_type = okx_instrument_type_from_symbol(symbol);
2546                        if !instrument_types.contains(&inst_type) {
2547                            log::error!(
2548                                "Instrument {instrument_id} type {inst_type:?} not in configured types {instrument_types:?}"
2549                            );
2550                            return;
2551                        }
2552                    }
2553
2554                    if !contract_filter_with_config_types(contract_types.as_ref(), &instrument) {
2555                        log::error!(
2556                            "Instrument {instrument_id} filtered out by contract_types config"
2557                        );
2558                        return;
2559                    }
2560
2561                    {
2562                        let _update_guard = update_lock
2563                            .mutex
2564                            .lock();
2565
2566                        if update_lock.write_seq.load(Ordering::SeqCst) == seq_before {
2567                            cache_instrument_updates(
2568                                std::slice::from_ref(&instrument),
2569                                &instruments,
2570                                &http,
2571                                ws_public.as_ref(),
2572                                ws_business.as_ref(),
2573                                &update_lock,
2574                            );
2575                        } else {
2576                            log::debug!(
2577                                "OKX instrument cache changed during request fetch, skipping cache update"
2578                            );
2579                        }
2580                    }
2581
2582                    let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
2583                        request_id,
2584                        client_id,
2585                        instrument.id(),
2586                        instrument,
2587                        start_nanos,
2588                        end_nanos,
2589                        clock.get_time_ns(),
2590                        params,
2591                    )));
2592
2593                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2594                        log::error!("Failed to send instrument response: {e}");
2595                    }
2596                }
2597                Err(e) if e.downcast_ref::<OKXInstrumentDefinitionError>().is_some() => {
2598                    log::warn!("Instrument request skipped: {e:?}");
2599                }
2600                Err(e) => log::error!("Instrument request failed: {e:?}"),
2601            }
2602        });
2603
2604        Ok(())
2605    }
2606
2607    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
2608        let http = self.http_client.clone();
2609        let sender = self.data_sender.clone();
2610        let instrument_id = request.instrument_id;
2611        let depth = request.depth.map(|n| n.get() as u32);
2612        let request_id = request.request_id;
2613        let client_id = request.client_id.unwrap_or(self.client_id);
2614        let params = request.params;
2615        let rpi = params
2616            .as_ref()
2617            .and_then(|params| params.get_bool("rpi"))
2618            .unwrap_or(false);
2619        let clock = self.clock;
2620
2621        self.spawn_task(async move {
2622            let result = if rpi {
2623                http.request_rpi_book_snapshot(instrument_id, depth).await
2624            } else {
2625                http.request_book_snapshot(instrument_id, depth).await
2626            };
2627
2628            match result.context("failed to request book snapshot from OKX") {
2629                Ok(book) => {
2630                    let response = DataResponse::Book(BookResponse::new(
2631                        request_id,
2632                        client_id,
2633                        instrument_id,
2634                        book,
2635                        None,
2636                        None,
2637                        clock.get_time_ns(),
2638                        params,
2639                    ));
2640
2641                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2642                        log::error!("Failed to send book snapshot response: {e}");
2643                    }
2644                }
2645                Err(e) => log::error!("Book snapshot request failed: {e:?}"),
2646            }
2647        });
2648
2649        Ok(())
2650    }
2651
2652    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
2653        let http = self.http_client.clone();
2654        let sender = self.data_sender.clone();
2655        let instrument_id = request.instrument_id;
2656        let start = request.start;
2657        let end = request.end;
2658        let limit = request.limit.map(|n| n.get() as u32);
2659        let request_id = request.request_id;
2660        let client_id = request.client_id.unwrap_or(self.client_id);
2661        let params = request.params;
2662        let clock = self.clock;
2663        let start_nanos = datetime_to_unix_nanos(start);
2664        let end_nanos = datetime_to_unix_nanos(end);
2665
2666        self.spawn_task(async move {
2667            match http
2668                .request_trades(instrument_id, start, end, limit)
2669                .await
2670                .context("failed to request trades from OKX")
2671            {
2672                Ok(trades) => {
2673                    let response = DataResponse::Trades(TradesResponse::new(
2674                        request_id,
2675                        client_id,
2676                        instrument_id,
2677                        trades,
2678                        start_nanos,
2679                        end_nanos,
2680                        clock.get_time_ns(),
2681                        params,
2682                    ));
2683
2684                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2685                        log::error!("Failed to send trades response: {e}");
2686                    }
2687                }
2688                Err(e) => log::error!("Trade request failed: {e:?}"),
2689            }
2690        });
2691
2692        Ok(())
2693    }
2694
2695    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
2696        let http = self.http_client.clone();
2697        let sender = self.data_sender.clone();
2698        let bar_type = request.bar_type;
2699        let start = request.start;
2700        let end = request.end;
2701        let limit = request.limit.map(|n| n.get() as u32);
2702        let request_id = request.request_id;
2703        let client_id = request.client_id.unwrap_or(self.client_id);
2704        let params = request.params;
2705        let clock = self.clock;
2706        let start_nanos = datetime_to_unix_nanos(start);
2707        let end_nanos = datetime_to_unix_nanos(end);
2708
2709        self.spawn_task(async move {
2710            match http
2711                .request_bars(bar_type, start, end, limit)
2712                .await
2713                .context("failed to request bars from OKX")
2714            {
2715                Ok(bars) => {
2716                    let response = DataResponse::Bars(BarsResponse::new(
2717                        request_id,
2718                        client_id,
2719                        bar_type,
2720                        bars,
2721                        start_nanos,
2722                        end_nanos,
2723                        clock.get_time_ns(),
2724                        params,
2725                    ));
2726
2727                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2728                        log::error!("Failed to send bars response: {e}");
2729                    }
2730                }
2731                Err(e) => log::error!("Bar request failed: {e:?}"),
2732            }
2733        });
2734
2735        Ok(())
2736    }
2737
2738    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
2739        let http = self.http_client.clone();
2740        let sender = self.data_sender.clone();
2741        let instrument_id = request.instrument_id;
2742        let start = request.start;
2743        let end = request.end;
2744        let limit = request.limit.map(|n| n.get() as u32);
2745        let request_id = request.request_id;
2746        let client_id = request.client_id.unwrap_or(self.client_id);
2747        let params = request.params;
2748        let clock = self.clock;
2749        let start_nanos = datetime_to_unix_nanos(start);
2750        let end_nanos = datetime_to_unix_nanos(end);
2751
2752        self.spawn_task(async move {
2753            match http
2754                .request_funding_rates(instrument_id, start, end, limit)
2755                .await
2756                .context("failed to request funding rates from OKX")
2757            {
2758                Ok(funding_rates) => {
2759                    let response = DataResponse::FundingRates(FundingRatesResponse::new(
2760                        request_id,
2761                        client_id,
2762                        instrument_id,
2763                        funding_rates,
2764                        start_nanos,
2765                        end_nanos,
2766                        clock.get_time_ns(),
2767                        params,
2768                    ));
2769
2770                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2771                        log::error!("Failed to send funding rates response: {e}");
2772                    }
2773                }
2774                Err(e) => log::error!("Funding rates request failed: {e:?}"),
2775            }
2776        });
2777
2778        Ok(())
2779    }
2780
2781    fn request_forward_prices(&self, request: RequestForwardPrices) -> anyhow::Result<()> {
2782        let http = self.http_client.clone();
2783        let sender = self.data_sender.clone();
2784        let underlying = request.underlying.to_string();
2785        let instrument_id = request.instrument_id;
2786        let request_id = request.request_id;
2787        let client_id = request.client_id.unwrap_or(self.client_id);
2788        let params = request.params;
2789        let clock = self.clock;
2790        let venue = *OKX_VENUE;
2791
2792        self.spawn_task(async move {
2793            match http
2794                .request_forward_prices(&underlying, instrument_id)
2795                .await
2796                .context("failed to request forward prices from OKX")
2797            {
2798                Ok(forward_prices) => {
2799                    let response = DataResponse::ForwardPrices(ForwardPricesResponse::new(
2800                        request_id,
2801                        client_id,
2802                        venue,
2803                        forward_prices,
2804                        clock.get_time_ns(),
2805                        params,
2806                    ));
2807
2808                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2809                        log::error!("Failed to send forward prices response: {e}");
2810                    }
2811                }
2812                Err(e) => {
2813                    log::error!("Forward prices request failed for {underlying}: {e:?}");
2814                    let response = DataResponse::ForwardPrices(ForwardPricesResponse::new(
2815                        request_id,
2816                        client_id,
2817                        venue,
2818                        Vec::new(),
2819                        clock.get_time_ns(),
2820                        params,
2821                    ));
2822
2823                    if let Err(e) = sender.send(DataEvent::Response(response)) {
2824                        log::error!("Failed to send forward prices response: {e}");
2825                    }
2826                }
2827            }
2828        });
2829
2830        Ok(())
2831    }
2832}
2833
2834/// Resolves the set of [`OKXGreeksType`] conventions for an option greeks subscription.
2835///
2836/// Reads the `greeks_convention` key from `params`, accepting either a single
2837/// [`GreeksConvention`] string (e.g. `"BLACK_SCHOLES"` or `"PRICE_ADJUSTED"`) or a
2838/// JSON array of such strings. Unrecognized entries log a warning and are skipped.
2839/// Returns the default set `{Bs, Pa}` when the key is absent, unparsable, or
2840/// yields no valid entries so every subscription defaults to both conventions.
2841pub(crate) fn parse_greeks_conventions_from_params(
2842    params: &Option<Params>,
2843) -> AHashSet<OKXGreeksType> {
2844    let default_set: AHashSet<OKXGreeksType> =
2845        [OKXGreeksType::Bs, OKXGreeksType::Pa].into_iter().collect();
2846
2847    let Some(value) = params.as_ref().and_then(|p| p.get("greeks_convention")) else {
2848        return default_set;
2849    };
2850
2851    let mut out = AHashSet::new();
2852    match value {
2853        serde_json::Value::String(s) => push_convention_str(&mut out, s),
2854        serde_json::Value::Array(items) => {
2855            for item in items {
2856                if let Some(s) = item.as_str() {
2857                    push_convention_str(&mut out, s);
2858                } else {
2859                    log::warn!("Ignoring non-string greeks_convention entry {item:?}");
2860                }
2861            }
2862        }
2863        other => {
2864            log::warn!(
2865                "Unsupported greeks_convention value {other:?}, defaulting to both conventions"
2866            );
2867        }
2868    }
2869
2870    if out.is_empty() { default_set } else { out }
2871}
2872
2873fn push_convention_str(out: &mut AHashSet<OKXGreeksType>, raw: &str) {
2874    match raw.parse::<GreeksConvention>() {
2875        Ok(convention) => {
2876            out.insert(convention.into());
2877        }
2878        Err(_) => log::warn!("Unrecognized greeks_convention {raw:?}, skipping"),
2879    }
2880}
2881
2882#[cfg(test)]
2883mod tests {
2884    use std::{collections::HashMap, net::SocketAddr, sync::Arc};
2885
2886    use axum::{Router, extract::Query, response::Json, routing::get};
2887    use nautilus_common::{live::runner::replace_data_event_sender, testing::wait_until_async};
2888    use nautilus_core::UUID4;
2889    use nautilus_model::{
2890        identifiers::Symbol,
2891        instruments::stubs::currency_pair_btcusdt,
2892        types::{Price, Quantity},
2893    };
2894    use nautilus_network::websocket::TransportBackend;
2895    use rstest::rstest;
2896    use serde_json::{Value, json};
2897
2898    use super::*;
2899    use crate::{
2900        common::{
2901            consts::OKX_CLIENT_ID, enums::OKXEnvironment, models::OKXInstrument,
2902            testing::load_test_json,
2903        },
2904        websocket::{enums::OKXWsChannel, messages::OKXWsFrame},
2905    };
2906
2907    struct DropSignal(Option<tokio::sync::oneshot::Sender<()>>);
2908
2909    impl Drop for DropSignal {
2910        fn drop(&mut self) {
2911            if let Some(sender) = self.0.take() {
2912                let _ = sender.send(());
2913            }
2914        }
2915    }
2916
2917    #[derive(Clone, Copy)]
2918    enum DataTaskBoundary {
2919        Reset,
2920        Dispose,
2921        RepeatedStop,
2922    }
2923
2924    fn both() -> AHashSet<OKXGreeksType> {
2925        [OKXGreeksType::Bs, OKXGreeksType::Pa].into_iter().collect()
2926    }
2927
2928    fn only(greeks_type: OKXGreeksType) -> AHashSet<OKXGreeksType> {
2929        [greeks_type].into_iter().collect()
2930    }
2931
2932    #[rstest]
2933    fn dispatch_parsed_data_emits_instrument_status() {
2934        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
2935        let instruments_by_symbol = Arc::new(AtomicMap::new());
2936        let status = InstrumentStatus::new(
2937            InstrumentId::from("USDG-SGD.OKX"),
2938            MarketStatusAction::Trading,
2939            UnixNanos::from(1u64),
2940            UnixNanos::from(2u64),
2941            None,
2942            None,
2943            Some(true),
2944            None,
2945            None,
2946        );
2947
2948        dispatch_parsed_data(
2949            NautilusWsMessage::InstrumentStatus(status),
2950            &sender,
2951            &instruments_by_symbol,
2952        );
2953
2954        match receiver.try_recv().expect("instrument status event") {
2955            DataEvent::InstrumentStatus(received) => assert_eq!(received, status),
2956            other => panic!("Expected DataEvent::InstrumentStatus, was {other:?}"),
2957        }
2958        assert!(instruments_by_symbol.load().is_empty());
2959    }
2960
2961    #[rstest]
2962    fn rejected_book_subscription_clears_sync_and_preserves_reconnect_intent() {
2963        let instrument_id = InstrumentId::from("OMI-USD.OKX");
2964        let mut pair = currency_pair_btcusdt();
2965        pair.id = instrument_id;
2966        pair.raw_symbol = Symbol::from("OMI-USD");
2967        let instrument = InstrumentAny::CurrencyPair(pair);
2968
2969        let instruments_by_symbol = Arc::new(AtomicMap::new());
2970        instruments_by_symbol.insert(Ustr::from("OMI-USD"), instrument);
2971        let book_channels = Arc::new(AtomicMap::new());
2972        book_channels.insert(instrument_id, OKXBookChannel::Book);
2973        let book_sync = BookSyncTracker::default();
2974        book_sync.record_subscription(instrument_id, Instant::now());
2975        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
2976        let http = offline_http_client();
2977        let update_lock = InstrumentUpdateLock::default();
2978        let mut quote_cache = QuoteCache::new();
2979        let mut funding_cache: AHashMap<Ustr, (Ustr, u64)> = AHashMap::new();
2980        let index_ticker_map = Arc::new(AtomicMap::new());
2981        let option_greeks_subs = Arc::new(AtomicMap::new());
2982        let task_group = TaskGroup::new();
2983        let tasks = task_group.spawner().expect("task spawner");
2984
2985        OKXDataClient::handle_ws_message(
2986            OKXWsMessage::SubscriptionFailed {
2987                channel: OKXWsChannel::Books,
2988                inst_id: Some(Ustr::from("OMI-USD")),
2989                code: "60018".to_string(),
2990                msg: "Channel does not exist".to_string(),
2991            },
2992            &sender,
2993            &instruments_by_symbol,
2994            &http,
2995            &OKXDataClientConfig::default(),
2996            &update_lock,
2997            &book_channels,
2998            &book_sync,
2999            None,
3000            None,
3001            &mut quote_cache,
3002            &mut funding_cache,
3003            &index_ticker_map,
3004            &option_greeks_subs,
3005            BookChannelScope::Public,
3006            Duration::ZERO,
3007            &tasks,
3008            get_atomic_clock_realtime(),
3009        );
3010
3011        assert_eq!(
3012            book_channels.load().get(&instrument_id),
3013            Some(&OKXBookChannel::Book),
3014            "rejected subscription must preserve the channel selected for reconnect"
3015        );
3016        assert!(
3017            book_sync
3018                .stale_books(Duration::ZERO, Instant::now())
3019                .is_empty(),
3020            "rejected subscription must remove book synchronization state"
3021        );
3022    }
3023
3024    #[rstest]
3025    fn rpi_sequence_gap_suppresses_updates_until_fresh_snapshot() {
3026        let instrument_id = InstrumentId::from("OMI-USD.OKX");
3027        let mut pair = currency_pair_btcusdt();
3028        pair.id = instrument_id;
3029        pair.raw_symbol = Symbol::from("OMI-USD");
3030        pair.price_precision = 7;
3031        pair.size_precision = 3;
3032        pair.price_increment = Price::from("0.0000001");
3033        pair.size_increment = Quantity::from("0.001");
3034        let instrument = InstrumentAny::CurrencyPair(pair);
3035
3036        let instruments_by_symbol = Arc::new(AtomicMap::new());
3037        instruments_by_symbol.insert(Ustr::from("OMI-USD"), instrument);
3038        let book_channels = Arc::new(AtomicMap::new());
3039        book_channels.insert(instrument_id, OKXBookChannel::BooksRpi);
3040        let book_sync = BookSyncTracker::default();
3041        book_sync.record_subscription(instrument_id, Instant::now());
3042        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3043        let http = offline_http_client();
3044        let update_lock = InstrumentUpdateLock::default();
3045        let index_ticker_map = Arc::new(AtomicMap::new());
3046        let option_greeks_subs = Arc::new(AtomicMap::new());
3047        let task_group = TaskGroup::new();
3048        let tasks = task_group.spawner().expect("task spawner");
3049        let mut quote_cache = QuoteCache::new();
3050        let mut funding_cache = AHashMap::new();
3051
3052        let snapshot = rpi_book_message("ws_books_rpi_snapshot.json");
3053        let update = rpi_book_message("ws_books_rpi_update.json");
3054        let mut gap = rpi_book_message("ws_books_rpi_update.json");
3055        let OKXWsMessage::RpiBookData { data, .. } = &mut gap else {
3056            unreachable!()
3057        };
3058        data[0].prev_seq_id -= 1;
3059
3060        let mut handle = |message| {
3061            OKXDataClient::handle_ws_message(
3062                message,
3063                &sender,
3064                &instruments_by_symbol,
3065                &http,
3066                &OKXDataClientConfig::default(),
3067                &update_lock,
3068                &book_channels,
3069                &book_sync,
3070                None,
3071                None,
3072                &mut quote_cache,
3073                &mut funding_cache,
3074                &index_ticker_map,
3075                &option_greeks_subs,
3076                BookChannelScope::Public,
3077                Duration::ZERO,
3078                &tasks,
3079                get_atomic_clock_realtime(),
3080            );
3081        };
3082
3083        handle(snapshot);
3084        assert!(matches!(receiver.try_recv(), Ok(DataEvent::Data(_))));
3085        assert!(matches!(
3086            receiver.try_recv(),
3087            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
3088        ));
3089
3090        handle(gap);
3091        handle(update);
3092        assert!(matches!(
3093            receiver.try_recv(),
3094            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
3095        ));
3096
3097        let mut recovered_snapshot = rpi_book_message("ws_books_rpi_snapshot.json");
3098        let OKXWsMessage::RpiBookData { data, .. } = &mut recovered_snapshot else {
3099            unreachable!()
3100        };
3101        data[0].seq_id = 2_000;
3102        handle(recovered_snapshot);
3103        assert!(matches!(receiver.try_recv(), Ok(DataEvent::Data(_))));
3104        assert!(matches!(
3105            receiver.try_recv(),
3106            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
3107        ));
3108    }
3109
3110    #[rstest]
3111    fn parse_conventions_returns_both_when_params_missing() {
3112        let result = parse_greeks_conventions_from_params(&None);
3113        assert_eq!(result, both());
3114    }
3115
3116    #[rstest]
3117    fn parse_conventions_returns_both_when_key_absent() {
3118        let mut params = Params::new();
3119        params.insert("other_key".to_string(), json!("value"));
3120        let result = parse_greeks_conventions_from_params(&Some(params));
3121        assert_eq!(result, both());
3122    }
3123
3124    #[rstest]
3125    #[case("BLACK_SCHOLES", OKXGreeksType::Bs)]
3126    #[case("PRICE_ADJUSTED", OKXGreeksType::Pa)]
3127    #[case("black_scholes", OKXGreeksType::Bs)]
3128    #[case("price_adjusted", OKXGreeksType::Pa)]
3129    fn parse_conventions_accepts_single_string(#[case] raw: &str, #[case] expected: OKXGreeksType) {
3130        let mut params = Params::new();
3131        params.insert("greeks_convention".to_string(), json!(raw));
3132        let result = parse_greeks_conventions_from_params(&Some(params));
3133        assert_eq!(result, only(expected));
3134    }
3135
3136    #[rstest]
3137    fn parse_conventions_accepts_list_of_strings() {
3138        let mut params = Params::new();
3139        params.insert(
3140            "greeks_convention".to_string(),
3141            json!(["BLACK_SCHOLES", "PRICE_ADJUSTED"]),
3142        );
3143        let result = parse_greeks_conventions_from_params(&Some(params));
3144        assert_eq!(result, both());
3145    }
3146
3147    #[rstest]
3148    fn parse_conventions_accepts_single_entry_list() {
3149        let mut params = Params::new();
3150        params.insert("greeks_convention".to_string(), json!(["PRICE_ADJUSTED"]));
3151        let result = parse_greeks_conventions_from_params(&Some(params));
3152        assert_eq!(result, only(OKXGreeksType::Pa));
3153    }
3154
3155    #[rstest]
3156    fn parse_conventions_deduplicates_list_entries() {
3157        let mut params = Params::new();
3158        params.insert(
3159            "greeks_convention".to_string(),
3160            json!(["BLACK_SCHOLES", "black_scholes"]),
3161        );
3162        let result = parse_greeks_conventions_from_params(&Some(params));
3163        assert_eq!(result, only(OKXGreeksType::Bs));
3164    }
3165
3166    #[rstest]
3167    fn parse_conventions_skips_unknown_list_entries() {
3168        let mut params = Params::new();
3169        params.insert(
3170            "greeks_convention".to_string(),
3171            json!(["BOGUS", "PRICE_ADJUSTED"]),
3172        );
3173        let result = parse_greeks_conventions_from_params(&Some(params));
3174        assert_eq!(result, only(OKXGreeksType::Pa));
3175    }
3176
3177    #[rstest]
3178    fn parse_conventions_falls_back_to_both_on_all_unknown() {
3179        let mut params = Params::new();
3180        params.insert("greeks_convention".to_string(), json!(["BOGUS"]));
3181        let result = parse_greeks_conventions_from_params(&Some(params));
3182        assert_eq!(result, both());
3183    }
3184
3185    #[rstest]
3186    #[case(json!(1))]
3187    #[case(json!(null))]
3188    #[case(json!(true))]
3189    #[case(json!({"nested": "object"}))]
3190    fn parse_conventions_falls_back_on_non_string_value(#[case] value: serde_json::Value) {
3191        let mut params = Params::new();
3192        params.insert("greeks_convention".to_string(), value);
3193        let result = parse_greeks_conventions_from_params(&Some(params));
3194        assert_eq!(result, both());
3195    }
3196
3197    #[rstest]
3198    fn parse_conventions_falls_back_on_unknown_single_string() {
3199        let mut params = Params::new();
3200        params.insert("greeks_convention".to_string(), json!("BOGUS"));
3201        let result = parse_greeks_conventions_from_params(&Some(params));
3202        assert_eq!(result, both());
3203    }
3204
3205    fn rpi_book_message(filename: &str) -> OKXWsMessage {
3206        let frame: OKXWsFrame = serde_json::from_str(&load_test_json(filename)).unwrap();
3207        let OKXWsFrame::RpiBookData { arg, action, data } = frame else {
3208            panic!("expected RPI book data");
3209        };
3210        OKXWsMessage::RpiBookData { arg, action, data }
3211    }
3212
3213    fn swap_definition(tick_sz: &str) -> Value {
3214        json!({
3215            "alias": "",
3216            "baseCcy": "",
3217            "category": "1",
3218            "ctMult": "1",
3219            "ctType": "linear",
3220            "ctVal": "0.01",
3221            "ctValCcy": "BTC",
3222            "expTime": "",
3223            "instFamily": "BTC-USDT",
3224            "instId": "BTC-USDT-SWAP",
3225            "instType": "SWAP",
3226            "lever": "125",
3227            "listTime": "1611916828000",
3228            "lotSz": "1",
3229            "maxIcebergSz": "100000000.0000000000000000",
3230            "maxLmtAmt": "20000000",
3231            "maxLmtSz": "100000000",
3232            "maxMktAmt": "",
3233            "maxMktSz": "30000",
3234            "maxStopSz": "30000",
3235            "maxTriggerSz": "100000000.0000000000000000",
3236            "maxTwapSz": "100000000.0000000000000000",
3237            "minSz": "1",
3238            "optType": "",
3239            "quoteCcy": "",
3240            "ruleType": "normal",
3241            "settleCcy": "USDT",
3242            "state": "live",
3243            "stk": "",
3244            "tickSz": tick_sz,
3245            "uly": "BTC-USDT"
3246        })
3247    }
3248
3249    fn ws_instruments_message(definition: Value) -> OKXWsMessage {
3250        let instrument: OKXInstrument =
3251            serde_json::from_value(definition).expect("valid OKXInstrument");
3252        OKXWsMessage::Instruments(vec![instrument])
3253    }
3254
3255    fn offline_http_client() -> OKXHttpClient {
3256        OKXHttpClient::new(
3257            Some("http://127.0.0.1:9".to_string()),
3258            5,
3259            0,
3260            1,
3261            1,
3262            OKXEnvironment::Live,
3263            None,
3264        )
3265        .expect("http client")
3266    }
3267
3268    fn offline_ws_client() -> OKXWebSocketClient {
3269        OKXWebSocketClient::new(
3270            Some("ws://127.0.0.1:9".to_string()),
3271            None,
3272            None,
3273            None,
3274            None,
3275            Some(OKX_WS_HEARTBEAT_SECS),
3276            None,
3277            TransportBackend::default(),
3278            None,
3279        )
3280        .expect("ws client")
3281    }
3282
3283    fn handle_instruments_message(
3284        sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
3285        instruments_by_symbol: &Arc<AtomicMap<Ustr, InstrumentAny>>,
3286        http_client: &OKXHttpClient,
3287        config: &OKXDataClientConfig,
3288        recovery_ws: Option<&OKXWebSocketClient>,
3289        business_ws: Option<&OKXWebSocketClient>,
3290        message: OKXWsMessage,
3291    ) {
3292        let book_channels = Arc::new(AtomicMap::new());
3293        let book_sync = BookSyncTracker::default();
3294        let update_lock = InstrumentUpdateLock::default();
3295        let mut quote_cache = QuoteCache::new();
3296        let mut funding_cache = AHashMap::new();
3297        let index_ticker_map = Arc::new(AtomicMap::new());
3298        let option_greeks_subs = Arc::new(AtomicMap::new());
3299        let task_group = TaskGroup::new();
3300        let tasks = task_group.spawner().expect("task spawner");
3301
3302        OKXDataClient::handle_ws_message(
3303            message,
3304            sender,
3305            instruments_by_symbol,
3306            http_client,
3307            config,
3308            &update_lock,
3309            &book_channels,
3310            &book_sync,
3311            recovery_ws,
3312            business_ws,
3313            &mut quote_cache,
3314            &mut funding_cache,
3315            &index_ticker_map,
3316            &option_greeks_subs,
3317            BookChannelScope::Public,
3318            Duration::ZERO,
3319            &tasks,
3320            get_atomic_clock_realtime(),
3321        );
3322    }
3323
3324    #[rstest]
3325    fn ws_instruments_publishes_new_definition_and_status() {
3326        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3327        let instruments_by_symbol = Arc::new(AtomicMap::new());
3328        let http = offline_http_client();
3329        let ws_public = offline_ws_client();
3330        let ws_business = offline_ws_client();
3331
3332        handle_instruments_message(
3333            &sender,
3334            &instruments_by_symbol,
3335            &http,
3336            &OKXDataClientConfig::default(),
3337            Some(&ws_public),
3338            Some(&ws_business),
3339            ws_instruments_message(swap_definition("0.1")),
3340        );
3341
3342        match receiver.try_recv().expect("instrument event") {
3343            DataEvent::Instrument(instrument) => {
3344                assert_eq!(instrument.id(), InstrumentId::from("BTC-USDT-SWAP.OKX"));
3345                assert_eq!(instrument.price_increment(), Price::from("0.1"));
3346            }
3347            other => panic!("Expected DataEvent::Instrument, was {other:?}"),
3348        }
3349
3350        match receiver.try_recv().expect("instrument status event") {
3351            DataEvent::InstrumentStatus(status) => {
3352                assert_eq!(
3353                    status.instrument_id,
3354                    InstrumentId::from("BTC-USDT-SWAP.OKX")
3355                );
3356                assert_eq!(status.action, MarketStatusAction::Trading);
3357            }
3358            other => panic!("Expected DataEvent::InstrumentStatus, was {other:?}"),
3359        }
3360        assert!(receiver.try_recv().is_err());
3361
3362        let symbol = Ustr::from("BTC-USDT-SWAP");
3363        let cached = instruments_by_symbol
3364            .get_cloned(&symbol)
3365            .expect("instrument cached in the shared cache");
3366        assert_eq!(cached.id(), InstrumentId::from("BTC-USDT-SWAP.OKX"));
3367        assert_eq!(cached.price_increment(), Price::from("0.1"));
3368        assert!(
3369            http.get_instrument(&symbol).is_some(),
3370            "HTTP client cache must be updated before publishing"
3371        );
3372        assert!(
3373            ws_public.instruments_snapshot().contains_key(&symbol),
3374            "public WebSocket cache must be updated before publishing"
3375        );
3376        assert!(
3377            ws_business.instruments_snapshot().contains_key(&symbol),
3378            "business WebSocket cache must be updated before publishing"
3379        );
3380    }
3381
3382    #[rstest]
3383    fn ws_instruments_unchanged_definition_emits_status_only() {
3384        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3385        let instruments_by_symbol = Arc::new(AtomicMap::new());
3386        let http = offline_http_client();
3387
3388        for _ in 0..2 {
3389            handle_instruments_message(
3390                &sender,
3391                &instruments_by_symbol,
3392                &http,
3393                &OKXDataClientConfig::default(),
3394                None,
3395                None,
3396                ws_instruments_message(swap_definition("0.1")),
3397            );
3398        }
3399
3400        assert!(matches!(receiver.try_recv(), Ok(DataEvent::Instrument(_))));
3401        assert!(matches!(
3402            receiver.try_recv(),
3403            Ok(DataEvent::InstrumentStatus(_))
3404        ));
3405
3406        match receiver
3407            .try_recv()
3408            .expect("status event for repeat definition")
3409        {
3410            DataEvent::InstrumentStatus(status) => {
3411                assert_eq!(
3412                    status.instrument_id,
3413                    InstrumentId::from("BTC-USDT-SWAP.OKX")
3414                );
3415            }
3416            other => panic!("Expected DataEvent::InstrumentStatus, was {other:?}"),
3417        }
3418        assert!(
3419            receiver.try_recv().is_err(),
3420            "unchanged definition must not be republished"
3421        );
3422    }
3423
3424    #[rstest]
3425    fn ws_instruments_changed_definition_is_republished() {
3426        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3427        let instruments_by_symbol = Arc::new(AtomicMap::new());
3428        let http = offline_http_client();
3429
3430        handle_instruments_message(
3431            &sender,
3432            &instruments_by_symbol,
3433            &http,
3434            &OKXDataClientConfig::default(),
3435            None,
3436            None,
3437            ws_instruments_message(swap_definition("0.1")),
3438        );
3439        handle_instruments_message(
3440            &sender,
3441            &instruments_by_symbol,
3442            &http,
3443            &OKXDataClientConfig::default(),
3444            None,
3445            None,
3446            ws_instruments_message(swap_definition("0.5")),
3447        );
3448
3449        assert!(matches!(receiver.try_recv(), Ok(DataEvent::Instrument(_))));
3450        assert!(matches!(
3451            receiver.try_recv(),
3452            Ok(DataEvent::InstrumentStatus(_))
3453        ));
3454
3455        match receiver.try_recv().expect("republished instrument") {
3456            DataEvent::Instrument(instrument) => {
3457                assert_eq!(instrument.id(), InstrumentId::from("BTC-USDT-SWAP.OKX"));
3458                assert_eq!(instrument.price_increment(), Price::from("0.5"));
3459            }
3460            other => panic!("Expected DataEvent::Instrument, was {other:?}"),
3461        }
3462
3463        match receiver
3464            .try_recv()
3465            .expect("status for republished instrument")
3466        {
3467            DataEvent::InstrumentStatus(status) => {
3468                assert_eq!(
3469                    status.instrument_id,
3470                    InstrumentId::from("BTC-USDT-SWAP.OKX")
3471                );
3472            }
3473            other => panic!("Expected DataEvent::InstrumentStatus, was {other:?}"),
3474        }
3475        assert!(receiver.try_recv().is_err());
3476
3477        let cached = instruments_by_symbol
3478            .get_cloned(&Ustr::from("BTC-USDT-SWAP"))
3479            .expect("instrument cached in the shared cache");
3480        assert_eq!(cached.price_increment(), Price::from("0.5"));
3481    }
3482
3483    #[rstest]
3484    fn ws_instruments_invalid_definition_emits_status_only() {
3485        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3486        let instruments_by_symbol = Arc::new(AtomicMap::new());
3487        let http = offline_http_client();
3488        let mut definition = swap_definition("0.1");
3489        definition["uly"] = json!("");
3490
3491        handle_instruments_message(
3492            &sender,
3493            &instruments_by_symbol,
3494            &http,
3495            &OKXDataClientConfig::default(),
3496            None,
3497            None,
3498            ws_instruments_message(definition),
3499        );
3500
3501        match receiver
3502            .try_recv()
3503            .expect("status event for invalid definition")
3504        {
3505            DataEvent::InstrumentStatus(status) => {
3506                assert_eq!(
3507                    status.instrument_id,
3508                    InstrumentId::from("BTC-USDT-SWAP.OKX")
3509                );
3510            }
3511            other => panic!("Expected DataEvent::InstrumentStatus, was {other:?}"),
3512        }
3513        assert!(
3514            receiver.try_recv().is_err(),
3515            "invalid definition must not publish an instrument event"
3516        );
3517        assert!(instruments_by_symbol.load().is_empty());
3518    }
3519
3520    #[rstest]
3521    fn ws_instruments_batch_publishes_each_valid_definition() {
3522        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3523        let instruments_by_symbol = Arc::new(AtomicMap::new());
3524        let http = offline_http_client();
3525        let mut eth_definition = swap_definition("0.01");
3526        eth_definition["instId"] = json!("ETH-USDT-SWAP");
3527        eth_definition["instFamily"] = json!("ETH-USDT");
3528        eth_definition["uly"] = json!("ETH-USDT");
3529        let batch = OKXWsMessage::Instruments(vec![
3530            serde_json::from_value(swap_definition("0.1")).expect("valid OKXInstrument"),
3531            serde_json::from_value(eth_definition).expect("valid OKXInstrument"),
3532        ]);
3533
3534        handle_instruments_message(
3535            &sender,
3536            &instruments_by_symbol,
3537            &http,
3538            &OKXDataClientConfig::default(),
3539            None,
3540            None,
3541            batch,
3542        );
3543
3544        let mut published = Vec::new();
3545        let mut statuses = Vec::new();
3546
3547        while let Ok(event) = receiver.try_recv() {
3548            match event {
3549                DataEvent::Instrument(instrument) => published.push(instrument.id()),
3550                DataEvent::InstrumentStatus(status) => statuses.push(status.instrument_id),
3551                other => panic!("Unexpected event {other:?}"),
3552            }
3553        }
3554
3555        assert_eq!(
3556            published,
3557            vec![
3558                InstrumentId::from("BTC-USDT-SWAP.OKX"),
3559                InstrumentId::from("ETH-USDT-SWAP.OKX")
3560            ],
3561            "every valid batch item must publish its definition"
3562        );
3563        assert_eq!(
3564            statuses,
3565            vec![
3566                InstrumentId::from("BTC-USDT-SWAP.OKX"),
3567                InstrumentId::from("ETH-USDT-SWAP.OKX")
3568            ],
3569            "every batch item must keep its status event"
3570        );
3571        assert_eq!(instruments_by_symbol.load().len(), 2);
3572    }
3573
3574    #[rstest]
3575    fn ws_instruments_respects_contract_type_filter() {
3576        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3577        let instruments_by_symbol = Arc::new(AtomicMap::new());
3578        let http = offline_http_client();
3579        let config = OKXDataClientConfig::builder()
3580            .contract_types(vec![OKXContractType::Inverse])
3581            .build();
3582
3583        handle_instruments_message(
3584            &sender,
3585            &instruments_by_symbol,
3586            &http,
3587            &config,
3588            None,
3589            None,
3590            ws_instruments_message(swap_definition("0.1")),
3591        );
3592
3593        match receiver.try_recv().expect("status event") {
3594            DataEvent::InstrumentStatus(status) => {
3595                assert_eq!(
3596                    status.instrument_id,
3597                    InstrumentId::from("BTC-USDT-SWAP.OKX")
3598                );
3599            }
3600            other => panic!("Expected DataEvent::InstrumentStatus, was {other:?}"),
3601        }
3602        assert!(
3603            receiver.try_recv().is_err(),
3604            "a definition excluded by the contract type filter must not publish"
3605        );
3606        assert!(
3607            instruments_by_symbol.load().is_empty(),
3608            "a filtered definition must not enter the instrument cache"
3609        );
3610
3611        let inverse_item = test_payload("http_get_instruments_swap.json")["data"][0].clone();
3612        assert_eq!(inverse_item["instId"], json!("BTC-USD-SWAP"));
3613        assert_eq!(inverse_item["ctType"], json!("inverse"));
3614        handle_instruments_message(
3615            &sender,
3616            &instruments_by_symbol,
3617            &http,
3618            &config,
3619            None,
3620            None,
3621            ws_instruments_message(inverse_item),
3622        );
3623
3624        match receiver.try_recv().expect("instrument event") {
3625            DataEvent::Instrument(instrument) => {
3626                assert_eq!(instrument.id(), InstrumentId::from("BTC-USD-SWAP.OKX"));
3627            }
3628            other => panic!("Expected DataEvent::Instrument, was {other:?}"),
3629        }
3630        assert!(matches!(
3631            receiver.try_recv(),
3632            Ok(DataEvent::InstrumentStatus(_))
3633        ));
3634        assert_eq!(instruments_by_symbol.load().len(), 1);
3635    }
3636
3637    #[rstest]
3638    fn ws_instruments_respects_family_filter() {
3639        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3640        let instruments_by_symbol = Arc::new(AtomicMap::new());
3641        let http = offline_http_client();
3642        let config = OKXDataClientConfig::builder()
3643            .instrument_types(vec![OKXInstrumentType::Swap])
3644            .instrument_families(vec!["BTC-USDT".to_string()])
3645            .build();
3646        let mut eth_definition = swap_definition("0.01");
3647        eth_definition["instId"] = json!("ETH-USDT-SWAP");
3648        eth_definition["instFamily"] = json!("ETH-USDT");
3649        eth_definition["uly"] = json!("ETH-USDT");
3650
3651        handle_instruments_message(
3652            &sender,
3653            &instruments_by_symbol,
3654            &http,
3655            &config,
3656            None,
3657            None,
3658            ws_instruments_message(eth_definition),
3659        );
3660
3661        match receiver.try_recv().expect("status event") {
3662            DataEvent::InstrumentStatus(status) => {
3663                assert_eq!(
3664                    status.instrument_id,
3665                    InstrumentId::from("ETH-USDT-SWAP.OKX")
3666                );
3667            }
3668            other => panic!("Expected DataEvent::InstrumentStatus, was {other:?}"),
3669        }
3670        assert!(
3671            receiver.try_recv().is_err(),
3672            "a definition outside the configured families must not publish"
3673        );
3674        assert!(instruments_by_symbol.load().is_empty());
3675
3676        handle_instruments_message(
3677            &sender,
3678            &instruments_by_symbol,
3679            &http,
3680            &config,
3681            None,
3682            None,
3683            ws_instruments_message(swap_definition("0.1")),
3684        );
3685
3686        match receiver.try_recv().expect("instrument event") {
3687            DataEvent::Instrument(instrument) => {
3688                assert_eq!(instrument.id(), InstrumentId::from("BTC-USDT-SWAP.OKX"));
3689            }
3690            other => panic!("Expected DataEvent::Instrument, was {other:?}"),
3691        }
3692        assert!(matches!(
3693            receiver.try_recv(),
3694            Ok(DataEvent::InstrumentStatus(_))
3695        ));
3696        assert_eq!(instruments_by_symbol.load().len(), 1);
3697    }
3698
3699    #[tokio::test]
3700    async fn reconcile_fetches_duplicate_configured_families_once() {
3701        let state = RefreshServerState {
3702            instruments_payload: Arc::new(tokio::sync::Mutex::new(test_payload(
3703                "http_get_instruments_swap.json",
3704            ))),
3705            ..RefreshServerState::default()
3706        };
3707        let addr = start_refresh_server(state.clone()).await;
3708        let http = refresh_http_client(addr);
3709        let config = OKXDataClientConfig::builder()
3710            .instrument_types(vec![OKXInstrumentType::Swap])
3711            .instrument_families(vec!["BTC-USD".to_string(), "BTC-USD".to_string()])
3712            .build();
3713        let instruments_by_symbol = Arc::new(AtomicMap::new());
3714        let update_lock = InstrumentUpdateLock::default();
3715        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3716
3717        let summary = reconcile_instruments(
3718            &http,
3719            &config,
3720            &instruments_by_symbol,
3721            &update_lock,
3722            None,
3723            None,
3724            &sender,
3725        )
3726        .await
3727        .expect("reconcile");
3728
3729        let queries = state.instrument_queries.lock().await;
3730        assert_eq!(
3731            queries.len(),
3732            1,
3733            "a duplicated family must be fetched only once"
3734        );
3735        drop(queries);
3736        assert_eq!(summary.fetched, 1);
3737        assert_eq!(summary.changed, 1);
3738        assert_eq!(
3739            instrument_events(&mut receiver).len(),
3740            1,
3741            "a duplicated family must not publish its instruments twice"
3742        );
3743    }
3744
3745    #[tokio::test]
3746    async fn reconcile_fetches_duplicate_configured_types_once() {
3747        let state = RefreshServerState {
3748            instruments_payload: Arc::new(tokio::sync::Mutex::new(test_payload(
3749                "http_get_instruments_swap.json",
3750            ))),
3751            ..RefreshServerState::default()
3752        };
3753        let addr = start_refresh_server(state.clone()).await;
3754        let http = refresh_http_client(addr);
3755        let config = OKXDataClientConfig::builder()
3756            .instrument_types(vec![OKXInstrumentType::Swap, OKXInstrumentType::Swap])
3757            .build();
3758        let instruments_by_symbol = Arc::new(AtomicMap::new());
3759        let update_lock = InstrumentUpdateLock::default();
3760        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3761
3762        let summary = reconcile_instruments(
3763            &http,
3764            &config,
3765            &instruments_by_symbol,
3766            &update_lock,
3767            None,
3768            None,
3769            &sender,
3770        )
3771        .await
3772        .expect("reconcile");
3773
3774        let queries = state.instrument_queries.lock().await;
3775        assert_eq!(
3776            queries.len(),
3777            1,
3778            "a duplicated type must be fetched only once"
3779        );
3780        drop(queries);
3781        assert_eq!(summary.fetched, 3);
3782        assert_eq!(summary.changed, 3);
3783        assert_eq!(
3784            instrument_events(&mut receiver).len(),
3785            3,
3786            "a duplicated type must not publish its instruments twice"
3787        );
3788    }
3789
3790    #[rstest]
3791    fn definition_in_scope_matches_events_family_on_series_id() {
3792        let okx_inst = OKXInstrument {
3793            inst_type: OKXInstrumentType::Events,
3794            inst_id: Ustr::from("BTC-ABOVE-DAILY-260224-1600-65000"),
3795            inst_id_code: Some(1000000001),
3796            uly: Ustr::from(""),
3797            inst_family: Ustr::from(""),
3798            series_id: Some(Ustr::from("BTC-ABOVE-DAILY")),
3799            inst_category: Some(crate::common::enums::OKXInstrumentCategory::Crypto),
3800            init_px_lmt_pct: String::new(),
3801            float_px_lmt_pct: String::new(),
3802            max_px_lmt_pct: String::new(),
3803            base_ccy: Ustr::from(""),
3804            quote_ccy: Ustr::from("USDT"),
3805            settle_ccy: Ustr::from("USDT"),
3806            ct_val: String::new(),
3807            ct_mult: String::new(),
3808            ct_val_ccy: String::new(),
3809            opt_type: crate::common::enums::OKXOptionType::None,
3810            stk: String::new(),
3811            list_time: Some(1769697132335),
3812            exp_time: Some(1769700732335),
3813            lever: String::new(),
3814            tick_sz: "0.001".to_string(),
3815            lot_sz: "1".to_string(),
3816            min_sz: "1".to_string(),
3817            ct_type: OKXContractType::None,
3818            state: OKXInstrumentStatus::Settling,
3819            rule_type: "normal".to_string(),
3820            max_lmt_sz: "1000000".to_string(),
3821            max_mkt_sz: "1000000".to_string(),
3822            max_lmt_amt: String::new(),
3823            max_mkt_amt: String::new(),
3824            max_twap_sz: String::new(),
3825            max_iceberg_sz: String::new(),
3826            max_trigger_sz: String::new(),
3827            max_stop_sz: String::new(),
3828            rpi: None,
3829            rpi_min_level: None,
3830            rpi_min_px_band: None,
3831        };
3832        let instrument = crate::common::parse::parse_event_contract_instrument(
3833            &okx_inst,
3834            None,
3835            None,
3836            None,
3837            None,
3838            UnixNanos::from(1u64),
3839        )
3840        .expect("parse events instrument");
3841        let matching = OKXDataClientConfig::builder()
3842            .instrument_types(vec![OKXInstrumentType::Events])
3843            .instrument_families(vec!["BTC-ABOVE-DAILY".to_string()])
3844            .build();
3845        let other = OKXDataClientConfig::builder()
3846            .instrument_types(vec![OKXInstrumentType::Events])
3847            .instrument_families(vec!["ETH-ABOVE-DAILY".to_string()])
3848            .build();
3849
3850        assert!(definition_in_scope(&matching, &okx_inst, &instrument));
3851        assert!(!definition_in_scope(&other, &okx_inst, &instrument));
3852    }
3853
3854    #[tokio::test]
3855    async fn ws_definition_matching_rest_fetch_is_not_republished() {
3856        let state = RefreshServerState {
3857            instruments_payload: Arc::new(tokio::sync::Mutex::new(test_payload(
3858                "http_get_instruments_swap.json",
3859            ))),
3860            ..RefreshServerState::default()
3861        };
3862        let addr = start_refresh_server(state).await;
3863        let http = refresh_http_client(addr);
3864        let config = OKXDataClientConfig::builder()
3865            .instrument_types(vec![OKXInstrumentType::Swap])
3866            .build();
3867        let instruments_by_symbol = Arc::new(AtomicMap::new());
3868        let update_lock = InstrumentUpdateLock::default();
3869        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
3870
3871        reconcile_instruments(
3872            &http,
3873            &config,
3874            &instruments_by_symbol,
3875            &update_lock,
3876            None,
3877            None,
3878            &sender,
3879        )
3880        .await
3881        .expect("reconcile");
3882        assert_eq!(instrument_events(&mut receiver).len(), 3);
3883
3884        // Feed the same venue definition back through the WebSocket handler:
3885        // cross-source parsing must compare equal, so nothing is republished
3886        let rest_item = test_payload("http_get_instruments_swap.json")["data"][2].clone();
3887        assert_eq!(rest_item["instId"], json!("BTC-USDT-SWAP"));
3888        handle_instruments_message(
3889            &sender,
3890            &instruments_by_symbol,
3891            &http,
3892            &OKXDataClientConfig::default(),
3893            None,
3894            None,
3895            ws_instruments_message(rest_item),
3896        );
3897
3898        match receiver.try_recv().expect("status event") {
3899            DataEvent::InstrumentStatus(status) => {
3900                assert_eq!(
3901                    status.instrument_id,
3902                    InstrumentId::from("BTC-USDT-SWAP.OKX")
3903                );
3904            }
3905            other => panic!("Expected DataEvent::InstrumentStatus, was {other:?}"),
3906        }
3907        assert!(
3908            receiver.try_recv().is_err(),
3909            "a definition identical to the REST fetch must not be republished"
3910        );
3911    }
3912
3913    #[rstest]
3914    fn definitions_match_ignores_event_timestamps() {
3915        let okx_instrument: OKXInstrument =
3916            serde_json::from_value(swap_definition("0.1")).expect("valid OKXInstrument");
3917        let first = parse_instrument_any(
3918            &okx_instrument,
3919            None,
3920            None,
3921            None,
3922            None,
3923            UnixNanos::from(1u64),
3924        )
3925        .expect("parse")
3926        .expect("instrument");
3927        let second = parse_instrument_any(
3928            &okx_instrument,
3929            None,
3930            None,
3931            None,
3932            None,
3933            UnixNanos::from(2u64),
3934        )
3935        .expect("parse")
3936        .expect("instrument");
3937
3938        assert!(instrument_definitions_match(&first, &second));
3939    }
3940
3941    #[rstest]
3942    fn definitions_match_detects_increment_changes() {
3943        let mut pair = currency_pair_btcusdt();
3944        let mut changed = pair.clone();
3945        changed.price_increment = Price::from("0.5");
3946        pair.ts_event = UnixNanos::from(1u64);
3947        changed.ts_event = UnixNanos::from(2u64);
3948
3949        assert!(!instrument_definitions_match(
3950            &InstrumentAny::CurrencyPair(pair),
3951            &InstrumentAny::CurrencyPair(changed),
3952        ));
3953    }
3954
3955    #[rstest]
3956    fn definitions_match_detects_info_changes() {
3957        let pair = currency_pair_btcusdt();
3958        let mut changed = pair.clone();
3959        let mut info = Params::new();
3960        info.insert("okx_rpi_min_level".to_string(), json!(5));
3961        changed.info = Some(info);
3962
3963        assert!(!instrument_definitions_match(
3964            &InstrumentAny::CurrencyPair(pair),
3965            &InstrumentAny::CurrencyPair(changed),
3966        ));
3967    }
3968
3969    #[rstest]
3970    fn definitions_match_detects_id_changes() {
3971        let pair = currency_pair_btcusdt();
3972        let mut other = pair.clone();
3973        other.id = InstrumentId::from("ETH-USDT.OKX");
3974
3975        assert!(!instrument_definitions_match(
3976            &InstrumentAny::CurrencyPair(pair),
3977            &InstrumentAny::CurrencyPair(other),
3978        ));
3979    }
3980
3981    #[derive(Clone, Default)]
3982    struct RefreshServerState {
3983        instruments_payload: Arc<tokio::sync::Mutex<Value>>,
3984        spreads_payload: Arc<tokio::sync::Mutex<Value>>,
3985        instrument_queries: Arc<tokio::sync::Mutex<Vec<HashMap<String, String>>>>,
3986        spread_queries: Arc<tokio::sync::Mutex<Vec<HashMap<String, String>>>>,
3987        fail_instruments: bool,
3988        gate_instruments: Option<Arc<tokio::sync::Semaphore>>,
3989    }
3990
3991    async fn start_refresh_server(state: RefreshServerState) -> SocketAddr {
3992        let instruments_state = state.clone();
3993        let spreads_state = state;
3994
3995        let router =
3996            Router::new()
3997                .route(
3998                    "/api/v5/public/instruments",
3999                    get(move |Query(params): Query<HashMap<String, String>>| {
4000                        let state = instruments_state.clone();
4001                        async move {
4002                            state.instrument_queries.lock().await.push(params.clone());
4003
4004                            if let Some(gate) = &state.gate_instruments {
4005                                gate.acquire()
4006                                    .await
4007                                    .expect("instruments gate open")
4008                                    .forget();
4009                            }
4010
4011                            if state.fail_instruments {
4012                                return (
4013                                    axum::http::StatusCode::INTERNAL_SERVER_ERROR,
4014                                    Json(json!({
4015                                        "code": "50000",
4016                                        "msg": "instruments endpoint unavailable",
4017                                        "data": []
4018                                    })),
4019                                );
4020                            }
4021
4022                            let family = params.get("instFamily").cloned();
4023                            let mut payload = state.instruments_payload.lock().await.clone();
4024
4025                            if let Some(family) = family
4026                                && let Some(data) =
4027                                    payload.get_mut("data").and_then(Value::as_array_mut)
4028                            {
4029                                data.retain(|item| {
4030                                    item.get("instFamily").and_then(Value::as_str)
4031                                        == Some(family.as_str())
4032                                });
4033                            }
4034                            (axum::http::StatusCode::OK, Json(payload))
4035                        }
4036                    }),
4037                )
4038                .route(
4039                    "/api/v5/sprd/spreads",
4040                    get(move |Query(params): Query<HashMap<String, String>>| {
4041                        let state = spreads_state.clone();
4042                        async move {
4043                            state.spread_queries.lock().await.push(params);
4044                            Json(state.spreads_payload.lock().await.clone())
4045                        }
4046                    }),
4047                )
4048                .route(
4049                    "/ws/public",
4050                    get(|ws: axum::extract::ws::WebSocketUpgrade| async move {
4051                        ws.on_upgrade(drain_ws)
4052                    }),
4053                )
4054                .route(
4055                    "/ws/business",
4056                    get(|ws: axum::extract::ws::WebSocketUpgrade| async move {
4057                        ws.on_upgrade(drain_ws)
4058                    }),
4059                );
4060
4061        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
4062            .await
4063            .expect("bind");
4064        let addr = listener.local_addr().expect("local_addr");
4065        tokio::spawn(async move { axum::serve(listener, router).await.expect("serve") });
4066        addr
4067    }
4068
4069    async fn drain_ws(mut socket: axum::extract::ws::WebSocket) {
4070        while socket.next().await.is_some() {}
4071    }
4072
4073    fn test_payload(filename: &str) -> Value {
4074        serde_json::from_str(&load_test_json(filename)).expect("valid json fixture")
4075    }
4076
4077    fn refresh_http_client(addr: SocketAddr) -> OKXHttpClient {
4078        OKXHttpClient::new(
4079            Some(format!("http://{addr}")),
4080            5,
4081            0,
4082            1,
4083            1,
4084            OKXEnvironment::Live,
4085            None,
4086        )
4087        .expect("http client")
4088    }
4089
4090    fn spot_refresh_state() -> RefreshServerState {
4091        RefreshServerState {
4092            instruments_payload: Arc::new(tokio::sync::Mutex::new(test_payload(
4093                "http_get_instruments_spot.json",
4094            ))),
4095            ..RefreshServerState::default()
4096        }
4097    }
4098
4099    fn instrument_events(
4100        receiver: &mut tokio::sync::mpsc::UnboundedReceiver<DataEvent>,
4101    ) -> Vec<InstrumentAny> {
4102        let mut events = Vec::new();
4103        while let Ok(DataEvent::Instrument(instrument)) = receiver.try_recv() {
4104            events.push(instrument);
4105        }
4106        events
4107    }
4108
4109    #[tokio::test]
4110    async fn reconcile_publishes_only_new_or_changed_and_retains_missing() {
4111        let state = spot_refresh_state();
4112        let addr = start_refresh_server(state.clone()).await;
4113        let http = refresh_http_client(addr);
4114        let config = OKXDataClientConfig::default();
4115        let instruments_by_symbol = Arc::new(AtomicMap::new());
4116        let update_lock = InstrumentUpdateLock::default();
4117        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4118
4119        let summary = reconcile_instruments(
4120            &http,
4121            &config,
4122            &instruments_by_symbol,
4123            &update_lock,
4124            None,
4125            None,
4126            &sender,
4127        )
4128        .await
4129        .expect("initial reconcile");
4130        assert_eq!(summary.fetched, 5);
4131        assert_eq!(summary.changed, 5);
4132        assert_eq!(summary.missing, 0);
4133        assert_eq!(instrument_events(&mut receiver).len(), 5);
4134        assert_eq!(instruments_by_symbol.load().len(), 5);
4135
4136        let summary = reconcile_instruments(
4137            &http,
4138            &config,
4139            &instruments_by_symbol,
4140            &update_lock,
4141            None,
4142            None,
4143            &sender,
4144        )
4145        .await
4146        .expect("unchanged reconcile");
4147        assert_eq!(summary.fetched, 5);
4148        assert_eq!(summary.changed, 0);
4149        assert_eq!(summary.missing, 0);
4150        assert!(
4151            instrument_events(&mut receiver).is_empty(),
4152            "unchanged definitions must not be republished"
4153        );
4154
4155        {
4156            let mut payload = state.instruments_payload.lock().await;
4157            payload["data"][0]["tickSz"] = json!("0.5");
4158        }
4159        let summary = reconcile_instruments(
4160            &http,
4161            &config,
4162            &instruments_by_symbol,
4163            &update_lock,
4164            None,
4165            None,
4166            &sender,
4167        )
4168        .await
4169        .expect("changed reconcile");
4170        assert_eq!(summary.changed, 1);
4171        let events = instrument_events(&mut receiver);
4172        assert_eq!(events.len(), 1);
4173        assert_eq!(events[0].id(), InstrumentId::from("BTC-USD.OKX"));
4174        assert_eq!(events[0].price_increment(), Price::from("0.5"));
4175
4176        {
4177            let mut payload = state.instruments_payload.lock().await;
4178            let mut new_instrument = payload["data"][0].clone();
4179            new_instrument["instId"] = json!("ETH-USDT");
4180            new_instrument["baseCcy"] = json!("ETH");
4181            new_instrument["quoteCcy"] = json!("USDT");
4182            payload["data"]
4183                .as_array_mut()
4184                .expect("data array")
4185                .push(new_instrument);
4186        }
4187        let summary = reconcile_instruments(
4188            &http,
4189            &config,
4190            &instruments_by_symbol,
4191            &update_lock,
4192            None,
4193            None,
4194            &sender,
4195        )
4196        .await
4197        .expect("new listing reconcile");
4198        assert_eq!(summary.fetched, 6);
4199        assert_eq!(summary.changed, 1);
4200        let events = instrument_events(&mut receiver);
4201        assert_eq!(events.len(), 1);
4202        assert_eq!(events[0].id(), InstrumentId::from("ETH-USDT.OKX"));
4203        assert_eq!(instruments_by_symbol.load().len(), 6);
4204
4205        {
4206            let mut payload = state.instruments_payload.lock().await;
4207            payload["data"]
4208                .as_array_mut()
4209                .expect("data array")
4210                .remove(0);
4211        }
4212        let summary = reconcile_instruments(
4213            &http,
4214            &config,
4215            &instruments_by_symbol,
4216            &update_lock,
4217            None,
4218            None,
4219            &sender,
4220        )
4221        .await
4222        .expect("removal reconcile");
4223        assert_eq!(summary.fetched, 5);
4224        assert_eq!(summary.changed, 0);
4225        assert_eq!(summary.missing, 1);
4226        assert!(
4227            instrument_events(&mut receiver).is_empty(),
4228            "removed instruments must not publish events"
4229        );
4230        assert!(
4231            instruments_by_symbol
4232                .get_cloned(&Ustr::from("BTC-USD"))
4233                .is_some(),
4234            "removed instruments are retained in the cache"
4235        );
4236    }
4237
4238    #[tokio::test]
4239    async fn reconcile_surfaces_fetch_errors_without_publishing() {
4240        let state = RefreshServerState {
4241            fail_instruments: true,
4242            ..spot_refresh_state()
4243        };
4244        let addr = start_refresh_server(state).await;
4245        let http = refresh_http_client(addr);
4246        let config = OKXDataClientConfig::default();
4247        let instruments_by_symbol = Arc::new(AtomicMap::new());
4248        let update_lock = InstrumentUpdateLock::default();
4249        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4250
4251        let result = reconcile_instruments(
4252            &http,
4253            &config,
4254            &instruments_by_symbol,
4255            &update_lock,
4256            None,
4257            None,
4258            &sender,
4259        )
4260        .await;
4261
4262        assert!(result.is_err(), "fetch failure must surface as an error");
4263        assert!(instruments_by_symbol.load().is_empty());
4264        assert!(
4265            receiver.try_recv().is_err(),
4266            "a failed reconcile must not publish events"
4267        );
4268    }
4269
4270    #[tokio::test]
4271    async fn reconcile_requests_each_configured_family() {
4272        let state = RefreshServerState {
4273            instruments_payload: Arc::new(tokio::sync::Mutex::new(test_payload(
4274                "http_get_instruments_swap.json",
4275            ))),
4276            ..RefreshServerState::default()
4277        };
4278        let addr = start_refresh_server(state.clone()).await;
4279        let http = refresh_http_client(addr);
4280        let config = OKXDataClientConfig::builder()
4281            .instrument_types(vec![OKXInstrumentType::Swap])
4282            .instrument_families(vec!["BTC-USD".to_string(), "BTC-USDT".to_string()])
4283            .build();
4284        let instruments_by_symbol = Arc::new(AtomicMap::new());
4285        let update_lock = InstrumentUpdateLock::default();
4286        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4287
4288        let summary = reconcile_instruments(
4289            &http,
4290            &config,
4291            &instruments_by_symbol,
4292            &update_lock,
4293            None,
4294            None,
4295            &sender,
4296        )
4297        .await
4298        .expect("reconcile");
4299
4300        let queries = state.instrument_queries.lock().await;
4301        let families: Vec<Option<String>> = queries
4302            .iter()
4303            .map(|query| query.get("instFamily").cloned())
4304            .collect();
4305        assert_eq!(queries.len(), 2);
4306        assert!(families.contains(&Some("BTC-USD".to_string())));
4307        assert!(families.contains(&Some("BTC-USDT".to_string())));
4308        drop(queries);
4309
4310        assert_eq!(summary.fetched, 2);
4311        assert_eq!(summary.changed, 2);
4312        let ids: Vec<InstrumentId> = instrument_events(&mut receiver)
4313            .iter()
4314            .map(Instrument::id)
4315            .collect();
4316        assert!(ids.contains(&InstrumentId::from("BTC-USD-SWAP.OKX")));
4317        assert!(ids.contains(&InstrumentId::from("BTC-USDT-SWAP.OKX")));
4318    }
4319
4320    #[rstest]
4321    #[case::inverse_keeps_inverse_only(vec![OKXContractType::Inverse], 1, "BTC-USD-SWAP.OKX")]
4322    #[case::linear_keeps_linear_only(vec![OKXContractType::Linear], 2, "BTC-USDT-SWAP.OKX")]
4323    #[tokio::test]
4324    async fn reconcile_applies_contract_type_filter(
4325        #[case] filter: Vec<OKXContractType>,
4326        #[case] expected_count: usize,
4327        #[case] expected_id: &str,
4328    ) {
4329        let state = RefreshServerState {
4330            instruments_payload: Arc::new(tokio::sync::Mutex::new(test_payload(
4331                "http_get_instruments_swap.json",
4332            ))),
4333            ..RefreshServerState::default()
4334        };
4335        let addr = start_refresh_server(state).await;
4336        let http = refresh_http_client(addr);
4337        let config = OKXDataClientConfig::builder()
4338            .instrument_types(vec![OKXInstrumentType::Swap])
4339            .contract_types(filter)
4340            .build();
4341        let instruments_by_symbol = Arc::new(AtomicMap::new());
4342        let update_lock = InstrumentUpdateLock::default();
4343        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4344
4345        let summary = reconcile_instruments(
4346            &http,
4347            &config,
4348            &instruments_by_symbol,
4349            &update_lock,
4350            None,
4351            None,
4352            &sender,
4353        )
4354        .await
4355        .expect("reconcile");
4356
4357        assert_eq!(summary.fetched, expected_count);
4358        assert_eq!(summary.changed, expected_count);
4359        let events = instrument_events(&mut receiver);
4360        assert_eq!(events.len(), expected_count);
4361        assert!(
4362            events
4363                .iter()
4364                .any(|i| i.id() == InstrumentId::from(expected_id)),
4365            "expected {expected_id} in filtered results"
4366        );
4367    }
4368
4369    #[tokio::test]
4370    async fn reconcile_includes_spreads_when_load_spreads_enabled() {
4371        let state = RefreshServerState {
4372            instruments_payload: Arc::new(tokio::sync::Mutex::new(test_payload(
4373                "http_get_instruments_spot.json",
4374            ))),
4375            spreads_payload: Arc::new(tokio::sync::Mutex::new(test_payload(
4376                "http_get_spreads.json",
4377            ))),
4378            ..RefreshServerState::default()
4379        };
4380        let addr = start_refresh_server(state.clone()).await;
4381        let http = refresh_http_client(addr);
4382        let config = OKXDataClientConfig::builder().load_spreads(true).build();
4383        let instruments_by_symbol = Arc::new(AtomicMap::new());
4384        let update_lock = InstrumentUpdateLock::default();
4385        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4386
4387        let summary = reconcile_instruments(
4388            &http,
4389            &config,
4390            &instruments_by_symbol,
4391            &update_lock,
4392            None,
4393            None,
4394            &sender,
4395        )
4396        .await
4397        .expect("reconcile");
4398
4399        assert_eq!(state.spread_queries.lock().await.len(), 1);
4400        assert_eq!(summary.fetched, 7);
4401        assert_eq!(summary.changed, 7);
4402        let ids: Vec<InstrumentId> = instrument_events(&mut receiver)
4403            .iter()
4404            .map(Instrument::id)
4405            .collect();
4406        assert!(ids.contains(&InstrumentId::from("ETH-USD-SWAP_ETH-USD-231229.OKX")));
4407        assert!(ids.contains(&InstrumentId::from("BTC-USDT_BTC-USDT-SWAP.OKX")));
4408    }
4409
4410    #[tokio::test]
4411    async fn reconcile_updates_all_caches_before_publishing() {
4412        let state = spot_refresh_state();
4413        let addr = start_refresh_server(state).await;
4414        let http = refresh_http_client(addr);
4415        let config = OKXDataClientConfig::default();
4416        let instruments_by_symbol = Arc::new(AtomicMap::new());
4417        let update_lock = Arc::new(InstrumentUpdateLock::default());
4418        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4419        let ws = offline_ws_client();
4420        let ws_business = offline_ws_client();
4421
4422        let instruments_task = instruments_by_symbol.clone();
4423        let update_lock_task = update_lock.clone();
4424        let http_task = http.clone();
4425        let ws_task = ws.clone();
4426        let ws_business_task = ws_business.clone();
4427
4428        let reconcile = tokio::spawn(async move {
4429            reconcile_instruments(
4430                &http_task,
4431                &config,
4432                &instruments_task,
4433                &update_lock_task,
4434                Some(&ws_task),
4435                Some(&ws_business_task),
4436                &sender,
4437            )
4438            .await
4439        });
4440
4441        let event = receiver.recv().await.expect("instrument event");
4442        let DataEvent::Instrument(instrument) = event else {
4443            panic!("Expected DataEvent::Instrument, was {event:?}");
4444        };
4445
4446        assert!(
4447            instruments_by_symbol
4448                .load()
4449                .contains_key(&instrument.symbol().inner()),
4450            "data client cache must be updated before publishing"
4451        );
4452        assert!(
4453            http.get_instrument(&instrument.symbol().inner()).is_some(),
4454            "HTTP client cache must be updated before publishing"
4455        );
4456        assert!(
4457            ws.instruments_snapshot()
4458                .contains_key(&instrument.symbol().inner()),
4459            "public WebSocket cache must be updated before publishing"
4460        );
4461        assert!(
4462            ws_business
4463                .instruments_snapshot()
4464                .contains_key(&instrument.symbol().inner()),
4465            "business WebSocket cache must be updated before publishing"
4466        );
4467        reconcile.await.expect("reconcile task").expect("reconcile");
4468    }
4469
4470    #[tokio::test]
4471    async fn reconcile_skips_publish_when_cache_changes_during_fetch() {
4472        let gate = Arc::new(tokio::sync::Semaphore::new(0));
4473        let state = RefreshServerState {
4474            gate_instruments: Some(gate.clone()),
4475            ..spot_refresh_state()
4476        };
4477        let addr = start_refresh_server(state.clone()).await;
4478        let http = refresh_http_client(addr);
4479        let config = OKXDataClientConfig::default();
4480        let instruments_by_symbol = Arc::new(AtomicMap::new());
4481        let update_lock = Arc::new(InstrumentUpdateLock::default());
4482        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4483
4484        gate.add_permits(1);
4485        let summary = reconcile_instruments(
4486            &http,
4487            &config,
4488            &instruments_by_symbol,
4489            &update_lock,
4490            None,
4491            None,
4492            &sender,
4493        )
4494        .await
4495        .expect("seed reconcile");
4496        assert_eq!(summary.changed, 5);
4497        assert_eq!(instrument_events(&mut receiver).len(), 5);
4498
4499        let reconcile = {
4500            let http = http.clone();
4501            let instruments_by_symbol = instruments_by_symbol.clone();
4502            let update_lock = update_lock.clone();
4503            let sender = sender.clone();
4504
4505            tokio::spawn(async move {
4506                reconcile_instruments(
4507                    &http,
4508                    &config,
4509                    &instruments_by_symbol,
4510                    &update_lock,
4511                    None,
4512                    None,
4513                    &sender,
4514                )
4515                .await
4516            })
4517        };
4518
4519        let deadline = Instant::now() + Duration::from_secs(5);
4520        while state.instrument_queries.lock().await.len() < 2 {
4521            assert!(Instant::now() < deadline, "refresh fetch not in flight");
4522            tokio::time::sleep(Duration::from_millis(10)).await;
4523        }
4524
4525        // A concurrent update lands while the refresh fetch is in flight
4526        let mut v2_item = test_payload("http_get_instruments_spot.json")["data"][0].clone();
4527        v2_item["tickSz"] = json!("0.5");
4528        let v2: OKXInstrument = serde_json::from_value(v2_item).expect("valid OKXInstrument");
4529        let v2 = parse_instrument_any(&v2, None, None, None, None, UnixNanos::from(1u64))
4530            .expect("parse")
4531            .expect("instrument");
4532        {
4533            let _guard = update_lock.mutex.lock();
4534            publish_instrument_updates(
4535                std::slice::from_ref(&v2),
4536                &instruments_by_symbol,
4537                &http,
4538                None,
4539                None,
4540                &update_lock,
4541                &sender,
4542            );
4543        }
4544
4545        match receiver.try_recv().expect("concurrent update event") {
4546            DataEvent::Instrument(instrument) => {
4547                assert_eq!(instrument.price_increment(), Price::from("0.5"));
4548            }
4549            other => panic!("Expected DataEvent::Instrument, was {other:?}"),
4550        }
4551
4552        gate.add_permits(1);
4553        let summary = reconcile.await.expect("reconcile task").expect("reconcile");
4554        assert_eq!(
4555            summary.changed, 0,
4556            "a pass whose snapshot went stale mid-fetch must skip publishing"
4557        );
4558        assert!(
4559            instrument_events(&mut receiver).is_empty(),
4560            "the stale pass must not republish the older definition"
4561        );
4562        let cached = instruments_by_symbol
4563            .get_cloned(&Ustr::from("BTC-USD"))
4564            .expect("instrument cached");
4565        assert_eq!(
4566            cached.price_increment(),
4567            Price::from("0.5"),
4568            "the instrument cache keeps the fresher concurrent definition"
4569        );
4570        assert_eq!(
4571            http.get_instrument(&Ustr::from("BTC-USD"))
4572                .map(|instrument| instrument.price_increment()),
4573            Some(Price::from("0.5")),
4574            "the HTTP cache keeps the fresher concurrent definition"
4575        );
4576    }
4577
4578    #[tokio::test]
4579    async fn spawn_instrument_refresh_skipped_when_interval_zero() {
4580        let config = OKXDataClientConfig::builder()
4581            .update_instruments_interval_mins(0)
4582            .build();
4583        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
4584        replace_data_event_sender(sender);
4585        let client = OKXDataClient::new(*OKX_CLIENT_ID, config).expect("data client");
4586
4587        client.register_instrument_refresh().unwrap();
4588        assert!(client.tasks.is_empty());
4589    }
4590
4591    #[rstest]
4592    #[case::reset(DataTaskBoundary::Reset)]
4593    #[case::dispose(DataTaskBoundary::Dispose)]
4594    #[case::repeated_stop(DataTaskBoundary::RepeatedStop)]
4595    #[tokio::test]
4596    async fn lifecycle_boundary_terminates_owned_data_task(#[case] boundary: DataTaskBoundary) {
4597        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
4598        replace_data_event_sender(sender);
4599        let mut client = OKXDataClient::new(*OKX_CLIENT_ID, OKXDataClientConfig::default())
4600            .expect("data client");
4601
4602        if matches!(boundary, DataTaskBoundary::RepeatedStop) {
4603            client.stop().expect("initial stop");
4604        }
4605
4606        let (drop_tx, drop_rx) = tokio::sync::oneshot::channel();
4607        let signal = DropSignal(Some(drop_tx));
4608        client.spawn_ws(
4609            async move {
4610                let _signal = signal;
4611                std::future::pending::<anyhow::Result<()>>().await
4612            },
4613            "pending lifecycle task",
4614        );
4615
4616        match boundary {
4617            DataTaskBoundary::Reset => client.reset().expect("reset"),
4618            DataTaskBoundary::Dispose => client.dispose().expect("dispose"),
4619            DataTaskBoundary::RepeatedStop => client.stop().expect("repeated stop"),
4620        }
4621
4622        tokio::time::timeout(Duration::from_secs(1), drop_rx)
4623            .await
4624            .expect("lifecycle boundary must drop the owned task")
4625            .expect("drop signal");
4626        terminate_tasks(&client.tasks, "test data client")
4627            .await
4628            .expect("data task terminated");
4629        assert!(client.tasks.is_empty());
4630    }
4631
4632    #[tokio::test]
4633    async fn reset_prevents_in_flight_request_from_publishing() {
4634        let gate = Arc::new(tokio::sync::Semaphore::new(0));
4635        let state = RefreshServerState {
4636            gate_instruments: Some(Arc::clone(&gate)),
4637            ..spot_refresh_state()
4638        };
4639        let addr = start_refresh_server(state.clone()).await;
4640        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4641        replace_data_event_sender(sender);
4642        let config = OKXDataClientConfig {
4643            instrument_types: vec![OKXInstrumentType::Spot],
4644            base_url_http: Some(format!("http://{addr}")),
4645            http_timeout_secs: 5,
4646            max_retries: 0,
4647            retry_delay_initial_ms: 1,
4648            retry_delay_max_ms: 1,
4649            ..OKXDataClientConfig::default()
4650        };
4651        let mut client = OKXDataClient::new(*OKX_CLIENT_ID, config).expect("data client");
4652        let request = RequestInstruments::new(
4653            None,
4654            None,
4655            Some(*OKX_CLIENT_ID),
4656            None,
4657            UUID4::new(),
4658            UnixNanos::default(),
4659            None,
4660        );
4661
4662        client
4663            .request_instruments(request)
4664            .expect("request instruments");
4665        wait_until_async(
4666            || {
4667                let state = state.clone();
4668                async move { !state.instrument_queries.lock().await.is_empty() }
4669            },
4670            Duration::from_secs(1),
4671        )
4672        .await;
4673
4674        client.reset().expect("reset");
4675        wait_until_async(
4676            || async { client.tasks.all_finished() },
4677            Duration::from_secs(1),
4678        )
4679        .await;
4680        gate.add_permits(1);
4681
4682        assert!(client.instruments_by_symbol.load().is_empty());
4683        assert!(matches!(
4684            receiver.try_recv(),
4685            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
4686        ));
4687        terminate_tasks(&client.tasks, "test data client")
4688            .await
4689            .expect("data task terminated");
4690    }
4691
4692    #[tokio::test]
4693    async fn spawn_instrument_refresh_registers_task() {
4694        let config = OKXDataClientConfig::builder()
4695            .update_instruments_interval_mins(60)
4696            .build();
4697        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
4698        replace_data_event_sender(sender);
4699        let client = OKXDataClient::new(*OKX_CLIENT_ID, config).expect("data client");
4700
4701        client.register_instrument_refresh().unwrap();
4702        assert_eq!(client.tasks.len(), 1);
4703
4704        terminate_tasks(&client.tasks, "test data client")
4705            .await
4706            .expect("refresh task joins after cancel");
4707    }
4708
4709    #[tokio::test]
4710    async fn reconnect_does_not_leak_refresh_tasks() {
4711        let state = spot_refresh_state();
4712        let addr = start_refresh_server(state).await;
4713        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
4714        replace_data_event_sender(sender);
4715        let config = OKXDataClientConfig {
4716            instrument_types: vec![OKXInstrumentType::Spot],
4717            base_url_http: Some(format!("http://{addr}")),
4718            base_url_ws_public: Some(format!("ws://{addr}/ws/public")),
4719            base_url_ws_business: Some(format!("ws://{addr}/ws/business")),
4720            environment: OKXEnvironment::Live,
4721            http_timeout_secs: 5,
4722            max_retries: 0,
4723            retry_delay_initial_ms: 1,
4724            retry_delay_max_ms: 1,
4725            book_stale_check_interval_secs: 0,
4726            update_instruments_interval_mins: 60,
4727            ..OKXDataClientConfig::default()
4728        };
4729        let mut client = OKXDataClient::new(*OKX_CLIENT_ID, config).expect("data client");
4730
4731        for cycle in 1..=2 {
4732            client.connect().await.expect("connect");
4733            assert_eq!(
4734                client.tasks.len(),
4735                3,
4736                "cycle {cycle}: two stream tasks and one refresh task"
4737            );
4738            client.disconnect().await.expect("disconnect");
4739            assert!(
4740                client.tasks.is_empty(),
4741                "cycle {cycle}: teardown must join every task"
4742            );
4743        }
4744
4745        client.connect().await.expect("connect");
4746        client.connect().await.expect("repeated connect is a no-op");
4747        assert_eq!(client.tasks.len(), 3);
4748        client.disconnect().await.expect("disconnect");
4749    }
4750
4751    #[tokio::test]
4752    async fn reset_drains_old_generation_before_reconnect() {
4753        let state = spot_refresh_state();
4754        let addr = start_refresh_server(state).await;
4755        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
4756        replace_data_event_sender(sender);
4757        let config = OKXDataClientConfig {
4758            instrument_types: vec![OKXInstrumentType::Spot],
4759            base_url_http: Some(format!("http://{addr}")),
4760            base_url_ws_public: Some(format!("ws://{addr}/ws/public")),
4761            base_url_ws_business: Some(format!("ws://{addr}/ws/business")),
4762            environment: OKXEnvironment::Live,
4763            http_timeout_secs: 5,
4764            max_retries: 0,
4765            retry_delay_initial_ms: 1,
4766            retry_delay_max_ms: 1,
4767            book_stale_check_interval_secs: 0,
4768            update_instruments_interval_mins: 60,
4769            ..OKXDataClientConfig::default()
4770        };
4771        let mut client = OKXDataClient::new(*OKX_CLIENT_ID, config).expect("data client");
4772
4773        client.connect().await.expect("initial connect");
4774        client.reset().expect("reset");
4775        client.connect().await.expect("reconnect after reset");
4776
4777        assert_eq!(client.tasks.len(), 3);
4778        assert!(!client.tasks.all_finished());
4779        client.disconnect().await.expect("disconnect");
4780    }
4781
4782    #[tokio::test]
4783    async fn reconnect_does_not_republish_unchanged_instruments() {
4784        let state = spot_refresh_state();
4785        let addr = start_refresh_server(state.clone()).await;
4786        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4787        replace_data_event_sender(sender);
4788        let config = OKXDataClientConfig {
4789            instrument_types: vec![OKXInstrumentType::Spot],
4790            base_url_http: Some(format!("http://{addr}")),
4791            base_url_ws_public: Some(format!("ws://{addr}/ws/public")),
4792            base_url_ws_business: Some(format!("ws://{addr}/ws/business")),
4793            environment: OKXEnvironment::Live,
4794            http_timeout_secs: 5,
4795            max_retries: 0,
4796            retry_delay_initial_ms: 1,
4797            retry_delay_max_ms: 1,
4798            book_stale_check_interval_secs: 0,
4799            update_instruments_interval_mins: 60,
4800            ..OKXDataClientConfig::default()
4801        };
4802        let mut client = OKXDataClient::new(*OKX_CLIENT_ID, config).expect("data client");
4803
4804        client.connect().await.expect("first connect");
4805        assert_eq!(
4806            instrument_events(&mut receiver).len(),
4807            5,
4808            "first connect publishes the full cache"
4809        );
4810        client.disconnect().await.expect("disconnect");
4811
4812        client.connect().await.expect("reconnect");
4813        assert!(
4814            instrument_events(&mut receiver).is_empty(),
4815            "reconnect must not republish unchanged instruments"
4816        );
4817        client.disconnect().await.expect("disconnect");
4818
4819        {
4820            let mut payload = state.instruments_payload.lock().await;
4821            payload["data"][0]["tickSz"] = json!("0.5");
4822        }
4823        client.connect().await.expect("third connect");
4824        let events = instrument_events(&mut receiver);
4825        assert_eq!(
4826            events.len(),
4827            1,
4828            "reconnect publishes only changed definitions"
4829        );
4830        assert_eq!(events[0].id(), InstrumentId::from("BTC-USD.OKX"));
4831        client.disconnect().await.expect("disconnect");
4832    }
4833
4834    #[tokio::test]
4835    async fn stop_cancels_refresh_task() {
4836        let state = spot_refresh_state();
4837        let addr = start_refresh_server(state).await;
4838        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
4839        replace_data_event_sender(sender);
4840        let config = OKXDataClientConfig {
4841            instrument_types: vec![OKXInstrumentType::Spot],
4842            base_url_http: Some(format!("http://{addr}")),
4843            base_url_ws_public: Some(format!("ws://{addr}/ws/public")),
4844            base_url_ws_business: Some(format!("ws://{addr}/ws/business")),
4845            environment: OKXEnvironment::Live,
4846            http_timeout_secs: 5,
4847            max_retries: 0,
4848            retry_delay_initial_ms: 1,
4849            retry_delay_max_ms: 1,
4850            book_stale_check_interval_secs: 0,
4851            update_instruments_interval_mins: 60,
4852            ..OKXDataClientConfig::default()
4853        };
4854        let mut client = OKXDataClient::new(*OKX_CLIENT_ID, config).expect("data client");
4855
4856        client.connect().await.expect("connect");
4857        assert_eq!(client.tasks.len(), 3);
4858
4859        client.stop().expect("stop");
4860        terminate_tasks(&client.tasks, "test data client")
4861            .await
4862            .expect("stop must cancel every spawned task");
4863
4864        client.disconnect().await.expect("disconnect");
4865    }
4866
4867    #[tokio::test]
4868    async fn zero_interval_disables_refresh_on_connect() {
4869        let addr = start_refresh_server(spot_refresh_state()).await;
4870        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
4871        replace_data_event_sender(sender);
4872        let config = OKXDataClientConfig {
4873            instrument_types: vec![OKXInstrumentType::Spot],
4874            base_url_http: Some(format!("http://{addr}")),
4875            base_url_ws_public: Some(format!("ws://{addr}/ws/public")),
4876            base_url_ws_business: Some(format!("ws://{addr}/ws/business")),
4877            environment: OKXEnvironment::Live,
4878            http_timeout_secs: 5,
4879            max_retries: 0,
4880            retry_delay_initial_ms: 1,
4881            retry_delay_max_ms: 1,
4882            book_stale_check_interval_secs: 0,
4883            update_instruments_interval_mins: 0,
4884            ..OKXDataClientConfig::default()
4885        };
4886        let mut client = OKXDataClient::new(*OKX_CLIENT_ID, config).expect("data client");
4887
4888        client.connect().await.expect("connect");
4889        assert_eq!(
4890            client.tasks.len(),
4891            2,
4892            "only the two stream tasks run when refresh is disabled"
4893        );
4894
4895        client.disconnect().await.expect("disconnect");
4896    }
4897}
nautilus_okx/data.rs

nautilus_okx/
data.rs
