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

nautilus_coinbase/websocket/
handler.rs
