nautilus_coinbase/websocket/
handler.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//! Feed handler for parsing Coinbase WebSocket messages into Nautilus types.
17
18use std::{fmt::Debug, sync::Arc};
19
20use ahash::AHashMap;
21use nautilus_core::{
22    AtomicMap, UnixNanos,
23    time::{AtomicTime, get_atomic_clock_realtime},
24};
25use nautilus_model::{
26    data::{Bar, BarType, InstrumentStatus, OrderBookDeltas, QuoteTick, TradeTick},
27    identifiers::{AccountId, InstrumentId, Symbol},
28    instruments::{Instrument, InstrumentAny},
29    reports::OrderStatusReport,
30};
31use nautilus_network::{RECONNECTED, websocket::WebSocketClient};
32use tokio_tungstenite::tungstenite::Message;
33use ustr::Ustr;
34
35use crate::{
36    common::consts::COINBASE_VENUE,
37    websocket::{
38        client::COINBASE_WS_SUBSCRIPTION_KEYS,
39        messages::{CoinbaseWsMessage, CoinbaseWsSubscription, WsEventType, WsOrderUpdate},
40        parse::{
41            parse_ws_candle, parse_ws_l2_snapshot, parse_ws_l2_update, parse_ws_status_product,
42            parse_ws_ticker, parse_ws_trade, parse_ws_user_event_to_order_status_report,
43        },
44    },
45};
46
47fn instrument_id_from_product(product_id: &Ustr) -> InstrumentId {
48    InstrumentId::new(Symbol::new(*product_id), *COINBASE_VENUE)
49}
50
51fn resolve_instrument_id_from_aliases(
52    aliases: &AHashMap<Ustr, Ustr>,
53    product_id: &Ustr,
54) -> InstrumentId {
55    let resolved = aliases.get(product_id).copied().unwrap_or(*product_id);
56    instrument_id_from_product(&resolved)
57}
58
59/// Commands sent from [`super::client::CoinbaseWebSocketClient`] to the feed handler.
60pub enum HandlerCommand {
61    /// Provides the network-level WebSocket client.
62    SetClient(WebSocketClient),
63    /// Subscribes to a channel for the given product IDs.
64    Subscribe(CoinbaseWsSubscription),
65    /// Unsubscribes from a channel.
66    Unsubscribe(CoinbaseWsSubscription),
67    /// Disconnects the WebSocket.
68    Disconnect,
69    /// Caches instruments for precision lookups during parsing.
70    InitializeInstruments(Vec<InstrumentAny>),
71    /// Updates a single instrument in the cache.
72    UpdateInstrument(Box<InstrumentAny>),
73    /// Registers a bar type for candle parsing.
74    AddBarType { key: String, bar_type: BarType },
75    /// Removes a bar type registration.
76    RemoveBarType { key: String },
77    /// Sets the account ID used when emitting user-channel execution reports.
78    SetAccountId(AccountId),
79}
80
81impl Debug for HandlerCommand {
82    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        match self {
84            Self::SetClient(_) => f.write_str("SetClient"),
85            Self::Subscribe(s) => write!(f, "Subscribe({:?})", s.channel),
86            Self::Unsubscribe(s) => write!(f, "Unsubscribe({:?})", s.channel),
87            Self::Disconnect => f.write_str("Disconnect"),
88            Self::InitializeInstruments(v) => write!(f, "InitializeInstruments({})", v.len()),
89            Self::UpdateInstrument(i) => write!(f, "UpdateInstrument({})", i.id()),
90            Self::AddBarType { key, .. } => write!(f, "AddBarType({key})"),
91            Self::RemoveBarType { key } => write!(f, "RemoveBarType({key})"),
92            Self::SetAccountId(id) => write!(f, "SetAccountId({id})"),
93        }
94    }
95}
96
97/// Carrier for a single user-channel order update.
98///
99/// Pairs the parsed [`OrderStatusReport`] with the resolved instrument and
100/// the raw venue payload so downstream consumers (e.g. the execution client)
101/// can diff cumulative quantity and fees against their own tracked state.
102///
103/// `is_snapshot` is true when the wrapping `WsUserEvent` was a `snapshot`
104/// type. Snapshots restate the current cumulative state of every open order
105/// and must NOT be interpreted as fresh fills, otherwise a cold start (or
106/// any state-clearing reconnect) would synthesize phantom fills covering the
107/// entire pre-existing cumulative quantity.
108#[derive(Debug, Clone)]
109pub struct UserOrderUpdate {
110    pub report: Box<OrderStatusReport>,
111    pub update: Box<WsOrderUpdate>,
112    pub instrument: InstrumentAny,
113    pub is_snapshot: bool,
114    pub ts_event: UnixNanos,
115    pub ts_init: UnixNanos,
116}
117
118/// Nautilus-typed messages produced by the feed handler.
119#[derive(Debug, Clone)]
120pub enum NautilusWsMessage {
121    /// Trade tick from market_trades channel.
122    Trade(TradeTick),
123    /// Quote tick from ticker channel.
124    Quote(QuoteTick),
125    /// Order book deltas from l2_data channel.
126    Deltas(OrderBookDeltas),
127    /// Bar from candles channel.
128    Bar(Bar),
129    /// Order status update from the user channel.
130    UserOrder(Box<UserOrderUpdate>),
131    /// Futures balance summary snapshot from the
132    /// `futures_balance_summary` channel.
133    FuturesBalanceSummary(Box<crate::websocket::messages::WsFcmBalanceSummary>),
134    /// Instrument status update from the `status` channel. Emitted for every
135    /// product the venue reports; the data client filters by subscription.
136    InstrumentStatus(Box<InstrumentStatus>),
137    /// The connection was re-established after a drop.
138    Reconnected,
139    /// An error occurred during message processing.
140    Error(String),
141}
142
143/// Processes raw WebSocket messages into Nautilus domain types.
144#[derive(Debug)]
145pub struct FeedHandler {
146    clock: &'static AtomicTime,
147    signal: Arc<std::sync::atomic::AtomicBool>,
148    client: Option<WebSocketClient>,
149    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
150    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
151    instruments: AHashMap<InstrumentId, InstrumentAny>,
152    /// Shared with [`super::client::CoinbaseWebSocketClient`]; consulted in
153    /// `resolve_instrument_id_from_aliases` to re-key inbound messages whose wire
154    /// `product_id` is the canonical alias of a subscribed/submitted product.
155    subscription_aliases: Arc<AtomicMap<Ustr, Ustr>>,
156    bar_types: AHashMap<String, BarType>,
157    account_id: Option<AccountId>,
158    buffer: Vec<NautilusWsMessage>,
159}
160
161impl FeedHandler {
162    /// Creates a new [`FeedHandler`] instance.
163    pub fn new(
164        signal: Arc<std::sync::atomic::AtomicBool>,
165        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
166        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
167        subscription_aliases: Arc<AtomicMap<Ustr, Ustr>>,
168    ) -> Self {
169        Self {
170            clock: get_atomic_clock_realtime(),
171            signal,
172            client: None,
173            cmd_rx,
174            raw_rx,
175            instruments: AHashMap::new(),
176            subscription_aliases,
177            bar_types: AHashMap::new(),
178            account_id: None,
179            buffer: Vec::new(),
180        }
181    }
182
183    /// Sets the account ID used to stamp user-channel execution reports.
184    pub fn set_account_id(&mut self, account_id: AccountId) {
185        self.account_id = Some(account_id);
186    }
187
188    /// Polls for the next output message, processing commands and raw messages.
189    ///
190    /// Returns `None` when the handler should shut down.
191    pub async fn next(&mut self) -> Option<NautilusWsMessage> {
192        // Check signal before draining buffer so disconnect takes
193        // priority over pending buffered messages
194        if self.signal.load(std::sync::atomic::Ordering::Acquire) {
195            self.buffer.clear();
196            return None;
197        }
198
199        if let Some(msg) = self.buffer.pop() {
200            return Some(msg);
201        }
202
203        loop {
204            if self.signal.load(std::sync::atomic::Ordering::Acquire) {
205                return None;
206            }
207
208            tokio::select! {
209                Some(cmd) = self.cmd_rx.recv() => {
210                    match cmd {
211                        HandlerCommand::SetClient(client) => {
212                            self.client = Some(client);
213                        }
214                        HandlerCommand::Subscribe(sub) => {
215                            self.send_subscription(&sub).await;
216                        }
217                        HandlerCommand::Unsubscribe(sub) => {
218                            self.send_subscription(&sub).await;
219                        }
220                        HandlerCommand::Disconnect => {
221                            if let Some(client) = self.client.take() {
222                                // Transition to CLOSED immediately without waiting
223                                // for ACTIVE (avoids blocking during reconnect)
224                                client.notify_closed();
225                            }
226                            return None;
227                        }
228                        HandlerCommand::InitializeInstruments(instruments) => {
229                            for inst in instruments {
230                                self.instruments.insert(inst.id(), inst);
231                            }
232                        }
233                        HandlerCommand::UpdateInstrument(inst) => {
234                            self.instruments.insert(inst.id(), *inst);
235                        }
236                        HandlerCommand::AddBarType { key, bar_type } => {
237                            self.bar_types.insert(key, bar_type);
238                        }
239                        HandlerCommand::RemoveBarType { key } => {
240                            self.bar_types.remove(&key);
241                        }
242                        HandlerCommand::SetAccountId(account_id) => {
243                            self.account_id = Some(account_id);
244                        }
245                    }
246                }
247                Some(raw) = self.raw_rx.recv() => {
248                    match raw {
249                        Message::Text(text) => {
250                            if let Some(msg) = self.handle_text(&text) {
251                                return Some(msg);
252                            }
253                        }
254                        Message::Ping(data) => {
255                            if let Some(client) = &self.client
256                                && let Err(e) = client.send_pong(data.to_vec()).await
257                            {
258                                log::error!("Failed to send pong: {e}");
259                            }
260                        }
261                        Message::Close(_) => return None,
262                        _ => {}
263                    }
264                }
265                else => return None,
266            }
267        }
268    }
269
270    async fn send_subscription(&self, sub: &CoinbaseWsSubscription) {
271        let Some(client) = &self.client else {
272            log::warn!("Cannot send subscription, no WebSocket client set");
273            return;
274        };
275
276        match serde_json::to_string(sub) {
277            Ok(json) => {
278                if let Err(e) = client
279                    .send_text(json, Some(COINBASE_WS_SUBSCRIPTION_KEYS.as_slice()))
280                    .await
281                {
282                    log::error!("Failed to send subscription: {e}");
283                }
284            }
285            Err(e) => log::error!("Failed to serialize subscription: {e}"),
286        }
287    }
288
289    fn handle_text(&mut self, text: &str) -> Option<NautilusWsMessage> {
290        if text == RECONNECTED {
291            return Some(NautilusWsMessage::Reconnected);
292        }
293
294        let ts_init = self.clock.get_time_ns();
295
296        let msg: CoinbaseWsMessage = match serde_json::from_str(text) {
297            Ok(m) => m,
298            Err(e) => {
299                log::warn!("Failed to parse WS message: {e}");
300                return None;
301            }
302        };
303
304        match msg {
305            CoinbaseWsMessage::L2Data {
306                timestamp, events, ..
307            } => self.handle_l2_events(&events, &timestamp, ts_init),
308            CoinbaseWsMessage::MarketTrades { events, .. } => {
309                self.handle_market_trades(&events, ts_init)
310            }
311            CoinbaseWsMessage::Ticker {
312                timestamp, events, ..
313            }
314            | CoinbaseWsMessage::TickerBatch {
315                timestamp, events, ..
316            } => self.handle_ticker(&events, &timestamp, ts_init),
317            CoinbaseWsMessage::Candles { events, .. } => self.handle_candles(&events, ts_init),
318            CoinbaseWsMessage::Heartbeats { .. } => None,
319            CoinbaseWsMessage::Subscriptions { events, .. } => {
320                // Coinbase emits this after every subscribe and unsubscribe
321                // with the full current subscription set, so it's noisy at
322                // INFO and not strictly a "confirmation" of the latest action.
323                log::debug!("Subscription state: {events:?}");
324                None
325            }
326            CoinbaseWsMessage::User {
327                timestamp, events, ..
328            } => self.handle_user_events(&events, &timestamp, ts_init),
329            CoinbaseWsMessage::FuturesBalanceSummary { events, .. } => {
330                self.handle_futures_balance_summary(events)
331            }
332            CoinbaseWsMessage::Status {
333                timestamp, events, ..
334            } => self.handle_status_events(&events, &timestamp, ts_init),
335        }
336    }
337
338    fn handle_l2_events(
339        &mut self,
340        events: &[crate::websocket::messages::WsL2DataEvent],
341        timestamp: &str,
342        ts_init: UnixNanos,
343    ) -> Option<NautilusWsMessage> {
344        let ts_event = match crate::http::parse::parse_rfc3339_timestamp(timestamp) {
345            Ok(ts) => ts,
346            Err(e) => {
347                log::warn!("Failed to parse L2 message timestamp {timestamp}: {e}");
348                ts_init
349            }
350        };
351
352        let mut first: Option<NautilusWsMessage> = None;
353        let aliases = self.subscription_aliases.load();
354
355        for event in events {
356            let instrument_id = resolve_instrument_id_from_aliases(&aliases, &event.product_id);
357
358            let instrument = match self.instruments.get(&instrument_id) {
359                Some(inst) => inst,
360                None => {
361                    log::warn!("No instrument cached for {instrument_id}");
362                    continue;
363                }
364            };
365
366            let result = match event.event_type {
367                WsEventType::Snapshot => parse_ws_l2_snapshot(event, instrument, ts_event, ts_init),
368                WsEventType::Update => parse_ws_l2_update(event, instrument, ts_event, ts_init),
369            };
370
371            match result {
372                Ok(deltas) => {
373                    let msg = NautilusWsMessage::Deltas(deltas);
374
375                    if first.is_none() {
376                        first = Some(msg);
377                    } else {
378                        self.buffer.push(msg);
379                    }
380                }
381                Err(e) => log::warn!("Failed to parse L2 event: {e}"),
382            }
383        }
384
385        if first.is_some() {
386            self.buffer.reverse();
387        }
388        first
389    }
390
391    fn handle_market_trades(
392        &mut self,
393        events: &[crate::websocket::messages::WsMarketTradesEvent],
394        ts_init: UnixNanos,
395    ) -> Option<NautilusWsMessage> {
396        let aliases = self.subscription_aliases.load();
397
398        for event in events {
399            for trade in &event.trades {
400                let instrument_id = resolve_instrument_id_from_aliases(&aliases, &trade.product_id);
401
402                let instrument = match self.instruments.get(&instrument_id) {
403                    Some(inst) => inst,
404                    None => {
405                        log::warn!("No instrument cached for {instrument_id}");
406                        continue;
407                    }
408                };
409
410                match parse_ws_trade(trade, instrument, ts_init) {
411                    Ok(tick) => {
412                        self.buffer_remaining_trades(events, event, trade, ts_init);
413                        // Reverse so pop() drains in exchange order
414                        self.buffer.reverse();
415                        return Some(NautilusWsMessage::Trade(tick));
416                    }
417                    Err(e) => log::warn!("Failed to parse trade: {e}"),
418                }
419            }
420        }
421        None
422    }
423
424    fn buffer_remaining_trades(
425        &mut self,
426        events: &[crate::websocket::messages::WsMarketTradesEvent],
427        current_event: &crate::websocket::messages::WsMarketTradesEvent,
428        current_trade: &crate::websocket::messages::WsTrade,
429        ts_init: UnixNanos,
430    ) {
431        let mut found_current = false;
432        let aliases = self.subscription_aliases.load();
433
434        for event in events {
435            let is_current_event = std::ptr::eq(event, current_event);
436
437            for trade in &event.trades {
438                if !found_current {
439                    if is_current_event && std::ptr::eq(trade, current_trade) {
440                        found_current = true;
441                    }
442                    continue;
443                }
444
445                let instrument_id = resolve_instrument_id_from_aliases(&aliases, &trade.product_id);
446
447                if let Some(instrument) = self.instruments.get(&instrument_id)
448                    && let Ok(tick) = parse_ws_trade(trade, instrument, ts_init)
449                {
450                    self.buffer.push(NautilusWsMessage::Trade(tick));
451                }
452            }
453        }
454    }
455
456    fn handle_ticker(
457        &mut self,
458        events: &[crate::websocket::messages::WsTickerEvent],
459        timestamp: &str,
460        ts_init: UnixNanos,
461    ) -> Option<NautilusWsMessage> {
462        let ts_event = crate::http::parse::parse_rfc3339_timestamp(timestamp).unwrap_or(ts_init);
463
464        let mut first: Option<NautilusWsMessage> = None;
465        let aliases = self.subscription_aliases.load();
466
467        for event in events {
468            for ticker in &event.tickers {
469                let instrument_id =
470                    resolve_instrument_id_from_aliases(&aliases, &ticker.product_id);
471
472                let instrument = match self.instruments.get(&instrument_id) {
473                    Some(inst) => inst,
474                    None => {
475                        log::warn!("No instrument cached for {instrument_id}");
476                        continue;
477                    }
478                };
479
480                match parse_ws_ticker(ticker, instrument, ts_event, ts_init) {
481                    Ok(quote) => {
482                        let msg = NautilusWsMessage::Quote(quote);
483
484                        if first.is_none() {
485                            first = Some(msg);
486                        } else {
487                            self.buffer.push(msg);
488                        }
489                    }
490                    Err(e) => log::warn!("Failed to parse ticker: {e}"),
491                }
492            }
493        }
494
495        if first.is_some() {
496            self.buffer.reverse();
497        }
498        first
499    }
500
501    fn handle_user_events(
502        &mut self,
503        events: &[crate::websocket::messages::WsUserEvent],
504        timestamp: &str,
505        ts_init: UnixNanos,
506    ) -> Option<NautilusWsMessage> {
507        let Some(account_id) = self.account_id else {
508            log::debug!(
509                "Dropping user event: account_id not set (call SetAccountId after connect)"
510            );
511            return None;
512        };
513
514        let ts_event = match crate::http::parse::parse_rfc3339_timestamp(timestamp) {
515            Ok(ts) => ts,
516            Err(e) => {
517                log::warn!("Failed to parse user message timestamp {timestamp}: {e}");
518                ts_init
519            }
520        };
521
522        let mut first: Option<NautilusWsMessage> = None;
523        let aliases = self.subscription_aliases.load();
524
525        for event in events {
526            let is_snapshot = matches!(event.event_type, WsEventType::Snapshot);
527
528            for order in &event.orders {
529                let instrument_id = resolve_instrument_id_from_aliases(&aliases, &order.product_id);
530                let instrument = match self.instruments.get(&instrument_id).cloned() {
531                    Some(inst) => inst,
532                    None => {
533                        log::warn!("No instrument cached for {instrument_id}");
534                        continue;
535                    }
536                };
537
538                self.emit_user_event_messages(
539                    order,
540                    &instrument,
541                    account_id,
542                    is_snapshot,
543                    ts_event,
544                    ts_init,
545                    &mut first,
546                );
547            }
548        }
549
550        if first.is_some() {
551            self.buffer.reverse();
552        }
553        first
554    }
555
556    #[allow(clippy::too_many_arguments)]
557    fn emit_user_event_messages(
558        &mut self,
559        order: &WsOrderUpdate,
560        instrument: &InstrumentAny,
561        account_id: AccountId,
562        is_snapshot: bool,
563        ts_event: UnixNanos,
564        ts_init: UnixNanos,
565        first: &mut Option<NautilusWsMessage>,
566    ) {
567        let report = match parse_ws_user_event_to_order_status_report(
568            order, instrument, account_id, ts_event, ts_init,
569        ) {
570            Ok(r) => r,
571            Err(e) => {
572                log::warn!("Failed to parse user order update: {e}");
573                return;
574            }
575        };
576
577        let msg = NautilusWsMessage::UserOrder(Box::new(UserOrderUpdate {
578            report: Box::new(report),
579            update: Box::new(order.clone()),
580            instrument: instrument.clone(),
581            is_snapshot,
582            ts_event,
583            ts_init,
584        }));
585
586        if first.is_none() {
587            *first = Some(msg);
588        } else {
589            self.buffer.push(msg);
590        }
591    }
592
593    fn handle_status_events(
594        &mut self,
595        events: &[crate::websocket::messages::WsStatusEvent],
596        timestamp: &str,
597        ts_init: UnixNanos,
598    ) -> Option<NautilusWsMessage> {
599        let ts_event = crate::http::parse::parse_rfc3339_timestamp(timestamp).unwrap_or(ts_init);
600
601        let mut first: Option<NautilusWsMessage> = None;
602        let aliases = self.subscription_aliases.load();
603
604        for event in events {
605            for product in &event.products {
606                let canonical = product.id;
607                let resolved = resolve_instrument_id_from_aliases(&aliases, &canonical);
608                let Some(status) = parse_ws_status_product(product, resolved, ts_event, ts_init)
609                else {
610                    continue;
611                };
612                let msg = NautilusWsMessage::InstrumentStatus(Box::new(status));
613
614                if first.is_none() {
615                    first = Some(msg);
616                } else {
617                    self.buffer.push(msg);
618                }
619            }
620        }
621
622        if first.is_some() {
623            self.buffer.reverse();
624        }
625        first
626    }
627
628    fn handle_futures_balance_summary(
629        &mut self,
630        events: Vec<crate::websocket::messages::WsFuturesBalanceSummaryEvent>,
631    ) -> Option<NautilusWsMessage> {
632        let mut first: Option<NautilusWsMessage> = None;
633
634        for event in events {
635            let msg = NautilusWsMessage::FuturesBalanceSummary(Box::new(event.fcm_balance_summary));
636
637            if first.is_none() {
638                first = Some(msg);
639            } else {
640                self.buffer.push(msg);
641            }
642        }
643
644        if first.is_some() {
645            self.buffer.reverse();
646        }
647        first
648    }
649
650    fn handle_candles(
651        &mut self,
652        events: &[crate::websocket::messages::WsCandlesEvent],
653        ts_init: UnixNanos,
654    ) -> Option<NautilusWsMessage> {
655        let mut first: Option<NautilusWsMessage> = None;
656        let aliases = self.subscription_aliases.load();
657
658        for event in events {
659            for candle in &event.candles {
660                let key = candle.product_id.as_str();
661
662                let bar_type = match self.bar_types.get(key) {
663                    Some(bt) => *bt,
664                    None => {
665                        log::debug!("No bar type registered for {key}");
666                        continue;
667                    }
668                };
669
670                let instrument_id =
671                    resolve_instrument_id_from_aliases(&aliases, &candle.product_id);
672
673                let instrument = match self.instruments.get(&instrument_id) {
674                    Some(inst) => inst,
675                    None => {
676                        log::warn!("No instrument cached for {instrument_id}");
677                        continue;
678                    }
679                };
680
681                match parse_ws_candle(candle, bar_type, instrument, ts_init) {
682                    Ok(bar) => {
683                        let msg = NautilusWsMessage::Bar(bar);
684
685                        if first.is_none() {
686                            first = Some(msg);
687                        } else {
688                            self.buffer.push(msg);
689                        }
690                    }
691                    Err(e) => log::warn!("Failed to parse candle: {e}"),
692                }
693            }
694        }
695
696        if first.is_some() {
697            self.buffer.reverse();
698        }
699        first
700    }
701}
702
703#[cfg(test)]
704mod tests {
705    use std::sync::{Arc, atomic::AtomicBool};
706
707    use nautilus_model::{
708        identifiers::Symbol,
709        instruments::CurrencyPair,
710        types::{Currency, Price, Quantity},
711    };
712    use rstest::rstest;
713
714    use super::*;
715    use crate::common::{consts::COINBASE_VENUE, testing::load_test_fixture};
716
717    fn test_handler() -> FeedHandler {
718        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
719        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
720        FeedHandler::new(
721            Arc::new(AtomicBool::new(false)),
722            cmd_rx,
723            raw_rx,
724            Arc::new(AtomicMap::new()),
725        )
726    }
727
728    fn btc_usd_instrument() -> InstrumentAny {
729        let instrument_id = InstrumentId::new(Symbol::new("BTC-USD"), *COINBASE_VENUE);
730        InstrumentAny::CurrencyPair(CurrencyPair::new(
731            instrument_id,
732            Symbol::new("BTC-USD"),
733            Currency::get_or_create_crypto("BTC"),
734            Currency::get_or_create_crypto("USD"),
735            2,
736            8,
737            Price::from("0.01"),
738            Quantity::from("0.00000001"),
739            None,
740            None,
741            None,
742            Some(Quantity::from("0.00000001")),
743            None,
744            None,
745            None,
746            None,
747            None,
748            None,
749            None,
750            None,
751            None,
752            None,
753            UnixNanos::default(),
754            UnixNanos::default(),
755        ))
756    }
757
758    #[rstest]
759    fn test_handle_text_drops_user_channel_when_account_id_unset() {
760        let json = load_test_fixture("ws_user.json");
761        let mut handler = test_handler();
762
763        // account_id is intentionally left unset; events should be dropped
764        assert!(handler.handle_text(&json).is_none());
765        assert!(handler.buffer.is_empty());
766    }
767
768    #[rstest]
769    fn test_handle_user_event_emits_user_order_update() {
770        use nautilus_model::{
771            enums::{OrderSide, OrderStatus},
772            identifiers::AccountId,
773            types::Quantity,
774        };
775
776        use crate::common::enums::CoinbaseProductType;
777
778        let json = load_test_fixture("ws_user.json");
779        let mut handler = test_handler();
780        handler.set_account_id(AccountId::new("COINBASE-001"));
781        handler
782            .instruments
783            .insert(btc_usd_instrument().id(), btc_usd_instrument());
784
785        let msg = handler
786            .handle_text(&json)
787            .expect("handler should emit a user-channel update");
788
789        match msg {
790            NautilusWsMessage::UserOrder(carrier) => {
791                // Status report fields.
792                assert_eq!(carrier.report.account_id.as_str(), "COINBASE-001");
793                assert_eq!(carrier.report.instrument_id, btc_usd_instrument().id());
794                assert_eq!(
795                    carrier.report.venue_order_id.as_str(),
796                    "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
797                );
798                assert_eq!(
799                    carrier.report.client_order_id.unwrap().as_str(),
800                    "11111-000000-000001"
801                );
802                assert_eq!(carrier.report.order_side, OrderSide::Buy);
803                assert_eq!(carrier.report.order_status, OrderStatus::Accepted);
804                assert_eq!(carrier.report.filled_qty, Quantity::from("0.00000000"));
805                assert_eq!(carrier.report.quantity, Quantity::from("0.00100000"));
806
807                // Raw venue update fields.
808                assert_eq!(carrier.update.product_id, "BTC-USD");
809                assert_eq!(carrier.update.product_type, CoinbaseProductType::Spot);
810                assert_eq!(carrier.update.cumulative_quantity, "0");
811                assert_eq!(carrier.update.leaves_quantity, "0.001");
812
813                // Carrier metadata.
814                assert_eq!(carrier.instrument.id(), btc_usd_instrument().id());
815                assert!(carrier.ts_event.as_u64() > 0);
816            }
817            other => panic!("expected UserOrder, was {other:?}"),
818        }
819    }
820
821    #[rstest]
822    fn test_handle_text_emits_instrument_status_from_status_channel() {
823        use nautilus_model::enums::MarketStatusAction;
824
825        let json = r#"{
826          "channel": "status",
827          "client_id": "",
828          "timestamp": "2023-02-09T20:29:49.753424311Z",
829          "sequence_num": 0,
830          "events": [
831            {
832              "type": "snapshot",
833              "products": [
834                {
835                  "product_type": "SPOT",
836                  "id": "BTC-USD",
837                  "base_currency": "BTC",
838                  "quote_currency": "USD",
839                  "base_increment": "0.00000001",
840                  "quote_increment": "0.01",
841                  "display_name": "BTC/USD",
842                  "status": "online",
843                  "status_message": "",
844                  "min_market_funds": "1"
845                },
846                {
847                  "product_type": "SPOT",
848                  "id": "ETH-USD",
849                  "base_currency": "ETH",
850                  "quote_currency": "USD",
851                  "base_increment": "0.00000001",
852                  "quote_increment": "0.01",
853                  "display_name": "ETH/USD",
854                  "status": "offline",
855                  "status_message": "maintenance",
856                  "min_market_funds": "1"
857                }
858              ]
859            }
860          ]
861        }"#;
862        let mut handler = test_handler();
863
864        let first = handler
865            .handle_text(json)
866            .expect("status channel must emit InstrumentStatus");
867        let NautilusWsMessage::InstrumentStatus(status) = first else {
868            panic!("expected InstrumentStatus, was {first:?}");
869        };
870        assert_eq!(status.instrument_id, btc_usd_instrument().id());
871        assert_eq!(status.action, MarketStatusAction::Trading);
872        assert_eq!(status.is_trading, Some(true));
873        assert!(status.reason.is_none());
874
875        // The second product (ETH-USD offline) is buffered behind the BTC one.
876        assert_eq!(handler.buffer.len(), 1);
877        let NautilusWsMessage::InstrumentStatus(status) = handler.buffer.pop().unwrap() else {
878            panic!("expected buffered InstrumentStatus");
879        };
880        assert_eq!(status.action, MarketStatusAction::Halt);
881        assert_eq!(status.is_trading, Some(false));
882        assert_eq!(
883            status.reason.map(|s| s.to_string()),
884            Some("maintenance".to_string())
885        );
886    }
887
888    #[rstest]
889    fn test_handle_l2_update_uses_batch_timestamp_for_all_deltas() {
890        let json = load_test_fixture("ws_l2_data_update.json");
891        let mut handler = test_handler();
892        handler
893            .instruments
894            .insert(btc_usd_instrument().id(), btc_usd_instrument());
895
896        let msg = handler
897            .handle_text(&json)
898            .expect("handler should emit deltas for a valid L2 update");
899
900        let deltas = match msg {
901            NautilusWsMessage::Deltas(d) => d,
902            other => panic!("expected Deltas, was {other:?}"),
903        };
904
905        assert!(!deltas.deltas.is_empty());
906        let expected_ts = deltas.deltas[0].ts_event;
907        for delta in &deltas.deltas {
908            assert_eq!(
909                delta.ts_event, expected_ts,
910                "all deltas in a batch must share ts_event"
911            );
912        }
913    }
914
915    #[rstest]
916    fn test_handle_l2_update_malformed_timestamp_falls_back_to_ts_init() {
917        let json = load_test_fixture("ws_l2_data_update.json")
918            .replace("2026-04-07T14:30:01.456789Z", "not-a-valid-timestamp");
919        let mut handler = test_handler();
920        handler
921            .instruments
922            .insert(btc_usd_instrument().id(), btc_usd_instrument());
923
924        let msg = handler
925            .handle_text(&json)
926            .expect("handler should still emit deltas when timestamp is malformed");
927
928        let deltas = match msg {
929            NautilusWsMessage::Deltas(d) => d,
930            other => panic!("expected Deltas, was {other:?}"),
931        };
932
933        assert!(!deltas.deltas.is_empty());
934        for delta in &deltas.deltas {
935            assert_eq!(
936                delta.ts_event, delta.ts_init,
937                "malformed timestamp must fall back to ts_init"
938            );
939        }
940    }
941
942    #[rstest]
943    fn test_handle_text_emits_futures_balance_summary_snapshot() {
944        use rust_decimal::Decimal;
945
946        let json = r#"{
947          "channel": "futures_balance_summary",
948          "client_id": "",
949          "timestamp": "2023-02-09T20:33:57.609931463Z",
950          "sequence_num": 0,
951          "events": [
952            {
953              "type": "snapshot",
954              "fcm_balance_summary": {
955                "futures_buying_power": "100.00",
956                "total_usd_balance": "200.00",
957                "cbi_usd_balance": "300.00",
958                "cfm_usd_balance": "400.00",
959                "total_open_orders_hold_amount": "500.00",
960                "unrealized_pnl": "600.00",
961                "daily_realized_pnl": "0",
962                "initial_margin": "700.00",
963                "available_margin": "800.00",
964                "liquidation_threshold": "900.00",
965                "liquidation_buffer_amount": "1000.00",
966                "liquidation_buffer_percentage": "1000",
967                "intraday_margin_window_measure": {
968                  "margin_window_type": "FCM_MARGIN_WINDOW_TYPE_INTRADAY",
969                  "margin_level": "MARGIN_LEVEL_TYPE_BASE",
970                  "initial_margin": "100.00",
971                  "maintenance_margin": "200.00",
972                  "liquidation_buffer_percentage": "1000",
973                  "total_hold": "100.00",
974                  "futures_buying_power": "400.00"
975                },
976                "overnight_margin_window_measure": {
977                  "margin_window_type": "FCM_MARGIN_WINDOW_TYPE_OVERNIGHT",
978                  "margin_level": "MARGIN_LEVEL_TYPE_BASE",
979                  "initial_margin": "300.00",
980                  "maintenance_margin": "200.00",
981                  "liquidation_buffer_percentage": "1000",
982                  "total_hold": "-30.00",
983                  "futures_buying_power": "2000.00"
984                }
985              }
986            }
987          ]
988        }"#;
989        let mut handler = test_handler();
990
991        let msg = handler
992            .handle_text(json)
993            .expect("handler should emit a futures balance summary");
994        match msg {
995            NautilusWsMessage::FuturesBalanceSummary(summary) => {
996                assert_eq!(summary.futures_buying_power, Decimal::from(100));
997                assert_eq!(summary.total_usd_balance, Decimal::from(200));
998                assert_eq!(summary.total_open_orders_hold_amount, Decimal::from(500));
999                assert_eq!(summary.available_margin, Decimal::from(800));
1000                let intraday = &summary.intraday_margin_window_measure;
1001                assert_eq!(intraday.initial_margin, Decimal::from(100));
1002                assert_eq!(intraday.maintenance_margin, Decimal::from(200));
1003                let overnight = &summary.overnight_margin_window_measure;
1004                assert_eq!(overnight.initial_margin, Decimal::from(300));
1005                assert_eq!(overnight.maintenance_margin, Decimal::from(200));
1006                // `total_hold` carries negative values on the wire; ensure
1007                // the signed decimal survives the round trip.
1008                assert_eq!(overnight.total_hold, "-30".parse::<Decimal>().unwrap());
1009            }
1010            other => panic!("expected FuturesBalanceSummary, was {other:?}"),
1011        }
1012    }
1013
1014    #[rstest]
1015    fn test_handle_text_routes_reconnected_sentinel() {
1016        let mut handler = test_handler();
1017        let result = handler.handle_text(RECONNECTED);
1018        assert!(matches!(result, Some(NautilusWsMessage::Reconnected)));
1019    }
1020
1021    #[rstest]
1022    fn test_signal_release_acquire_exits_handler_loop() {
1023        use std::sync::atomic::Ordering;
1024
1025        let signal = Arc::new(AtomicBool::new(false));
1026        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
1027        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
1028        let mut handler =
1029            FeedHandler::new(signal.clone(), cmd_rx, raw_rx, Arc::new(AtomicMap::new()));
1030
1031        signal.store(true, Ordering::Release);
1032
1033        let runtime = tokio::runtime::Builder::new_current_thread()
1034            .enable_all()
1035            .build()
1036            .unwrap();
1037        let result = runtime.block_on(async { handler.next().await });
1038        assert!(result.is_none(), "{result:?}");
1039    }
1040}
nautilus_coinbase/websocket/handler.rs

nautilus_coinbase/websocket/
handler.rs
