Skip to main content

nautilus_dydx/
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 dYdX adapter.
17
18use std::{
19    str::FromStr,
20    sync::{
21        Arc,
22        atomic::{AtomicBool, Ordering},
23    },
24};
25
26use anyhow::Context;
27use dashmap::DashMap;
28use futures_util::{Stream, StreamExt, pin_mut};
29use nautilus_common::{
30    clients::DataClient,
31    live::{runner::get_data_event_sender, runtime::get_runtime},
32    messages::{
33        DataEvent, DataResponse,
34        data::{
35            BarsResponse, BookResponse, FundingRatesResponse, InstrumentResponse,
36            InstrumentsResponse, RequestBars, RequestBookSnapshot, RequestFundingRates,
37            RequestInstrument, RequestInstruments, RequestTrades, SubscribeBars,
38            SubscribeBookDeltas, SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument,
39            SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices, SubscribeQuotes,
40            SubscribeTrades, TradesResponse, UnsubscribeBars, UnsubscribeBookDeltas,
41            UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrument,
42            UnsubscribeInstrumentStatus, UnsubscribeInstruments, UnsubscribeMarkPrices,
43            UnsubscribeQuotes, UnsubscribeTrades,
44        },
45    },
46};
47use nautilus_core::{
48    AtomicMap, AtomicSet,
49    datetime::datetime_to_unix_nanos,
50    time::{AtomicTime, get_atomic_clock_realtime},
51};
52use nautilus_model::{
53    data::{
54        Bar, BarSpecification, BarType, BookOrder, Data as NautilusData, FundingRateUpdate,
55        IndexPriceUpdate, InstrumentStatus, MarkPriceUpdate, OrderBookDelta, OrderBookDeltas,
56        OrderBookDeltas_API, QuoteTick,
57    },
58    enums::{BookAction, BookType, MarketStatusAction, OrderSide, RecordFlag},
59    identifiers::{ClientId, InstrumentId, Symbol, Venue},
60    instruments::{Instrument, InstrumentAny},
61    orderbook::OrderBook,
62    types::Quantity,
63};
64use rust_decimal::Decimal;
65use tokio::{task::JoinHandle, time::Duration};
66use tokio_util::sync::CancellationToken;
67use ustr::Ustr;
68
69use crate::{
70    common::{
71        consts::DYDX_VENUE,
72        enums::DydxCandleResolution,
73        instrument_cache::InstrumentCache,
74        parse::{extract_raw_symbol, parse_price},
75    },
76    config::DydxDataClientConfig,
77    http::client::DydxHttpClient,
78    websocket::{
79        client::{DydxWebSocketClient, candle_ids_from_topics},
80        enums::DydxWsOutputMessage,
81        parse as ws_parse,
82    },
83};
84
85struct WsMessageContext {
86    clock: &'static AtomicTime,
87    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
88    instrument_cache: Arc<InstrumentCache>,
89    order_books: Arc<DashMap<InstrumentId, OrderBook>>,
90    last_quotes: Arc<DashMap<InstrumentId, QuoteTick>>,
91    ws_client: DydxWebSocketClient,
92    http_client: DydxHttpClient,
93    active_quote_subs: Arc<AtomicSet<InstrumentId>>,
94    active_delta_subs: Arc<AtomicSet<InstrumentId>>,
95    active_trade_subs: Arc<AtomicSet<InstrumentId>>,
96    active_bar_subs: Arc<AtomicMap<(InstrumentId, String), BarType>>,
97    incomplete_bars: Arc<DashMap<BarType, Bar>>,
98    bar_type_mappings: Arc<AtomicMap<String, BarType>>,
99    active_mark_price_subs: Arc<AtomicSet<InstrumentId>>,
100    active_index_price_subs: Arc<AtomicSet<InstrumentId>>,
101    active_funding_rate_subs: Arc<AtomicSet<InstrumentId>>,
102    active_instrument_status_subs: Arc<AtomicSet<InstrumentId>>,
103    last_instrument_statuses: Arc<DashMap<InstrumentId, InstrumentStatus>>,
104    bars_timestamp_on_close: bool,
105    pending_bars: Arc<DashMap<String, Bar>>,
106    seen_tickers: Arc<AtomicSet<Ustr>>,
107}
108
109/// dYdX data client for live market data streaming and historical data requests.
110///
111/// This client integrates with the Nautilus DataEngine to provide:
112/// - Real-time market data via WebSocket subscriptions
113/// - Historical data via REST API requests
114/// - Automatic instrument discovery and caching
115/// - Connection lifecycle management
116#[derive(Debug)]
117pub struct DydxDataClient {
118    clock: &'static AtomicTime,
119    client_id: ClientId,
120    config: DydxDataClientConfig,
121    http_client: DydxHttpClient,
122    ws_client: DydxWebSocketClient,
123    is_connected: AtomicBool,
124    cancellation_token: CancellationToken,
125    tasks: Vec<JoinHandle<()>>,
126    data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
127    instrument_cache: Arc<InstrumentCache>,
128    order_books: Arc<DashMap<InstrumentId, OrderBook>>,
129    last_quotes: Arc<DashMap<InstrumentId, QuoteTick>>,
130    incomplete_bars: Arc<DashMap<BarType, Bar>>,
131    bar_type_mappings: Arc<AtomicMap<String, BarType>>,
132    active_quote_subs: Arc<AtomicSet<InstrumentId>>,
133    active_delta_subs: Arc<AtomicSet<InstrumentId>>,
134    active_trade_subs: Arc<AtomicSet<InstrumentId>>,
135    active_bar_subs: Arc<AtomicMap<(InstrumentId, String), BarType>>,
136    active_mark_price_subs: Arc<AtomicSet<InstrumentId>>,
137    active_index_price_subs: Arc<AtomicSet<InstrumentId>>,
138    active_funding_rate_subs: Arc<AtomicSet<InstrumentId>>,
139    active_instrument_status_subs: Arc<AtomicSet<InstrumentId>>,
140    last_instrument_statuses: Arc<DashMap<InstrumentId, InstrumentStatus>>,
141}
142
143impl DydxDataClient {
144    fn map_bar_spec_to_resolution(spec: &BarSpecification) -> anyhow::Result<&'static str> {
145        let resolution: &'static str = DydxCandleResolution::from_bar_spec(spec)?.into();
146        Ok(resolution)
147    }
148
149    /// Creates a new [`DydxDataClient`] instance.
150    ///
151    /// # Errors
152    ///
153    /// Returns an error if the client fails to initialize.
154    pub fn new(
155        client_id: ClientId,
156        config: DydxDataClientConfig,
157        http_client: DydxHttpClient,
158        ws_client: DydxWebSocketClient,
159    ) -> anyhow::Result<Self> {
160        let clock = get_atomic_clock_realtime();
161        let data_sender = get_data_event_sender();
162
163        let instrument_cache = Arc::clone(http_client.instrument_cache());
164
165        Ok(Self {
166            clock,
167            client_id,
168            config,
169            http_client,
170            ws_client,
171            is_connected: AtomicBool::new(false),
172            cancellation_token: CancellationToken::new(),
173            tasks: Vec::new(),
174            data_sender,
175            instrument_cache,
176            order_books: Arc::new(DashMap::new()),
177            last_quotes: Arc::new(DashMap::new()),
178            incomplete_bars: Arc::new(DashMap::new()),
179            bar_type_mappings: Arc::new(AtomicMap::new()),
180            active_quote_subs: Arc::new(AtomicSet::new()),
181            active_delta_subs: Arc::new(AtomicSet::new()),
182            active_trade_subs: Arc::new(AtomicSet::new()),
183            active_bar_subs: Arc::new(AtomicMap::new()),
184            active_mark_price_subs: Arc::new(AtomicSet::new()),
185            active_index_price_subs: Arc::new(AtomicSet::new()),
186            active_funding_rate_subs: Arc::new(AtomicSet::new()),
187            active_instrument_status_subs: Arc::new(AtomicSet::new()),
188            last_instrument_statuses: Arc::new(DashMap::new()),
189        })
190    }
191
192    /// Returns the venue for this data client.
193    #[must_use]
194    pub fn venue(&self) -> Venue {
195        *DYDX_VENUE
196    }
197
198    /// Returns a reference to the client configuration.
199    #[must_use]
200    pub fn config(&self) -> &DydxDataClientConfig {
201        &self.config
202    }
203
204    /// Returns `true` when the client is connected.
205    #[must_use]
206    pub fn is_connected(&self) -> bool {
207        self.is_connected.load(Ordering::Relaxed)
208    }
209
210    fn spawn_ws<F>(&self, fut: F, context: &'static str)
211    where
212        F: std::future::Future<Output = anyhow::Result<()>> + Send + 'static,
213    {
214        get_runtime().spawn(async move {
215            if let Err(e) = fut.await {
216                log::error!("{context}: {e:?}");
217            }
218        });
219    }
220
221    fn spawn_ws_stream_handler(
222        &mut self,
223        stream: impl Stream<Item = DydxWsOutputMessage> + Send + 'static,
224        ctx: WsMessageContext,
225    ) {
226        let cancellation = self.cancellation_token.clone();
227
228        let handle = get_runtime().spawn(async move {
229            log::debug!("Message processing task started");
230            pin_mut!(stream);
231
232            loop {
233                tokio::select! {
234                    maybe_msg = stream.next() => {
235                        match maybe_msg {
236                            Some(msg) => Self::handle_ws_message(msg, &ctx),
237                            None => {
238                                log::debug!("WebSocket message channel closed");
239                                break;
240                            }
241                        }
242                    }
243                    () = cancellation.cancelled() => {
244                        log::debug!("WebSocket message task cancelled");
245                        break;
246                    }
247                }
248            }
249            log::debug!("WebSocket stream handler ended");
250        });
251
252        self.tasks.push(handle);
253    }
254
255    async fn await_tasks_with_timeout(&mut self, timeout: Duration) {
256        for handle in self.tasks.drain(..) {
257            let _ = tokio::time::timeout(timeout, handle).await;
258        }
259    }
260
261    async fn bootstrap_instruments(&self) -> anyhow::Result<Vec<InstrumentAny>> {
262        self.http_client
263            .fetch_and_cache_instruments()
264            .await
265            .context("failed to load instruments from dYdX")?;
266
267        let instruments: Vec<InstrumentAny> = self.http_client.all_instruments();
268
269        if instruments.is_empty() {
270            log::warn!("No instruments were loaded");
271            return Ok(instruments);
272        }
273
274        log::debug!("Loaded {} instruments into shared cache", instruments.len());
275
276        self.ws_client.cache_instruments(instruments.clone());
277
278        for instrument in &instruments {
279            if let Err(e) = self
280                .data_sender
281                .send(DataEvent::Instrument(instrument.clone()))
282            {
283                log::warn!("Failed to publish instrument {}: {e}", instrument.id());
284            }
285        }
286        log::debug!("Published {} instruments to data engine", instruments.len());
287
288        Ok(instruments)
289    }
290}
291
292#[async_trait::async_trait(?Send)]
293impl DataClient for DydxDataClient {
294    fn client_id(&self) -> ClientId {
295        self.client_id
296    }
297
298    fn venue(&self) -> Option<Venue> {
299        Some(*DYDX_VENUE)
300    }
301
302    fn start(&mut self) -> anyhow::Result<()> {
303        log::info!(
304            "Starting: client_id={}, is_testnet={}",
305            self.client_id,
306            self.http_client.is_testnet()
307        );
308        Ok(())
309    }
310
311    fn stop(&mut self) -> anyhow::Result<()> {
312        log::info!("Stopping {}", self.client_id);
313        self.cancellation_token.cancel();
314        self.is_connected.store(false, Ordering::Relaxed);
315        Ok(())
316    }
317
318    fn reset(&mut self) -> anyhow::Result<()> {
319        log::debug!("Resetting {}", self.client_id);
320        self.is_connected.store(false, Ordering::Relaxed);
321        self.cancellation_token = CancellationToken::new();
322        // Abort remaining tasks instead of just dropping handles to prevent resource leaks
323        for handle in self.tasks.drain(..) {
324            handle.abort();
325        }
326        Ok(())
327    }
328
329    fn dispose(&mut self) -> anyhow::Result<()> {
330        log::debug!("Disposing {}", self.client_id);
331        self.stop()
332    }
333
334    async fn connect(&mut self) -> anyhow::Result<()> {
335        if self.is_connected() {
336            return Ok(());
337        }
338
339        log::info!("Connecting");
340
341        self.bootstrap_instruments().await?;
342
343        self.ws_client
344            .connect()
345            .await
346            .context("failed to connect dYdX websocket")?;
347
348        self.ws_client
349            .subscribe_markets()
350            .await
351            .context("failed to subscribe to markets channel")?;
352
353        let seen_tickers: Arc<AtomicSet<Ustr>> = Arc::new(AtomicSet::new());
354
355        for instrument in self.instrument_cache.all_instruments() {
356            let id = instrument.id();
357            let ticker = extract_raw_symbol(id.symbol.as_str());
358            seen_tickers.insert(Ustr::from(ticker));
359        }
360
361        let ctx = WsMessageContext {
362            clock: self.clock,
363            data_sender: self.data_sender.clone(),
364            instrument_cache: self.instrument_cache.clone(),
365            order_books: self.order_books.clone(),
366            last_quotes: self.last_quotes.clone(),
367            ws_client: self.ws_client.clone(),
368            http_client: self.http_client.clone(),
369            active_quote_subs: self.active_quote_subs.clone(),
370            active_delta_subs: self.active_delta_subs.clone(),
371            active_trade_subs: self.active_trade_subs.clone(),
372            active_bar_subs: self.active_bar_subs.clone(),
373            incomplete_bars: self.incomplete_bars.clone(),
374            bar_type_mappings: self.bar_type_mappings.clone(),
375            active_mark_price_subs: self.active_mark_price_subs.clone(),
376            active_index_price_subs: self.active_index_price_subs.clone(),
377            active_funding_rate_subs: self.active_funding_rate_subs.clone(),
378            active_instrument_status_subs: self.active_instrument_status_subs.clone(),
379            last_instrument_statuses: self.last_instrument_statuses.clone(),
380            bars_timestamp_on_close: self.ws_client.bars_timestamp_on_close(),
381            pending_bars: Arc::new(DashMap::new()),
382            seen_tickers,
383        };
384
385        let stream = self.ws_client.stream();
386        self.spawn_ws_stream_handler(stream, ctx);
387
388        self.is_connected.store(true, Ordering::Relaxed);
389        log::info!("Connected");
390
391        Ok(())
392    }
393
394    async fn disconnect(&mut self) -> anyhow::Result<()> {
395        if !self.is_connected() {
396            return Ok(());
397        }
398
399        log::info!("Disconnecting");
400
401        self.cancellation_token.cancel();
402
403        self.await_tasks_with_timeout(Duration::from_secs(5)).await;
404
405        self.ws_client
406            .disconnect()
407            .await
408            .context("failed to disconnect dYdX websocket")?;
409
410        self.last_instrument_statuses.clear();
411        self.is_connected.store(false, Ordering::Relaxed);
412        log::info!("Disconnected dYdX data client");
413
414        Ok(())
415    }
416
417    fn is_connected(&self) -> bool {
418        self.is_connected.load(Ordering::Relaxed)
419    }
420
421    fn is_disconnected(&self) -> bool {
422        !self.is_connected()
423    }
424
425    fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
426        log::debug!(
427            "subscribe_instruments: dYdX instruments discovered via global v4_markets channel"
428        );
429        Ok(())
430    }
431
432    fn subscribe_instrument(&mut self, cmd: SubscribeInstrument) -> anyhow::Result<()> {
433        if let Some(instrument) = self.instrument_cache.get(&cmd.instrument_id) {
434            log::debug!("Sending cached instrument for {}", cmd.instrument_id);
435            if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
436                log::warn!("Failed to send instrument {}: {e}", cmd.instrument_id);
437            }
438        } else {
439            log::warn!(
440                "Instrument {} not found in cache (available: {})",
441                cmd.instrument_id,
442                self.instrument_cache.len()
443            );
444        }
445        Ok(())
446    }
447
448    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
449        if cmd.book_type != BookType::L2_MBP {
450            anyhow::bail!(
451                "dYdX only supports L2_MBP order book deltas, received {:?}",
452                cmd.book_type
453            );
454        }
455
456        self.ensure_order_book(cmd.instrument_id, BookType::L2_MBP);
457        self.active_delta_subs.insert(cmd.instrument_id);
458
459        let ws = self.ws_client.clone();
460        let instrument_id = cmd.instrument_id;
461
462        self.spawn_ws(
463            async move {
464                ws.subscribe_orderbook(instrument_id)
465                    .await
466                    .context("orderbook subscription")
467            },
468            "dYdX orderbook subscription",
469        );
470
471        Ok(())
472    }
473
474    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
475        log::debug!(
476            "Subscribe_quotes for {}: subscribing to orderbook WS channel for quote synthesis",
477            cmd.instrument_id
478        );
479
480        self.ensure_order_book(cmd.instrument_id, BookType::L2_MBP);
481        self.active_quote_subs.insert(cmd.instrument_id);
482        let ws = self.ws_client.clone();
483        let instrument_id = cmd.instrument_id;
484
485        self.spawn_ws(
486            async move {
487                ws.subscribe_orderbook(instrument_id)
488                    .await
489                    .context("orderbook subscription (for quotes)")
490            },
491            "dYdX orderbook subscription (quotes)",
492        );
493
494        Ok(())
495    }
496
497    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
498        let ws = self.ws_client.clone();
499        let instrument_id = cmd.instrument_id;
500
501        self.active_trade_subs.insert(instrument_id);
502
503        self.spawn_ws(
504            async move {
505                ws.subscribe_trades(instrument_id)
506                    .await
507                    .context("trade subscription")
508            },
509            "dYdX trade subscription",
510        );
511
512        Ok(())
513    }
514
515    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
516        let instrument_id = cmd.instrument_id;
517        self.active_mark_price_subs.insert(instrument_id);
518        log::debug!("Subscribed to mark prices for {instrument_id} (via v4_markets channel)");
519        Ok(())
520    }
521
522    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
523        let instrument_id = cmd.instrument_id;
524        self.active_index_price_subs.insert(instrument_id);
525        log::debug!("Subscribed to index prices for {instrument_id} (via v4_markets channel)");
526        Ok(())
527    }
528
529    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
530        let ws = self.ws_client.clone();
531        let instrument_id = cmd.bar_type.instrument_id();
532        let spec = cmd.bar_type.spec();
533
534        let resolution = Self::map_bar_spec_to_resolution(&spec)?;
535        let bar_type = cmd.bar_type;
536        self.active_bar_subs
537            .insert((instrument_id, resolution.to_string()), bar_type);
538
539        let ticker = extract_raw_symbol(instrument_id.symbol.as_str());
540        let topic = format!("{ticker}/{resolution}");
541        self.bar_type_mappings.insert(topic, bar_type);
542
543        self.spawn_ws(
544            async move {
545                ws.subscribe_candles(instrument_id, resolution)
546                    .await
547                    .context("candles subscription")
548            },
549            "dYdX candles subscription",
550        );
551
552        Ok(())
553    }
554
555    fn subscribe_funding_rates(&mut self, cmd: SubscribeFundingRates) -> anyhow::Result<()> {
556        let instrument_id = cmd.instrument_id;
557        self.active_funding_rate_subs.insert(instrument_id);
558        log::debug!("Subscribed to funding rates for {instrument_id} (via v4_markets channel)");
559        Ok(())
560    }
561
562    fn subscribe_instrument_status(
563        &mut self,
564        cmd: SubscribeInstrumentStatus,
565    ) -> anyhow::Result<()> {
566        let instrument_id = cmd.instrument_id;
567        self.active_instrument_status_subs.insert(instrument_id);
568        log::debug!("Subscribed to instrument status for {instrument_id} (via v4_markets channel)");
569
570        // Replay last known status (initial snapshot arrives before subscription)
571        if let Some(status) = self.last_instrument_statuses.get(&instrument_id)
572            && let Err(e) = self.data_sender.send(DataEvent::InstrumentStatus(*status))
573        {
574            log::error!("Failed to replay instrument status for {instrument_id}: {e}");
575        }
576
577        Ok(())
578    }
579
580    fn unsubscribe_instruments(&mut self, _cmd: &UnsubscribeInstruments) -> anyhow::Result<()> {
581        log::debug!("unsubscribe_instruments: dYdX markets channel is global; no-op");
582        Ok(())
583    }
584
585    fn unsubscribe_instrument(&mut self, _cmd: &UnsubscribeInstrument) -> anyhow::Result<()> {
586        log::debug!("unsubscribe_instrument: dYdX markets channel is global; no-op");
587        Ok(())
588    }
589
590    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
591        self.active_delta_subs.remove(&cmd.instrument_id);
592
593        let ws = self.ws_client.clone();
594        let instrument_id = cmd.instrument_id;
595
596        self.spawn_ws(
597            async move {
598                ws.unsubscribe_orderbook(instrument_id)
599                    .await
600                    .context("orderbook unsubscription")
601            },
602            "dYdX orderbook unsubscription",
603        );
604
605        Ok(())
606    }
607
608    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
609        log::debug!(
610            "unsubscribe_quotes for {}: removing quote subscription",
611            cmd.instrument_id
612        );
613
614        self.active_quote_subs.remove(&cmd.instrument_id);
615
616        let ws = self.ws_client.clone();
617        let instrument_id = cmd.instrument_id;
618
619        self.spawn_ws(
620            async move {
621                ws.unsubscribe_orderbook(instrument_id)
622                    .await
623                    .context("orderbook unsubscription (for quotes)")
624            },
625            "dYdX orderbook unsubscription (quotes)",
626        );
627
628        Ok(())
629    }
630
631    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
632        self.active_trade_subs.remove(&cmd.instrument_id);
633
634        let ws = self.ws_client.clone();
635        let instrument_id = cmd.instrument_id;
636
637        self.spawn_ws(
638            async move {
639                ws.unsubscribe_trades(instrument_id)
640                    .await
641                    .context("trade unsubscription")
642            },
643            "dYdX trade unsubscription",
644        );
645
646        Ok(())
647    }
648
649    fn unsubscribe_mark_prices(&mut self, cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
650        self.active_mark_price_subs.remove(&cmd.instrument_id);
651        log::debug!("Unsubscribed from mark prices for {}", cmd.instrument_id);
652        Ok(())
653    }
654
655    fn unsubscribe_index_prices(&mut self, cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
656        self.active_index_price_subs.remove(&cmd.instrument_id);
657        log::debug!("Unsubscribed from index prices for {}", cmd.instrument_id);
658        Ok(())
659    }
660
661    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
662        let ws = self.ws_client.clone();
663        let instrument_id = cmd.bar_type.instrument_id();
664        let spec = cmd.bar_type.spec();
665
666        let resolution = Self::map_bar_spec_to_resolution(&spec)?;
667
668        self.active_bar_subs
669            .remove(&(instrument_id, resolution.to_string()));
670
671        let ticker = extract_raw_symbol(instrument_id.symbol.as_str());
672        let topic = format!("{ticker}/{resolution}");
673        self.bar_type_mappings.remove(&topic);
674
675        self.spawn_ws(
676            async move {
677                ws.unsubscribe_candles(instrument_id, resolution)
678                    .await
679                    .context("candles unsubscription")
680            },
681            "dYdX candles unsubscription",
682        );
683
684        Ok(())
685    }
686
687    fn unsubscribe_funding_rates(&mut self, cmd: &UnsubscribeFundingRates) -> anyhow::Result<()> {
688        self.active_funding_rate_subs.remove(&cmd.instrument_id);
689        log::debug!("Unsubscribed from funding rates for {}", cmd.instrument_id);
690        Ok(())
691    }
692
693    fn unsubscribe_instrument_status(
694        &mut self,
695        cmd: &UnsubscribeInstrumentStatus,
696    ) -> anyhow::Result<()> {
697        self.active_instrument_status_subs
698            .remove(&cmd.instrument_id);
699        log::debug!(
700            "Unsubscribed from instrument status for {}",
701            cmd.instrument_id
702        );
703        Ok(())
704    }
705
706    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
707        if request.start.is_some() {
708            log::warn!(
709                "Requesting instrument {} with specified `start` which has no effect",
710                request.instrument_id
711            );
712        }
713
714        if request.end.is_some() {
715            log::warn!(
716                "Requesting instrument {} with specified `end` which has no effect",
717                request.instrument_id
718            );
719        }
720
721        let instrument_cache = self.instrument_cache.clone();
722        let sender = self.data_sender.clone();
723        let http = self.http_client.clone();
724        let instrument_id = request.instrument_id;
725        let request_id = request.request_id;
726        let client_id = request.client_id.unwrap_or(self.client_id);
727        let start = request.start;
728        let end = request.end;
729        let params = request.params;
730        let clock = self.clock;
731        let start_nanos = datetime_to_unix_nanos(start);
732        let end_nanos = datetime_to_unix_nanos(end);
733
734        get_runtime().spawn(async move {
735            let instrument = match http.request_instruments(None, None, None).await {
736                Ok(instruments) => {
737                    for inst in &instruments {
738                        instrument_cache.insert_instrument_only(inst.clone());
739                    }
740                    instruments.into_iter().find(|i| i.id() == instrument_id)
741                }
742                Err(e) => {
743                    log::error!("Failed to fetch instruments from dYdX: {e:?}");
744                    None
745                }
746            };
747
748            if let Some(inst) = instrument {
749                let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
750                    request_id,
751                    client_id,
752                    instrument_id,
753                    inst,
754                    start_nanos,
755                    end_nanos,
756                    clock.get_time_ns(),
757                    params,
758                )));
759
760                if let Err(e) = sender.send(DataEvent::Response(response)) {
761                    log::error!("Failed to send instrument response: {e}");
762                }
763            } else {
764                log::error!("Instrument {instrument_id} not found");
765            }
766        });
767
768        Ok(())
769    }
770
771    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
772        let http = self.http_client.clone();
773        let sender = self.data_sender.clone();
774        let instrument_cache = self.instrument_cache.clone();
775        let request_id = request.request_id;
776        let client_id = request.client_id.unwrap_or(self.client_id);
777        let venue = self.venue();
778        let start = request.start;
779        let end = request.end;
780        let params = request.params;
781        let clock = self.clock;
782        let start_nanos = datetime_to_unix_nanos(start);
783        let end_nanos = datetime_to_unix_nanos(end);
784
785        get_runtime().spawn(async move {
786            match http.request_instruments(None, None, None).await {
787                Ok(instruments) => {
788                    log::debug!("Fetched {} instruments from dYdX", instruments.len());
789
790                    for instrument in &instruments {
791                        instrument_cache.insert_instrument_only(instrument.clone());
792                    }
793
794                    let response = DataResponse::Instruments(InstrumentsResponse::new(
795                        request_id,
796                        client_id,
797                        venue,
798                        instruments,
799                        start_nanos,
800                        end_nanos,
801                        clock.get_time_ns(),
802                        params,
803                    ));
804
805                    if let Err(e) = sender.send(DataEvent::Response(response)) {
806                        log::error!("Failed to send instruments response: {e}");
807                    }
808                }
809                Err(e) => {
810                    log::error!("Failed to fetch instruments from dYdX: {e:?}");
811
812                    let response = DataResponse::Instruments(InstrumentsResponse::new(
813                        request_id,
814                        client_id,
815                        venue,
816                        Vec::new(),
817                        start_nanos,
818                        end_nanos,
819                        clock.get_time_ns(),
820                        params,
821                    ));
822
823                    if let Err(e) = sender.send(DataEvent::Response(response)) {
824                        log::error!("Failed to send empty instruments response: {e}");
825                    }
826                }
827            }
828        });
829
830        Ok(())
831    }
832
833    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
834        if request.depth.is_some() {
835            log::warn!(
836                "Requesting book snapshot for {} with specified `depth` which has no effect",
837                request.instrument_id
838            );
839        }
840
841        let http_client = self.http_client.clone();
842        let sender = self.data_sender.clone();
843        let instrument_id = request.instrument_id;
844        let request_id = request.request_id;
845        let client_id = request.client_id.unwrap_or(self.client_id);
846        let params = request.params;
847        let clock = self.clock;
848
849        get_runtime().spawn(async move {
850            let mut book = OrderBook::new(instrument_id, BookType::L2_MBP);
851
852            match http_client.request_orderbook_snapshot(instrument_id).await {
853                Ok(deltas) => {
854                    if let Err(e) = book.apply_deltas(&deltas) {
855                        log::error!("Failed to apply book snapshot for {instrument_id}: {e}");
856                        book.reset();
857                    }
858                }
859                Err(e) => {
860                    log::error!("Book snapshot request failed for {instrument_id}: {e:?}");
861                }
862            }
863
864            let response = DataResponse::Book(BookResponse::new(
865                request_id,
866                client_id,
867                instrument_id,
868                book,
869                None,
870                None,
871                clock.get_time_ns(),
872                params,
873            ));
874
875            if let Err(e) = sender.send(DataEvent::Response(response)) {
876                log::error!("Failed to send book snapshot response: {e}");
877            }
878        });
879
880        Ok(())
881    }
882
883    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
884        let http_client = self.http_client.clone();
885        let sender = self.data_sender.clone();
886        let instrument_id = request.instrument_id;
887        let start = request.start;
888        let end = request.end;
889        let limit = request.limit.map(|n| n.get() as u32);
890        let request_id = request.request_id;
891        let client_id = request.client_id.unwrap_or(self.client_id);
892        let params = request.params;
893        let clock = self.clock;
894        let start_nanos = datetime_to_unix_nanos(start);
895        let end_nanos = datetime_to_unix_nanos(end);
896
897        get_runtime().spawn(async move {
898            match http_client
899                .request_trade_ticks(instrument_id, start, end, limit)
900                .await
901                .context("failed to request trades from dYdX")
902            {
903                Ok(trades) => {
904                    let response = DataResponse::Trades(TradesResponse::new(
905                        request_id,
906                        client_id,
907                        instrument_id,
908                        trades,
909                        start_nanos,
910                        end_nanos,
911                        clock.get_time_ns(),
912                        params,
913                    ));
914
915                    if let Err(e) = sender.send(DataEvent::Response(response)) {
916                        log::error!("Failed to send trades response: {e}");
917                    }
918                }
919                Err(e) => {
920                    log::error!("Trade request failed for {instrument_id}: {e:?}");
921
922                    let response = DataResponse::Trades(TradesResponse::new(
923                        request_id,
924                        client_id,
925                        instrument_id,
926                        Vec::new(),
927                        start_nanos,
928                        end_nanos,
929                        clock.get_time_ns(),
930                        params,
931                    ));
932
933                    if let Err(e) = sender.send(DataEvent::Response(response)) {
934                        log::error!("Failed to send empty trades response: {e}");
935                    }
936                }
937            }
938        });
939
940        Ok(())
941    }
942
943    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
944        let http_client = self.http_client.clone();
945        let sender = self.data_sender.clone();
946        let bar_type = request.bar_type;
947        let start = request.start;
948        let end = request.end;
949        let limit = request.limit.map(|n| n.get() as u32);
950        let request_id = request.request_id;
951        let client_id = request.client_id.unwrap_or(self.client_id);
952        let params = request.params;
953        let clock = self.clock;
954        let start_nanos = datetime_to_unix_nanos(start);
955        let end_nanos = datetime_to_unix_nanos(end);
956
957        get_runtime().spawn(async move {
958            match http_client
959                .request_bars(bar_type, start, end, limit, true)
960                .await
961                .context("failed to request bars from dYdX")
962            {
963                Ok(bars) => {
964                    let response = DataResponse::Bars(BarsResponse::new(
965                        request_id,
966                        client_id,
967                        bar_type,
968                        bars,
969                        start_nanos,
970                        end_nanos,
971                        clock.get_time_ns(),
972                        params,
973                    ));
974
975                    if let Err(e) = sender.send(DataEvent::Response(response)) {
976                        log::error!("Failed to send bars response: {e}");
977                    }
978                }
979                Err(e) => {
980                    log::error!("Bar request failed for {bar_type}: {e:?}");
981
982                    let response = DataResponse::Bars(BarsResponse::new(
983                        request_id,
984                        client_id,
985                        bar_type,
986                        Vec::new(),
987                        start_nanos,
988                        end_nanos,
989                        clock.get_time_ns(),
990                        params,
991                    ));
992
993                    if let Err(e) = sender.send(DataEvent::Response(response)) {
994                        log::error!("Failed to send empty bars response: {e}");
995                    }
996                }
997            }
998        });
999
1000        Ok(())
1001    }
1002
1003    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
1004        let http_client = self.http_client.clone();
1005        let sender = self.data_sender.clone();
1006        let instrument_id = request.instrument_id;
1007        let start = request.start;
1008        let end = request.end;
1009        let limit = request.limit.map(|n| n.get() as u32);
1010        let request_id = request.request_id;
1011        let client_id = request.client_id.unwrap_or(self.client_id);
1012        let params = request.params;
1013        let clock = self.clock;
1014        let start_nanos = datetime_to_unix_nanos(start);
1015        let end_nanos = datetime_to_unix_nanos(end);
1016
1017        get_runtime().spawn(async move {
1018            match http_client
1019                .request_funding_rates(instrument_id, start, end, limit)
1020                .await
1021                .context("failed to request funding rates from dYdX")
1022            {
1023                Ok(funding_rates) => {
1024                    let response = DataResponse::FundingRates(FundingRatesResponse::new(
1025                        request_id,
1026                        client_id,
1027                        instrument_id,
1028                        funding_rates,
1029                        start_nanos,
1030                        end_nanos,
1031                        clock.get_time_ns(),
1032                        params,
1033                    ));
1034
1035                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1036                        log::error!("Failed to send funding rates response: {e}");
1037                    }
1038                }
1039                Err(e) => {
1040                    log::error!("Funding rates request failed for {instrument_id}: {e:?}");
1041
1042                    let response = DataResponse::FundingRates(FundingRatesResponse::new(
1043                        request_id,
1044                        client_id,
1045                        instrument_id,
1046                        Vec::new(),
1047                        start_nanos,
1048                        end_nanos,
1049                        clock.get_time_ns(),
1050                        params,
1051                    ));
1052
1053                    if let Err(e) = sender.send(DataEvent::Response(response)) {
1054                        log::error!("Failed to send empty funding rates response: {e}");
1055                    }
1056                }
1057            }
1058        });
1059
1060        Ok(())
1061    }
1062}
1063
1064impl DydxDataClient {
1065    /// Returns a cached instrument by InstrumentId.
1066    #[must_use]
1067    pub fn get_instrument(&self, instrument_id: &InstrumentId) -> Option<InstrumentAny> {
1068        self.instrument_cache.get(instrument_id)
1069    }
1070
1071    /// Returns all cached instruments.
1072    #[must_use]
1073    pub fn get_instruments(&self) -> Vec<InstrumentAny> {
1074        self.instrument_cache.all_instruments()
1075    }
1076
1077    /// Caches a single instrument.
1078    pub fn cache_instrument(&self, instrument: InstrumentAny) {
1079        self.instrument_cache.insert_instrument_only(instrument);
1080    }
1081
1082    /// Caches multiple instruments.
1083    ///
1084    /// Clears the existing cache first, then adds all provided instruments.
1085    pub fn cache_instruments(&self, instruments: Vec<InstrumentAny>) {
1086        self.instrument_cache.clear();
1087        self.instrument_cache.insert_instruments_only(instruments);
1088    }
1089
1090    fn ensure_order_book(&self, instrument_id: InstrumentId, book_type: BookType) {
1091        self.order_books
1092            .entry(instrument_id)
1093            .or_insert_with(|| OrderBook::new(instrument_id, book_type));
1094    }
1095
1096    /// Returns the BarType for a given WebSocket candle topic.
1097    #[must_use]
1098    pub fn get_bar_type_for_topic(&self, topic: &str) -> Option<BarType> {
1099        self.bar_type_mappings.load().get(topic).copied()
1100    }
1101
1102    /// Returns all registered bar topics.
1103    #[must_use]
1104    pub fn get_bar_topics(&self) -> Vec<String> {
1105        self.bar_type_mappings.load().keys().cloned().collect()
1106    }
1107
1108    fn handle_ws_message(message: DydxWsOutputMessage, ctx: &WsMessageContext) {
1109        let ts_init = ctx.clock.get_time_ns();
1110
1111        match message {
1112            DydxWsOutputMessage::Trades { id, contents } => {
1113                let Some(instrument) = ctx.instrument_cache.get_by_market(&id) else {
1114                    log::warn!("No instrument cached for market {id}");
1115                    return;
1116                };
1117                let instrument_id = instrument.id();
1118
1119                match ws_parse::parse_trade_ticks(instrument_id, &instrument, &contents, ts_init) {
1120                    Ok(data) => {
1121                        Self::handle_data_message(
1122                            data,
1123                            &ctx.data_sender,
1124                            &ctx.incomplete_bars,
1125                            ctx.clock,
1126                        );
1127                    }
1128                    Err(e) => log::error!("Failed to parse trade ticks for {id}: {e}"),
1129                }
1130            }
1131            DydxWsOutputMessage::OrderbookSnapshot { id, contents } => {
1132                let Some(instrument) = ctx.instrument_cache.get_by_market(&id) else {
1133                    log::warn!("No instrument cached for market {id}");
1134                    return;
1135                };
1136                let instrument_id = instrument.id();
1137
1138                match ws_parse::parse_orderbook_snapshot(
1139                    &instrument_id,
1140                    &contents,
1141                    instrument.price_precision(),
1142                    instrument.size_precision(),
1143                    ts_init,
1144                ) {
1145                    Ok(deltas) => {
1146                        Self::handle_deltas_message(
1147                            deltas,
1148                            &ctx.data_sender,
1149                            &ctx.order_books,
1150                            &ctx.last_quotes,
1151                            &ctx.instrument_cache,
1152                            &ctx.active_quote_subs,
1153                            &ctx.active_delta_subs,
1154                        );
1155                    }
1156                    Err(e) => log::error!("Failed to parse orderbook snapshot for {id}: {e}"),
1157                }
1158            }
1159            DydxWsOutputMessage::OrderbookUpdate { id, contents } => {
1160                let Some(instrument) = ctx.instrument_cache.get_by_market(&id) else {
1161                    log::warn!("No instrument cached for market {id}");
1162                    return;
1163                };
1164                let instrument_id = instrument.id();
1165
1166                match ws_parse::parse_orderbook_deltas(
1167                    &instrument_id,
1168                    &contents,
1169                    instrument.price_precision(),
1170                    instrument.size_precision(),
1171                    ts_init,
1172                ) {
1173                    Ok(deltas) => {
1174                        Self::handle_deltas_message(
1175                            deltas,
1176                            &ctx.data_sender,
1177                            &ctx.order_books,
1178                            &ctx.last_quotes,
1179                            &ctx.instrument_cache,
1180                            &ctx.active_quote_subs,
1181                            &ctx.active_delta_subs,
1182                        );
1183                    }
1184                    Err(e) => log::error!("Failed to parse orderbook deltas for {id}: {e}"),
1185                }
1186            }
1187            DydxWsOutputMessage::OrderbookBatch { id, updates } => {
1188                let Some(instrument) = ctx.instrument_cache.get_by_market(&id) else {
1189                    log::warn!("No instrument cached for market {id}");
1190                    return;
1191                };
1192                let instrument_id = instrument.id();
1193                let price_precision = instrument.price_precision();
1194                let size_precision = instrument.size_precision();
1195
1196                let mut all_deltas = Vec::new();
1197                let last_idx = updates.len().saturating_sub(1);
1198
1199                for (i, update) in updates.iter().enumerate() {
1200                    let is_last = i == last_idx;
1201                    let result = if is_last {
1202                        ws_parse::parse_orderbook_deltas(
1203                            &instrument_id,
1204                            update,
1205                            price_precision,
1206                            size_precision,
1207                            ts_init,
1208                        )
1209                        .map(|d| d.deltas)
1210                    } else {
1211                        ws_parse::parse_orderbook_deltas_with_flag(
1212                            &instrument_id,
1213                            update,
1214                            price_precision,
1215                            size_precision,
1216                            ts_init,
1217                            false,
1218                        )
1219                    };
1220
1221                    match result {
1222                        Ok(deltas) => all_deltas.extend(deltas),
1223                        Err(e) => {
1224                            log::error!("Failed to parse orderbook batch delta {i} for {id}: {e}");
1225                            return;
1226                        }
1227                    }
1228                }
1229
1230                if all_deltas.is_empty() {
1231                    return;
1232                }
1233                let deltas = OrderBookDeltas::new(instrument_id, all_deltas);
1234                Self::handle_deltas_message(
1235                    deltas,
1236                    &ctx.data_sender,
1237                    &ctx.order_books,
1238                    &ctx.last_quotes,
1239                    &ctx.instrument_cache,
1240                    &ctx.active_quote_subs,
1241                    &ctx.active_delta_subs,
1242                );
1243            }
1244            DydxWsOutputMessage::Candles { id, contents } => {
1245                let parts: Vec<&str> = id.splitn(2, '/').collect();
1246                if parts.len() != 2 {
1247                    log::warn!("Unexpected candle topic format: {id}");
1248                    return;
1249                }
1250                let ticker = parts[0];
1251
1252                let Some(bar_type) = ctx.bar_type_mappings.load().get(&id).copied() else {
1253                    log::debug!("No bar type mapping for candle topic {id}");
1254                    return;
1255                };
1256
1257                let Some(instrument) = ctx.instrument_cache.get_by_market(ticker) else {
1258                    log::warn!("No instrument cached for market {ticker}");
1259                    return;
1260                };
1261
1262                match ws_parse::parse_candle_bar(
1263                    bar_type,
1264                    &instrument,
1265                    &contents,
1266                    ctx.bars_timestamp_on_close,
1267                    ts_init,
1268                ) {
1269                    Ok(bar) => {
1270                        let prev = ctx.pending_bars.get(&id).map(|r| *r);
1271                        if let Some(prev_bar) = prev
1272                            && bar.ts_event != prev_bar.ts_event
1273                        {
1274                            Self::emit_bar_guarded(prev_bar, ctx);
1275                        }
1276                        ctx.pending_bars.insert(id, bar);
1277                    }
1278                    Err(e) => log::error!("Failed to parse candle bar for {id}: {e}"),
1279                }
1280            }
1281            DydxWsOutputMessage::Markets(contents) => {
1282                Self::handle_markets_message(&contents, ctx, ts_init);
1283            }
1284            DydxWsOutputMessage::SubaccountSubscribed(_) => {
1285                log::debug!("Ignoring subaccount subscribed on data client");
1286            }
1287            DydxWsOutputMessage::SubaccountsChannelData(_) => {
1288                log::debug!("Ignoring subaccounts channel data on data client");
1289            }
1290            DydxWsOutputMessage::BlockHeight { .. } => {
1291                log::debug!("Ignoring block height on data client");
1292            }
1293            DydxWsOutputMessage::Error(err) => {
1294                log::warn!("dYdX WS error: {err}");
1295            }
1296            DydxWsOutputMessage::Reconnected { topics } => {
1297                let reconnected_candles = candle_ids_from_topics(&topics);
1298                ctx.pending_bars
1299                    .retain(|id, _| !reconnected_candles.contains(id));
1300
1301                let total_subs = ctx.active_quote_subs.len()
1302                    + ctx.active_delta_subs.len()
1303                    + ctx.active_trade_subs.len()
1304                    + ctx.active_bar_subs.len();
1305
1306                log::info!(
1307                    "dYdX WS reconnected; handler replayed channel subscriptions (active data subscriptions: total={}, quotes={}, deltas={}, trades={}, bars={})",
1308                    total_subs,
1309                    ctx.active_quote_subs.len(),
1310                    ctx.active_delta_subs.len(),
1311                    ctx.active_trade_subs.len(),
1312                    ctx.active_bar_subs.len()
1313                );
1314            }
1315        }
1316    }
1317
1318    fn instrument_id_from_ticker(ticker: &str) -> InstrumentId {
1319        let symbol = format!("{ticker}-PERP");
1320        InstrumentId::new(Symbol::new(&symbol), *DYDX_VENUE)
1321    }
1322
1323    fn handle_markets_message(
1324        contents: &crate::websocket::messages::DydxMarketsContents,
1325        ctx: &WsMessageContext,
1326        ts_init: nautilus_core::UnixNanos,
1327    ) {
1328        if let Some(ref oracle_prices) = contents.oracle_prices {
1329            for (ticker, oracle_data) in oracle_prices {
1330                let instrument_id = Self::instrument_id_from_ticker(ticker);
1331
1332                let Ok(price) = parse_price(&oracle_data.oracle_price, "oracle_price") else {
1333                    log::warn!("Failed to parse oracle price for {ticker}");
1334                    continue;
1335                };
1336
1337                if ctx.active_mark_price_subs.contains(&instrument_id) {
1338                    let mark_price = MarkPriceUpdate::new(instrument_id, price, ts_init, ts_init);
1339                    let data = NautilusData::MarkPriceUpdate(mark_price);
1340                    if let Err(e) = ctx.data_sender.send(DataEvent::Data(data)) {
1341                        log::error!("Failed to emit mark price for {instrument_id}: {e}");
1342                    }
1343                }
1344
1345                if ctx.active_index_price_subs.contains(&instrument_id) {
1346                    let index_price = IndexPriceUpdate::new(instrument_id, price, ts_init, ts_init);
1347                    let data = NautilusData::IndexPriceUpdate(index_price);
1348                    if let Err(e) = ctx.data_sender.send(DataEvent::Data(data)) {
1349                        log::error!("Failed to emit index price for {instrument_id}: {e}");
1350                    }
1351                }
1352            }
1353        }
1354
1355        Self::handle_markets_trading_data(contents.trading.as_ref(), ctx, ts_init, false);
1356        Self::handle_markets_trading_data(contents.markets.as_ref(), ctx, ts_init, true);
1357    }
1358
1359    fn handle_markets_trading_data(
1360        trading: Option<
1361            &std::collections::HashMap<String, crate::websocket::messages::DydxMarketTradingUpdate>,
1362        >,
1363        ctx: &WsMessageContext,
1364        ts_init: nautilus_core::UnixNanos,
1365        is_snapshot: bool,
1366    ) {
1367        let Some(trading_map) = trading else {
1368            return;
1369        };
1370
1371        for (ticker, update) in trading_map {
1372            let instrument_id = Self::instrument_id_from_ticker(ticker);
1373
1374            if let Some(status) = &update.status {
1375                if *status == crate::common::enums::DydxMarketStatus::Unknown {
1376                    log::warn!("Skipping unmodeled dYdX market status for {instrument_id}");
1377                } else {
1378                    let action = MarketStatusAction::from(*status);
1379                    let is_trading =
1380                        matches!(status, crate::common::enums::DydxMarketStatus::Active);
1381
1382                    let instrument_status = InstrumentStatus::new(
1383                        instrument_id,
1384                        action,
1385                        ts_init,
1386                        ts_init,
1387                        None,
1388                        None,
1389                        Some(is_trading),
1390                        None,
1391                        None,
1392                    );
1393
1394                    ctx.last_instrument_statuses
1395                        .insert(instrument_id, instrument_status);
1396
1397                    if ctx.active_instrument_status_subs.contains(&instrument_id)
1398                        && let Err(e) = ctx
1399                            .data_sender
1400                            .send(DataEvent::InstrumentStatus(instrument_status))
1401                    {
1402                        log::error!("Failed to emit instrument status for {instrument_id}: {e}");
1403                    }
1404                }
1405            }
1406
1407            let ticker_ustr = Ustr::from(ticker.as_str());
1408            if !ctx.seen_tickers.contains(&ticker_ustr) {
1409                let is_active = update
1410                    .status
1411                    .as_ref()
1412                    .is_none_or(|s| matches!(s, crate::common::enums::DydxMarketStatus::Active));
1413                if ctx.instrument_cache.get_by_market(ticker).is_some() {
1414                    ctx.seen_tickers.insert(ticker_ustr);
1415                } else if is_active {
1416                    ctx.seen_tickers.insert(ticker_ustr);
1417                    Self::handle_new_instrument_discovered(ticker, ctx);
1418                }
1419            }
1420
1421            if let Some(ref rate_str) = update.next_funding_rate {
1422                if let Ok(rate) = Decimal::from_str(rate_str) {
1423                    if ctx.active_funding_rate_subs.contains(&instrument_id) {
1424                        let funding_rate = FundingRateUpdate {
1425                            instrument_id,
1426                            rate,
1427                            interval: Some(60),
1428                            next_funding_ns: None,
1429                            ts_event: ts_init,
1430                            ts_init,
1431                        };
1432
1433                        if let Err(e) = ctx.data_sender.send(DataEvent::FundingRate(funding_rate)) {
1434                            log::error!("Failed to emit funding rate for {instrument_id}: {e}");
1435                        }
1436                    }
1437                } else {
1438                    log::warn!("Failed to parse next_funding_rate for {ticker}: {rate_str}");
1439                }
1440            }
1441
1442            if is_snapshot
1443                && let Some(ref oracle_price_str) = update.oracle_price
1444                && let Ok(price) = parse_price(oracle_price_str, "oracle_price")
1445            {
1446                if ctx.active_mark_price_subs.contains(&instrument_id) {
1447                    let mark_price = MarkPriceUpdate::new(instrument_id, price, ts_init, ts_init);
1448                    let data = NautilusData::MarkPriceUpdate(mark_price);
1449
1450                    if let Err(e) = ctx.data_sender.send(DataEvent::Data(data)) {
1451                        log::error!("Failed to emit mark price for {instrument_id}: {e}");
1452                    }
1453                }
1454
1455                if ctx.active_index_price_subs.contains(&instrument_id) {
1456                    let index_price = IndexPriceUpdate::new(instrument_id, price, ts_init, ts_init);
1457                    let data = NautilusData::IndexPriceUpdate(index_price);
1458
1459                    if let Err(e) = ctx.data_sender.send(DataEvent::Data(data)) {
1460                        log::error!("Failed to emit index price for {instrument_id}: {e}");
1461                    }
1462                }
1463            }
1464        }
1465    }
1466
1467    fn emit_bar_guarded(bar: Bar, ctx: &WsMessageContext) {
1468        let current_time_ns = ctx.clock.get_time_ns();
1469        if bar.ts_event <= current_time_ns {
1470            ctx.incomplete_bars.remove(&bar.bar_type);
1471            if let Err(e) = ctx
1472                .data_sender
1473                .send(DataEvent::Data(NautilusData::Bar(bar)))
1474            {
1475                log::error!("Failed to emit completed bar: {e}");
1476            }
1477        } else {
1478            ctx.incomplete_bars.insert(bar.bar_type, bar);
1479        }
1480    }
1481
1482    fn handle_new_instrument_discovered(ticker: &str, ctx: &WsMessageContext) {
1483        log::debug!("New instrument discovered via WebSocket: {ticker}");
1484
1485        let http_client = ctx.http_client.clone();
1486        let ws_client = ctx.ws_client.clone();
1487        let data_sender = ctx.data_sender.clone();
1488        let ticker = ticker.to_string();
1489
1490        get_runtime().spawn(async move {
1491            match http_client.fetch_and_cache_single_instrument(&ticker).await {
1492                Ok(Some(instrument)) => {
1493                    ws_client.cache_instrument(instrument.clone());
1494                    if let Err(e) = data_sender.send(DataEvent::Instrument(instrument)) {
1495                        log::error!("Failed to emit new instrument: {e}");
1496                    }
1497                    log::debug!("Fetched and cached new instrument: {ticker}");
1498                }
1499                Ok(None) => {
1500                    log::warn!("New instrument {ticker} not found or inactive");
1501                }
1502                Err(e) => {
1503                    log::error!("Failed to fetch new instrument {ticker}: {e}");
1504                }
1505            }
1506        });
1507    }
1508
1509    fn handle_data_message(
1510        payloads: Vec<NautilusData>,
1511        data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
1512        incomplete_bars: &Arc<DashMap<BarType, Bar>>,
1513        clock: &'static AtomicTime,
1514    ) {
1515        for data in payloads {
1516            // Filter bars through incomplete bars cache
1517            if let NautilusData::Bar(bar) = data {
1518                Self::handle_bar_message(bar, data_sender, incomplete_bars, clock);
1519            } else if let Err(e) = data_sender.send(DataEvent::Data(data)) {
1520                log::error!("Failed to emit data event: {e}");
1521            }
1522        }
1523    }
1524
1525    fn handle_bar_message(
1526        bar: Bar,
1527        data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
1528        incomplete_bars: &Arc<DashMap<BarType, Bar>>,
1529        clock: &'static AtomicTime,
1530    ) {
1531        let current_time_ns = clock.get_time_ns();
1532        let bar_type = bar.bar_type;
1533
1534        if bar.ts_event <= current_time_ns {
1535            // Bar is complete - emit it and remove from incomplete cache
1536            incomplete_bars.remove(&bar_type);
1537
1538            if let Err(e) = data_sender.send(DataEvent::Data(NautilusData::Bar(bar))) {
1539                log::error!("Failed to emit completed bar: {e}");
1540            }
1541        } else {
1542            // Bar is incomplete - cache it (updates existing entry)
1543            log::trace!(
1544                "Caching incomplete bar for {} (ts_event={}, current={})",
1545                bar_type,
1546                bar.ts_event,
1547                current_time_ns
1548            );
1549            incomplete_bars.insert(bar_type, bar);
1550        }
1551    }
1552
1553    fn resolve_crossed_order_book(
1554        book: &mut OrderBook,
1555        venue_deltas: &OrderBookDeltas,
1556        instrument: &InstrumentAny,
1557    ) -> anyhow::Result<OrderBookDeltas> {
1558        let instrument_id = venue_deltas.instrument_id;
1559        let ts_init = venue_deltas.ts_init;
1560        let mut all_deltas = venue_deltas.deltas.clone();
1561
1562        // If the input batch is a snapshot, every synthetic and terminator delta must
1563        // carry F_SNAPSHOT as well so consumers apply the whole batch as one
1564        // replacement image rather than a snapshot followed by standalone updates.
1565        let snapshot_flag = RecordFlag::F_SNAPSHOT as u8;
1566        let is_snapshot_batch = venue_deltas
1567            .deltas
1568            .iter()
1569            .any(|d| d.flags & snapshot_flag != 0);
1570        let synthetic_flags = if is_snapshot_batch { snapshot_flag } else { 0 };
1571
1572        // Apply the original venue deltas first
1573        book.apply_deltas(venue_deltas)?;
1574
1575        // Check if orderbook is crossed
1576        let mut is_crossed = if let (Some(bid_price), Some(ask_price)) =
1577            (book.best_bid_price(), book.best_ask_price())
1578        {
1579            bid_price >= ask_price
1580        } else {
1581            false
1582        };
1583
1584        // Iteratively uncross the orderbook
1585        while is_crossed {
1586            log::debug!(
1587                "Resolving crossed order book for {}: bid={:?} >= ask={:?}",
1588                instrument_id,
1589                book.best_bid_price(),
1590                book.best_ask_price()
1591            );
1592
1593            let bid_price = match book.best_bid_price() {
1594                Some(p) => p,
1595                None => break,
1596            };
1597            let ask_price = match book.best_ask_price() {
1598                Some(p) => p,
1599                None => break,
1600            };
1601            let bid_size = match book.best_bid_size() {
1602                Some(s) => s,
1603                None => break,
1604            };
1605            let ask_size = match book.best_ask_size() {
1606                Some(s) => s,
1607                None => break,
1608            };
1609
1610            let mut temp_deltas = Vec::new();
1611
1612            if bid_size > ask_size {
1613                // Remove ask level, reduce bid level
1614                let new_bid_size = Quantity::from_decimal_dp(
1615                    bid_size.as_decimal() - ask_size.as_decimal(),
1616                    instrument.size_precision(),
1617                )?;
1618                temp_deltas.push(OrderBookDelta::new(
1619                    instrument_id,
1620                    BookAction::Update,
1621                    BookOrder::new(OrderSide::Buy, bid_price, new_bid_size, 0),
1622                    synthetic_flags,
1623                    0,
1624                    ts_init,
1625                    ts_init,
1626                ));
1627                temp_deltas.push(OrderBookDelta::new(
1628                    instrument_id,
1629                    BookAction::Delete,
1630                    BookOrder::new(
1631                        OrderSide::Sell,
1632                        ask_price,
1633                        Quantity::zero(instrument.size_precision()),
1634                        0,
1635                    ),
1636                    synthetic_flags,
1637                    0,
1638                    ts_init,
1639                    ts_init,
1640                ));
1641            } else if bid_size < ask_size {
1642                // Remove bid level, reduce ask level
1643                let new_ask_size = Quantity::from_decimal_dp(
1644                    ask_size.as_decimal() - bid_size.as_decimal(),
1645                    instrument.size_precision(),
1646                )?;
1647                temp_deltas.push(OrderBookDelta::new(
1648                    instrument_id,
1649                    BookAction::Update,
1650                    BookOrder::new(OrderSide::Sell, ask_price, new_ask_size, 0),
1651                    synthetic_flags,
1652                    0,
1653                    ts_init,
1654                    ts_init,
1655                ));
1656                temp_deltas.push(OrderBookDelta::new(
1657                    instrument_id,
1658                    BookAction::Delete,
1659                    BookOrder::new(
1660                        OrderSide::Buy,
1661                        bid_price,
1662                        Quantity::zero(instrument.size_precision()),
1663                        0,
1664                    ),
1665                    synthetic_flags,
1666                    0,
1667                    ts_init,
1668                    ts_init,
1669                ));
1670            } else {
1671                // Equal sizes: remove both levels
1672                temp_deltas.push(OrderBookDelta::new(
1673                    instrument_id,
1674                    BookAction::Delete,
1675                    BookOrder::new(
1676                        OrderSide::Buy,
1677                        bid_price,
1678                        Quantity::zero(instrument.size_precision()),
1679                        0,
1680                    ),
1681                    synthetic_flags,
1682                    0,
1683                    ts_init,
1684                    ts_init,
1685                ));
1686                temp_deltas.push(OrderBookDelta::new(
1687                    instrument_id,
1688                    BookAction::Delete,
1689                    BookOrder::new(
1690                        OrderSide::Sell,
1691                        ask_price,
1692                        Quantity::zero(instrument.size_precision()),
1693                        0,
1694                    ),
1695                    synthetic_flags,
1696                    0,
1697                    ts_init,
1698                    ts_init,
1699                ));
1700            }
1701
1702            // Apply temporary deltas to the book
1703            let temp_deltas_obj = OrderBookDeltas::new(instrument_id, temp_deltas.clone());
1704            book.apply_deltas(&temp_deltas_obj)?;
1705            all_deltas.extend(temp_deltas);
1706
1707            // Check if still crossed
1708            is_crossed = if let (Some(bid_price), Some(ask_price)) =
1709                (book.best_bid_price(), book.best_ask_price())
1710            {
1711                bid_price >= ask_price
1712            } else {
1713                false
1714            };
1715        }
1716
1717        // Set F_LAST on the final delta, preserving F_SNAPSHOT when the batch is a
1718        // snapshot so consumers close the replacement image correctly.
1719        if let Some(last_delta) = all_deltas.last_mut() {
1720            last_delta.flags = synthetic_flags | RecordFlag::F_LAST as u8;
1721        }
1722
1723        Ok(OrderBookDeltas::new(instrument_id, all_deltas))
1724    }
1725
1726    fn handle_deltas_message(
1727        deltas: OrderBookDeltas,
1728        data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
1729        order_books: &Arc<DashMap<InstrumentId, OrderBook>>,
1730        last_quotes: &Arc<DashMap<InstrumentId, QuoteTick>>,
1731        instrument_cache: &Arc<InstrumentCache>,
1732        active_quote_subs: &Arc<AtomicSet<InstrumentId>>,
1733        active_delta_subs: &Arc<AtomicSet<InstrumentId>>,
1734    ) {
1735        let instrument_id = deltas.instrument_id;
1736
1737        // Get instrument for crossed orderbook resolution
1738        let instrument = match instrument_cache.get(&instrument_id) {
1739            Some(inst) => inst,
1740            None => {
1741                log::error!("Cannot resolve crossed order book: no instrument for {instrument_id}");
1742                // Still emit the raw deltas if delta subscription is active
1743                if active_delta_subs.contains(&instrument_id)
1744                    && let Err(e) = data_sender.send(DataEvent::Data(NautilusData::from(
1745                        OrderBookDeltas_API::new(deltas),
1746                    )))
1747                {
1748                    log::error!("Failed to emit order book deltas: {e}");
1749                }
1750                return;
1751            }
1752        };
1753
1754        // Always maintain local orderbook -- both subscription types need book state
1755        let mut book = order_books
1756            .entry(instrument_id)
1757            .or_insert_with(|| OrderBook::new(instrument_id, BookType::L2_MBP));
1758
1759        // Resolve crossed orderbook (applies deltas internally)
1760        let resolved_deltas =
1761            match Self::resolve_crossed_order_book(&mut book, &deltas, &instrument) {
1762                Ok(d) => d,
1763                Err(e) => {
1764                    log::error!("Failed to resolve crossed order book for {instrument_id}: {e}");
1765                    return;
1766                }
1767            };
1768
1769        // Conditionally emit QuoteTick if instrument has quote subscription
1770        if active_quote_subs.contains(&instrument_id) {
1771            // Generate QuoteTick from updated top-of-book
1772            // Edge case: If orderbook is empty after deltas, fall back to last quote
1773            let quote_opt = if let (Some(bid_price), Some(ask_price)) =
1774                (book.best_bid_price(), book.best_ask_price())
1775                && let (Some(bid_size), Some(ask_size)) =
1776                    (book.best_bid_size(), book.best_ask_size())
1777            {
1778                Some(QuoteTick::new(
1779                    instrument_id,
1780                    bid_price,
1781                    ask_price,
1782                    bid_size,
1783                    ask_size,
1784                    resolved_deltas.ts_event,
1785                    resolved_deltas.ts_init,
1786                ))
1787            } else {
1788                // Edge case: Empty orderbook levels - use last quote as fallback
1789                if book.best_bid_price().is_none() && book.best_ask_price().is_none() {
1790                    log::debug!(
1791                        "Empty orderbook for {instrument_id} after applying deltas, using last quote"
1792                    );
1793                    last_quotes.get(&instrument_id).map(|q| *q)
1794                } else {
1795                    None
1796                }
1797            };
1798
1799            if let Some(quote) = quote_opt {
1800                // Only emit when top-of-book changes
1801                let emit_quote = !matches!(
1802                    last_quotes.get(&instrument_id),
1803                    Some(existing) if *existing == quote
1804                );
1805
1806                if emit_quote {
1807                    last_quotes.insert(instrument_id, quote);
1808                    if let Err(e) = data_sender.send(DataEvent::Data(NautilusData::Quote(quote))) {
1809                        log::error!("Failed to emit quote tick: {e}");
1810                    }
1811                }
1812            } else if book.best_bid_price().is_some() || book.best_ask_price().is_some() {
1813                // Partial orderbook (only one side) - log but don't emit
1814                log::debug!(
1815                    "Incomplete top-of-book for {instrument_id} (bid={:?}, ask={:?})",
1816                    book.best_bid_price(),
1817                    book.best_ask_price()
1818                );
1819            }
1820        }
1821
1822        // Conditionally emit OrderBookDeltas if instrument has delta subscription
1823        if active_delta_subs.contains(&instrument_id) {
1824            let data: NautilusData = OrderBookDeltas_API::new(resolved_deltas).into();
1825            if let Err(e) = data_sender.send(DataEvent::Data(data)) {
1826                log::error!("Failed to emit order book deltas event: {e}");
1827            }
1828        }
1829    }
1830}
1831
1832#[cfg(test)]
1833mod tests {
1834    use nautilus_core::UnixNanos;
1835    use nautilus_model::{
1836        data::{BookOrder, OrderBookDelta, OrderBookDeltas},
1837        enums::{BookAction, BookType, OrderSide, RecordFlag},
1838        identifiers::{InstrumentId, Symbol},
1839        instruments::{CryptoPerpetual, InstrumentAny},
1840        orderbook::OrderBook,
1841        types::{Currency, Price, Quantity},
1842    };
1843    use rstest::rstest;
1844    use rust_decimal_macros::dec;
1845
1846    use super::*;
1847    use crate::common::consts::DYDX_VENUE;
1848
1849    fn test_instrument() -> InstrumentAny {
1850        let instrument_id = InstrumentId::new(Symbol::new("BTC-USD-PERP"), *DYDX_VENUE);
1851        InstrumentAny::CryptoPerpetual(CryptoPerpetual::new(
1852            instrument_id,
1853            instrument_id.symbol,
1854            Currency::BTC(),
1855            Currency::USD(),
1856            Currency::USD(),
1857            false,
1858            2,                   // price_precision
1859            8,                   // size_precision (wide enough to reveal f64 rounding)
1860            Price::new(0.01, 2), // price_increment
1861            Quantity::new(0.00000001, 8),
1862            None,
1863            None,
1864            None,
1865            None,
1866            None,
1867            None,
1868            None,
1869            None,
1870            None,
1871            None,
1872            None,
1873            None,
1874            None,
1875            None,
1876            UnixNanos::default(),
1877            UnixNanos::default(),
1878        ))
1879    }
1880
1881    fn seed_book_with_levels(
1882        instrument_id: InstrumentId,
1883        bids: &[(f64, f64)],
1884        asks: &[(f64, f64)],
1885    ) -> OrderBook {
1886        let mut book = OrderBook::new(instrument_id, BookType::L2_MBP);
1887        let ts = UnixNanos::default();
1888
1889        let mut deltas: Vec<OrderBookDelta> = Vec::new();
1890        deltas.push(OrderBookDelta::clear(instrument_id, 0, ts, ts));
1891        for (price, size) in bids {
1892            deltas.push(OrderBookDelta::new(
1893                instrument_id,
1894                BookAction::Add,
1895                BookOrder::new(
1896                    OrderSide::Buy,
1897                    Price::new(*price, 2),
1898                    Quantity::new(*size, 8),
1899                    0,
1900                ),
1901                0,
1902                0,
1903                ts,
1904                ts,
1905            ));
1906        }
1907
1908        for (price, size) in asks {
1909            deltas.push(OrderBookDelta::new(
1910                instrument_id,
1911                BookAction::Add,
1912                BookOrder::new(
1913                    OrderSide::Sell,
1914                    Price::new(*price, 2),
1915                    Quantity::new(*size, 8),
1916                    0,
1917                ),
1918                0,
1919                0,
1920                ts,
1921                ts,
1922            ));
1923        }
1924
1925        if let Some(last) = deltas.last_mut() {
1926            last.flags = RecordFlag::F_LAST as u8;
1927        }
1928
1929        book.apply_deltas(&OrderBookDeltas::new(instrument_id, deltas))
1930            .expect("failed to apply seed deltas");
1931        book
1932    }
1933
1934    fn crossing_bid_deltas(
1935        instrument_id: InstrumentId,
1936        bid_price: f64,
1937        bid_size: f64,
1938    ) -> OrderBookDeltas {
1939        let ts = UnixNanos::default();
1940        let delta = OrderBookDelta::new(
1941            instrument_id,
1942            BookAction::Add,
1943            BookOrder::new(
1944                OrderSide::Buy,
1945                Price::new(bid_price, 2),
1946                Quantity::new(bid_size, 8),
1947                0,
1948            ),
1949            RecordFlag::F_LAST as u8,
1950            0,
1951            ts,
1952            ts,
1953        );
1954        OrderBookDeltas::new(instrument_id, vec![delta])
1955    }
1956
1957    #[rstest]
1958    fn test_resolve_crossed_order_book_preserves_decimal_precision() {
1959        // Book seeded uncrossed: bid at 99.00 / ask at 100.05 size=0.50000000.
1960        // Venue delta adds a crossing bid at 100.10 size=1.00000001.
1961        // The reducing side (Buy) must end up at size = 0.50000001 exactly --
1962        // f64 subtraction of 1.00000001 - 0.5 would round this to 0.50000000 at 8 dp.
1963        let instrument = test_instrument();
1964        let instrument_id = instrument.id();
1965        let mut book = seed_book_with_levels(
1966            instrument_id,
1967            &[(99.00, 1.00000000)],
1968            &[(100.05, 0.50000000)],
1969        );
1970
1971        let venue_deltas = crossing_bid_deltas(instrument_id, 100.10, 1.00000001);
1972
1973        let resolved =
1974            DydxDataClient::resolve_crossed_order_book(&mut book, &venue_deltas, &instrument)
1975                .expect("resolution should succeed");
1976
1977        // An Update on the Buy side at the crossing price must carry the exact
1978        // Decimal-subtracted remainder (0.50000001), not the f64-rounded 0.50000000.
1979        let update = resolved
1980            .deltas
1981            .iter()
1982            .find(|d| {
1983                d.action == BookAction::Update
1984                    && d.order.side == OrderSide::Buy
1985                    && d.order.price.as_decimal() == dec!(100.10)
1986            })
1987            .expect("expected a Buy Update delta from crossed-book resolution");
1988        assert_eq!(update.order.size.as_decimal(), dec!(0.50000001));
1989
1990        // The terminal delta must carry F_LAST so downstream buffering flushes.
1991        assert_eq!(
1992            resolved.deltas.last().unwrap().flags,
1993            RecordFlag::F_LAST as u8,
1994        );
1995
1996        // Book is no longer crossed.
1997        if let (Some(bid), Some(ask)) = (book.best_bid_price(), book.best_ask_price()) {
1998            assert!(bid < ask, "book still crossed: bid={bid:?} ask={ask:?}");
1999        }
2000    }
2001
2002    fn crossing_snapshot_batch(
2003        instrument_id: InstrumentId,
2004        bid_price: f64,
2005        bid_size: f64,
2006    ) -> OrderBookDeltas {
2007        let ts = UnixNanos::default();
2008        let snapshot = RecordFlag::F_SNAPSHOT as u8;
2009        let last = RecordFlag::F_LAST as u8;
2010        // Mimic an inbound snapshot: every delta carries F_SNAPSHOT; terminator also
2011        // carries F_LAST.
2012        let deltas = vec![OrderBookDelta::new(
2013            instrument_id,
2014            BookAction::Add,
2015            BookOrder::new(
2016                OrderSide::Buy,
2017                Price::new(bid_price, 2),
2018                Quantity::new(bid_size, 8),
2019                0,
2020            ),
2021            snapshot | last,
2022            0,
2023            ts,
2024            ts,
2025        )];
2026        OrderBookDeltas::new(instrument_id, deltas)
2027    }
2028
2029    /// A crossed snapshot must exit `resolve_crossed_order_book` still flagged as a
2030    /// snapshot: every delta keeps `F_SNAPSHOT` and the terminator carries
2031    /// `F_SNAPSHOT | F_LAST`, so downstream consumers treat the emitted batch as a
2032    /// complete replacement image.
2033    #[rstest]
2034    fn test_resolve_crossed_order_book_preserves_snapshot_flags() {
2035        let instrument = test_instrument();
2036        let instrument_id = instrument.id();
2037        let mut book = seed_book_with_levels(
2038            instrument_id,
2039            &[(99.00, 1.00000000)],
2040            &[(100.05, 0.50000000)],
2041        );
2042
2043        let venue_deltas = crossing_snapshot_batch(instrument_id, 100.10, 1.00000001);
2044
2045        let resolved =
2046            DydxDataClient::resolve_crossed_order_book(&mut book, &venue_deltas, &instrument)
2047                .expect("resolution should succeed");
2048
2049        let snapshot = RecordFlag::F_SNAPSHOT as u8;
2050        let last = RecordFlag::F_LAST as u8;
2051
2052        // Every delta must still carry F_SNAPSHOT; the terminator carries both.
2053        for (idx, delta) in resolved.deltas.iter().enumerate() {
2054            assert!(
2055                delta.flags & snapshot != 0,
2056                "delta at index {idx} lost F_SNAPSHOT: flags={:#010b}",
2057                delta.flags,
2058            );
2059        }
2060        assert_eq!(
2061            resolved.deltas.last().unwrap().flags,
2062            snapshot | last,
2063            "snapshot terminator must be F_SNAPSHOT | F_LAST",
2064        );
2065    }
2066
2067    #[rstest]
2068    fn test_resolve_crossed_order_book_equal_sizes_removes_both_levels() {
2069        // Seed bid at 99.00 / ask at 100.05 size=1.0, then add a crossing bid at
2070        // 100.10 size=1.0 -- both top-of-book sides match in size and must be deleted.
2071        let instrument = test_instrument();
2072        let instrument_id = instrument.id();
2073        let mut book = seed_book_with_levels(
2074            instrument_id,
2075            &[(99.00, 1.00000000)],
2076            &[(100.05, 1.00000000)],
2077        );
2078
2079        let venue_deltas = crossing_bid_deltas(instrument_id, 100.10, 1.00000000);
2080
2081        let resolved =
2082            DydxDataClient::resolve_crossed_order_book(&mut book, &venue_deltas, &instrument)
2083                .expect("resolution should succeed");
2084
2085        // Equal-size branch must emit two Deletes (one per side) at top-of-book.
2086        let deletes_count = resolved
2087            .deltas
2088            .iter()
2089            .filter(|d| {
2090                d.action == BookAction::Delete
2091                    && (d.order.price.as_decimal() == dec!(100.10)
2092                        || d.order.price.as_decimal() == dec!(100.05))
2093            })
2094            .count();
2095        assert_eq!(deletes_count, 2);
2096    }
2097}