nautilus_coinbase/websocket/
messages.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//! WebSocket message types for the Coinbase Advanced Trade API.
17//!
18//! All incoming messages share an envelope with `channel`, `timestamp`,
19//! `sequence_num`, and a channel-specific `events` array. Outgoing
20//! subscription messages use a flat format with `type`, `product_ids`,
21//! `channel`, and `jwt`.
22
23use std::collections::HashMap;
24
25use nautilus_core::string::secret::SecretString;
26use rust_decimal::Decimal;
27use serde::{Deserialize, Serialize};
28use ustr::Ustr;
29use zeroize::Zeroize;
30
31use crate::common::{
32    enums::{
33        CoinbaseContractExpiryType, CoinbaseMarginLevel, CoinbaseMarginWindowType,
34        CoinbaseOrderSide, CoinbaseOrderStatus, CoinbaseOrderType, CoinbaseProductStatus,
35        CoinbaseProductType, CoinbaseRiskManagedBy, CoinbaseTimeInForce, CoinbaseTriggerStatus,
36        CoinbaseWsChannel,
37    },
38    parse::{
39        deserialize_decimal_from_str, deserialize_product_status_or_unknown,
40        deserialize_product_type_or_unknown,
41    },
42};
43
44/// Subscribe or unsubscribe request sent to the WebSocket.
45///
46/// Public channels (`level2`, `market_trades`, `ticker`, etc.) do not require
47/// a JWT. Set `jwt` to `None` for unauthenticated subscriptions; the field
48/// is omitted from the serialized JSON.
49#[derive(Debug, Clone, Serialize, Zeroize)]
50pub struct CoinbaseWsSubscription {
51    /// `"subscribe"` or `"unsubscribe"`.
52    #[serde(rename = "type")]
53    #[zeroize(skip)]
54    pub msg_type: CoinbaseWsAction,
55    /// Product IDs to subscribe to (omitted for channel-level subscriptions).
56    #[serde(skip_serializing_if = "Vec::is_empty")]
57    #[zeroize(skip)]
58    pub product_ids: Vec<Ustr>,
59    /// Channel name (subscription-side, e.g. `level2`).
60    #[zeroize(skip)]
61    pub channel: CoinbaseWsChannel,
62    /// JWT for authentication (required for `user` and `futures_balance_summary`).
63    #[serde(skip_serializing_if = "Option::is_none")]
64    pub jwt: Option<SecretString>,
65}
66
67/// WebSocket subscription action type.
68#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
69#[serde(rename_all = "snake_case")]
70pub enum CoinbaseWsAction {
71    Subscribe,
72    Unsubscribe,
73}
74
75/// Top-level WebSocket message dispatched by channel.
76///
77/// Uses serde internally-tagged enum on the `channel` field so each variant
78/// deserializes only the events relevant to that channel.
79#[derive(Debug, Clone, Deserialize)]
80#[serde(tag = "channel")]
81pub enum CoinbaseWsMessage {
82    /// Order book snapshot or incremental update.
83    #[serde(rename = "l2_data")]
84    L2Data {
85        timestamp: String,
86        sequence_num: u64,
87        events: Vec<WsL2DataEvent>,
88    },
89
90    /// Market trade executions.
91    #[serde(rename = "market_trades")]
92    MarketTrades {
93        timestamp: String,
94        sequence_num: u64,
95        events: Vec<WsMarketTradesEvent>,
96    },
97
98    /// Price ticker for a single product.
99    #[serde(rename = "ticker")]
100    Ticker {
101        timestamp: String,
102        sequence_num: u64,
103        events: Vec<WsTickerEvent>,
104    },
105
106    /// Batched ticker updates for multiple products.
107    #[serde(rename = "ticker_batch")]
108    TickerBatch {
109        timestamp: String,
110        sequence_num: u64,
111        events: Vec<WsTickerEvent>,
112    },
113
114    /// OHLC candle updates.
115    #[serde(rename = "candles")]
116    Candles {
117        timestamp: String,
118        sequence_num: u64,
119        events: Vec<WsCandlesEvent>,
120    },
121
122    /// User order status updates.
123    ///
124    /// The feed handler deserializes this channel but ignores it until the
125    /// execution client is wired.
126    #[serde(rename = "user")]
127    User {
128        timestamp: String,
129        sequence_num: u64,
130        events: Vec<WsUserEvent>,
131    },
132
133    /// Connection heartbeat.
134    #[serde(rename = "heartbeats")]
135    Heartbeats {
136        timestamp: String,
137        sequence_num: u64,
138        events: Vec<WsHeartbeatEvent>,
139    },
140
141    /// Futures balance summary (requires auth).
142    ///
143    /// The feed handler deserializes this channel but ignores it until account
144    /// state handling is added.
145    #[serde(rename = "futures_balance_summary")]
146    FuturesBalanceSummary {
147        timestamp: String,
148        sequence_num: u64,
149        events: Vec<WsFuturesBalanceSummaryEvent>,
150    },
151
152    /// System status updates.
153    ///
154    /// The feed handler parses each product entry into an `InstrumentStatus`
155    /// event; the data client filters emissions to subscribed instruments.
156    #[serde(rename = "status")]
157    Status {
158        timestamp: String,
159        sequence_num: u64,
160        events: Vec<WsStatusEvent>,
161    },
162
163    /// Subscription confirmation.
164    #[serde(rename = "subscriptions")]
165    Subscriptions {
166        timestamp: String,
167        sequence_num: u64,
168        events: Vec<WsSubscriptionsEvent>,
169    },
170}
171
172/// L2 data event containing book updates.
173#[derive(Debug, Clone, Deserialize)]
174pub struct WsL2DataEvent {
175    /// `"snapshot"` for initial state, `"update"` for incremental.
176    #[serde(rename = "type")]
177    pub event_type: WsEventType,
178    pub product_id: Ustr,
179    pub updates: Vec<WsL2Update>,
180}
181
182/// A single order book level update.
183#[derive(Debug, Clone, Deserialize)]
184pub struct WsL2Update {
185    pub side: WsBookSide,
186    pub event_time: String,
187    pub price_level: String,
188    pub new_quantity: String,
189}
190
191/// Book side in L2 data messages.
192#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
193#[serde(rename_all = "snake_case")]
194pub enum WsBookSide {
195    Bid,
196    Offer,
197}
198
199/// Market trades event.
200#[derive(Debug, Clone, Deserialize)]
201pub struct WsMarketTradesEvent {
202    /// `"snapshot"` or `"update"`.
203    #[serde(rename = "type")]
204    pub event_type: WsEventType,
205    pub trades: Vec<WsTrade>,
206}
207
208/// A single trade from the market_trades channel.
209#[derive(Debug, Clone, Deserialize)]
210pub struct WsTrade {
211    pub trade_id: String,
212    pub product_id: Ustr,
213    pub price: String,
214    pub size: String,
215    pub side: CoinbaseOrderSide,
216    pub time: String,
217}
218
219/// Ticker event.
220#[derive(Debug, Clone, Deserialize)]
221pub struct WsTickerEvent {
222    /// `"snapshot"` or `"update"`.
223    #[serde(rename = "type")]
224    pub event_type: WsEventType,
225    pub tickers: Vec<WsTicker>,
226}
227
228/// Ticker data for a single product.
229#[derive(Debug, Clone, Deserialize)]
230pub struct WsTicker {
231    pub product_id: Ustr,
232    pub price: String,
233    pub volume_24_h: String,
234    pub low_24_h: String,
235    pub high_24_h: String,
236    #[serde(default)]
237    pub low_52_w: String,
238    #[serde(default)]
239    pub high_52_w: String,
240    pub price_percent_chg_24_h: String,
241    pub best_bid: String,
242    pub best_bid_quantity: String,
243    pub best_ask: String,
244    pub best_ask_quantity: String,
245}
246
247/// Candles event.
248#[derive(Debug, Clone, Deserialize)]
249pub struct WsCandlesEvent {
250    /// `"snapshot"` or `"update"`.
251    #[serde(rename = "type")]
252    pub event_type: WsEventType,
253    pub candles: Vec<WsCandle>,
254}
255
256/// A single candle from the candles channel.
257#[derive(Debug, Clone, Deserialize)]
258pub struct WsCandle {
259    pub start: String,
260    pub high: String,
261    pub low: String,
262    pub open: String,
263    pub close: String,
264    pub volume: String,
265    pub product_id: Ustr,
266}
267
268/// User event containing order status updates.
269#[derive(Debug, Clone, Deserialize)]
270pub struct WsUserEvent {
271    /// `"snapshot"` or `"update"`.
272    #[serde(rename = "type")]
273    pub event_type: WsEventType,
274    pub orders: Vec<WsOrderUpdate>,
275}
276
277/// Order status update from the user channel.
278#[derive(Debug, Clone, Deserialize)]
279pub struct WsOrderUpdate {
280    pub order_id: String,
281    pub client_order_id: String,
282    pub contract_expiry_type: CoinbaseContractExpiryType,
283    pub cumulative_quantity: String,
284    pub leaves_quantity: String,
285    pub avg_price: String,
286    pub total_fees: String,
287    pub status: CoinbaseOrderStatus,
288    pub product_id: Ustr,
289    pub product_type: CoinbaseProductType,
290    pub creation_time: String,
291    pub order_side: CoinbaseOrderSide,
292    pub order_type: CoinbaseOrderType,
293    pub risk_managed_by: CoinbaseRiskManagedBy,
294    pub time_in_force: CoinbaseTimeInForce,
295    pub trigger_status: CoinbaseTriggerStatus,
296    #[serde(default)]
297    pub cancel_reason: String,
298    #[serde(default)]
299    pub reject_reason: String,
300    #[serde(default)]
301    pub total_value_after_fees: String,
302}
303
304/// Heartbeat event.
305#[derive(Debug, Clone, Deserialize)]
306pub struct WsHeartbeatEvent {
307    pub current_time: String,
308    pub heartbeat_counter: u64,
309}
310
311/// Futures balance summary event.
312#[derive(Debug, Clone, Deserialize)]
313pub struct WsFuturesBalanceSummaryEvent {
314    #[serde(rename = "type")]
315    pub event_type: WsEventType,
316    pub fcm_balance_summary: WsFcmBalanceSummary,
317}
318
319/// Futures balance summary snapshot.
320#[derive(Debug, Clone, Deserialize)]
321pub struct WsFcmBalanceSummary {
322    #[serde(deserialize_with = "deserialize_decimal_from_str")]
323    pub futures_buying_power: Decimal,
324    #[serde(deserialize_with = "deserialize_decimal_from_str")]
325    pub total_usd_balance: Decimal,
326    #[serde(deserialize_with = "deserialize_decimal_from_str")]
327    pub cbi_usd_balance: Decimal,
328    #[serde(deserialize_with = "deserialize_decimal_from_str")]
329    pub cfm_usd_balance: Decimal,
330    #[serde(deserialize_with = "deserialize_decimal_from_str")]
331    pub total_open_orders_hold_amount: Decimal,
332    #[serde(deserialize_with = "deserialize_decimal_from_str")]
333    pub unrealized_pnl: Decimal,
334    #[serde(deserialize_with = "deserialize_decimal_from_str")]
335    pub daily_realized_pnl: Decimal,
336    #[serde(deserialize_with = "deserialize_decimal_from_str")]
337    pub initial_margin: Decimal,
338    #[serde(deserialize_with = "deserialize_decimal_from_str")]
339    pub available_margin: Decimal,
340    #[serde(deserialize_with = "deserialize_decimal_from_str")]
341    pub liquidation_threshold: Decimal,
342    #[serde(deserialize_with = "deserialize_decimal_from_str")]
343    pub liquidation_buffer_amount: Decimal,
344    #[serde(deserialize_with = "deserialize_decimal_from_str")]
345    pub liquidation_buffer_percentage: Decimal,
346    pub intraday_margin_window_measure: WsMarginWindowMeasure,
347    pub overnight_margin_window_measure: WsMarginWindowMeasure,
348}
349
350/// Margin window summary inside a futures balance snapshot.
351#[derive(Debug, Clone, Deserialize)]
352pub struct WsMarginWindowMeasure {
353    pub margin_window_type: CoinbaseMarginWindowType,
354    pub margin_level: CoinbaseMarginLevel,
355    #[serde(deserialize_with = "deserialize_decimal_from_str")]
356    pub initial_margin: Decimal,
357    #[serde(deserialize_with = "deserialize_decimal_from_str")]
358    pub maintenance_margin: Decimal,
359    #[serde(deserialize_with = "deserialize_decimal_from_str")]
360    pub liquidation_buffer_percentage: Decimal,
361    #[serde(deserialize_with = "deserialize_decimal_from_str")]
362    pub total_hold: Decimal,
363    #[serde(deserialize_with = "deserialize_decimal_from_str")]
364    pub futures_buying_power: Decimal,
365}
366
367/// Status channel event.
368#[derive(Debug, Clone, Deserialize)]
369pub struct WsStatusEvent {
370    #[serde(rename = "type")]
371    pub event_type: WsEventType,
372    #[serde(default)]
373    pub products: Vec<WsStatusProduct>,
374}
375
376/// Status channel product snapshot.
377#[derive(Debug, Clone, Deserialize)]
378pub struct WsStatusProduct {
379    #[serde(deserialize_with = "deserialize_product_type_or_unknown")]
380    pub product_type: CoinbaseProductType,
381    pub id: Ustr,
382    pub base_currency: Ustr,
383    pub quote_currency: Ustr,
384    pub base_increment: String,
385    pub quote_increment: String,
386    pub display_name: String,
387    #[serde(deserialize_with = "deserialize_product_status_or_unknown")]
388    pub status: CoinbaseProductStatus,
389    pub status_message: String,
390    #[serde(deserialize_with = "deserialize_decimal_from_str")]
391    pub min_market_funds: Decimal,
392}
393
394/// Subscription confirmation event.
395#[derive(Debug, Clone, Deserialize)]
396pub struct WsSubscriptionsEvent {
397    pub subscriptions: HashMap<CoinbaseWsChannel, Vec<Ustr>>,
398}
399
400/// Event type discriminator for snapshot vs incremental update.
401#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
402#[serde(rename_all = "snake_case")]
403pub enum WsEventType {
404    Snapshot,
405    Update,
406}
407
408#[cfg(test)]
409mod tests {
410    use rstest::rstest;
411
412    use super::*;
413    use crate::common::testing::load_test_fixture;
414
415    #[rstest]
416    fn test_deserialize_l2_snapshot() {
417        let json = load_test_fixture("ws_l2_data_snapshot.json");
418        let msg: CoinbaseWsMessage = serde_json::from_str(&json).unwrap();
419
420        match msg {
421            CoinbaseWsMessage::L2Data {
422                timestamp,
423                sequence_num,
424                events,
425            } => {
426                assert!(!timestamp.is_empty());
427                assert_eq!(sequence_num, 0);
428                assert_eq!(events.len(), 1);
429
430                let event = &events[0];
431                assert_eq!(event.event_type, WsEventType::Snapshot);
432                assert_eq!(event.product_id, "BTC-USD");
433                assert!(!event.updates.is_empty());
434
435                let bid = event
436                    .updates
437                    .iter()
438                    .find(|u| u.side == WsBookSide::Bid)
439                    .expect("should have a bid update");
440                assert!(!bid.price_level.is_empty());
441                assert!(!bid.new_quantity.is_empty());
442            }
443            other => panic!("Expected L2Data, was {other:?}"),
444        }
445    }
446
447    #[rstest]
448    fn test_deserialize_l2_update() {
449        let json = load_test_fixture("ws_l2_data_update.json");
450        let msg: CoinbaseWsMessage = serde_json::from_str(&json).unwrap();
451
452        match msg {
453            CoinbaseWsMessage::L2Data {
454                sequence_num,
455                events,
456                ..
457            } => {
458                assert!(sequence_num > 0);
459                assert_eq!(events[0].event_type, WsEventType::Update);
460            }
461            other => panic!("Expected L2Data, was {other:?}"),
462        }
463    }
464
465    #[rstest]
466    fn test_deserialize_market_trades() {
467        let json = load_test_fixture("ws_market_trades.json");
468        let msg: CoinbaseWsMessage = serde_json::from_str(&json).unwrap();
469
470        match msg {
471            CoinbaseWsMessage::MarketTrades { events, .. } => {
472                assert_eq!(events.len(), 1);
473                assert!(!events[0].trades.is_empty());
474
475                let trade = &events[0].trades[0];
476                assert_eq!(trade.product_id, "BTC-USD");
477                assert!(!trade.price.is_empty());
478                assert!(!trade.size.is_empty());
479                assert!(!trade.trade_id.is_empty());
480            }
481            other => panic!("Expected MarketTrades, was {other:?}"),
482        }
483    }
484
485    #[rstest]
486    fn test_deserialize_ticker() {
487        let json = load_test_fixture("ws_ticker.json");
488        let msg: CoinbaseWsMessage = serde_json::from_str(&json).unwrap();
489
490        match msg {
491            CoinbaseWsMessage::Ticker { events, .. } => {
492                assert_eq!(events.len(), 1);
493                assert!(!events[0].tickers.is_empty());
494
495                let ticker = &events[0].tickers[0];
496                assert_eq!(ticker.product_id, "BTC-USD");
497                assert!(!ticker.best_bid.is_empty());
498                assert!(!ticker.best_ask.is_empty());
499                assert!(!ticker.best_bid_quantity.is_empty());
500                assert!(!ticker.best_ask_quantity.is_empty());
501            }
502            other => panic!("Expected Ticker, was {other:?}"),
503        }
504    }
505
506    #[rstest]
507    fn test_deserialize_candles() {
508        let json = load_test_fixture("ws_candles.json");
509        let msg: CoinbaseWsMessage = serde_json::from_str(&json).unwrap();
510
511        match msg {
512            CoinbaseWsMessage::Candles { events, .. } => {
513                assert_eq!(events.len(), 1);
514                assert!(!events[0].candles.is_empty());
515
516                let candle = &events[0].candles[0];
517                assert_eq!(candle.product_id, "BTC-USD");
518                assert!(!candle.open.is_empty());
519                assert!(!candle.high.is_empty());
520                assert!(!candle.low.is_empty());
521                assert!(!candle.close.is_empty());
522                assert!(!candle.volume.is_empty());
523            }
524            other => panic!("Expected Candles, was {other:?}"),
525        }
526    }
527
528    #[rstest]
529    fn test_deserialize_user_order_update() {
530        let json = load_test_fixture("ws_user.json");
531        let msg: CoinbaseWsMessage = serde_json::from_str(&json).unwrap();
532
533        match msg {
534            CoinbaseWsMessage::User { events, .. } => {
535                assert_eq!(events.len(), 1);
536                assert!(!events[0].orders.is_empty());
537
538                let order = &events[0].orders[0];
539                assert!(!order.order_id.is_empty());
540                assert_eq!(order.product_id, "BTC-USD");
541                assert_eq!(order.status, CoinbaseOrderStatus::Open);
542                assert_eq!(order.order_side, CoinbaseOrderSide::Buy);
543                assert_eq!(order.order_type, CoinbaseOrderType::Limit);
544                assert_eq!(
545                    order.contract_expiry_type,
546                    CoinbaseContractExpiryType::Unknown
547                );
548                assert_eq!(order.product_type, CoinbaseProductType::Spot);
549                assert_eq!(order.risk_managed_by, CoinbaseRiskManagedBy::Unknown);
550                assert_eq!(order.time_in_force, CoinbaseTimeInForce::GoodUntilCancelled);
551                assert_eq!(
552                    order.trigger_status,
553                    CoinbaseTriggerStatus::InvalidOrderType
554                );
555            }
556            other => panic!("Expected User, was {other:?}"),
557        }
558    }
559
560    #[rstest]
561    fn test_deserialize_heartbeat() {
562        let json = load_test_fixture("ws_heartbeats.json");
563        let msg: CoinbaseWsMessage = serde_json::from_str(&json).unwrap();
564
565        match msg {
566            CoinbaseWsMessage::Heartbeats { events, .. } => {
567                assert_eq!(events.len(), 1);
568                assert!(!events[0].current_time.is_empty());
569                assert!(events[0].heartbeat_counter > 0);
570            }
571            other => panic!("Expected Heartbeats, was {other:?}"),
572        }
573    }
574
575    #[rstest]
576    fn test_deserialize_status_channel() {
577        let json = r#"{
578          "channel": "status",
579          "client_id": "",
580          "timestamp": "2023-02-09T20:29:49.753424311Z",
581          "sequence_num": 0,
582          "events": [
583            {
584              "type": "snapshot",
585              "products": [
586                {
587                  "product_type": "SPOT",
588                  "id": "BTC-USD",
589                  "base_currency": "BTC",
590                  "quote_currency": "USD",
591                  "base_increment": "0.00000001",
592                  "quote_increment": "0.01",
593                  "display_name": "BTC/USD",
594                  "status": "online",
595                  "status_message": "",
596                  "min_market_funds": "1"
597                }
598              ]
599            }
600          ]
601        }"#;
602        let msg: CoinbaseWsMessage = serde_json::from_str(json).unwrap();
603
604        match msg {
605            CoinbaseWsMessage::Status { events, .. } => {
606                assert_eq!(events.len(), 1);
607                assert_eq!(events[0].event_type, WsEventType::Snapshot);
608                assert_eq!(events[0].products.len(), 1);
609                let product = &events[0].products[0];
610                assert_eq!(product.id, "BTC-USD");
611                assert_eq!(product.product_type, CoinbaseProductType::Spot);
612                assert_eq!(product.status, CoinbaseProductStatus::Online);
613                assert_eq!(product.min_market_funds, Decimal::ONE);
614            }
615            other => panic!("Expected Status, was {other:?}"),
616        }
617    }
618
619    #[rstest]
620    fn test_deserialize_status_channel_unknown_status_does_not_fail() {
621        // Coinbase can introduce new product status values; deserialization must
622        // not hard-fail on an unrecognized status (e.g. a future "auction" state).
623        let json = r#"{
624          "channel": "status",
625          "client_id": "",
626          "timestamp": "2023-02-09T20:29:49.753424311Z",
627          "sequence_num": 0,
628          "events": [
629            {
630              "type": "snapshot",
631              "products": [
632                {
633                  "product_type": "SPOT",
634                  "id": "BTC-USD",
635                  "base_currency": "BTC",
636                  "quote_currency": "USD",
637                  "base_increment": "0.00000001",
638                  "quote_increment": "0.01",
639                  "display_name": "BTC/USD",
640                  "status": "auction",
641                  "status_message": "",
642                  "min_market_funds": "1"
643                }
644              ]
645            }
646          ]
647        }"#;
648        let msg: CoinbaseWsMessage = serde_json::from_str(json).unwrap();
649
650        match msg {
651            CoinbaseWsMessage::Status { events, .. } => {
652                assert_eq!(events[0].products[0].status, CoinbaseProductStatus::Unknown);
653            }
654            other => panic!("Expected Status, was {other:?}"),
655        }
656    }
657
658    #[rstest]
659    fn test_deserialize_status_channel_unknown_product_type_does_not_fail() {
660        // A new product type on the status channel must fall back to Unknown rather
661        // than hard-failing deserialization of the whole status message.
662        let json = r#"{
663          "channel": "status",
664          "client_id": "",
665          "timestamp": "2023-02-09T20:29:49.753424311Z",
666          "sequence_num": 0,
667          "events": [
668            {
669              "type": "snapshot",
670              "products": [
671                {
672                  "product_type": "PERPETUAL",
673                  "id": "BTC-PERP",
674                  "base_currency": "BTC",
675                  "quote_currency": "USD",
676                  "base_increment": "0.00000001",
677                  "quote_increment": "0.01",
678                  "display_name": "BTC-PERP",
679                  "status": "online",
680                  "status_message": "",
681                  "min_market_funds": "1"
682                }
683              ]
684            }
685          ]
686        }"#;
687        let msg: CoinbaseWsMessage = serde_json::from_str(json).unwrap();
688
689        match msg {
690            CoinbaseWsMessage::Status { events, .. } => {
691                assert_eq!(
692                    events[0].products[0].product_type,
693                    CoinbaseProductType::Unknown
694                );
695            }
696            other => panic!("Expected Status, was {other:?}"),
697        }
698    }
699
700    #[rstest]
701    fn test_deserialize_futures_balance_summary_channel() {
702        let json = r#"{
703          "channel": "futures_balance_summary",
704          "client_id": "",
705          "timestamp": "2023-02-09T20:33:57.609931463Z",
706          "sequence_num": 0,
707          "events": [
708            {
709              "type": "snapshot",
710              "fcm_balance_summary": {
711                "futures_buying_power": "100.00",
712                "total_usd_balance": "200.00",
713                "cbi_usd_balance": "300.00",
714                "cfm_usd_balance": "400.00",
715                "total_open_orders_hold_amount": "500.00",
716                "unrealized_pnl": "600.00",
717                "daily_realized_pnl": "0",
718                "initial_margin": "700.00",
719                "available_margin": "800.00",
720                "liquidation_threshold": "900.00",
721                "liquidation_buffer_amount": "1000.00",
722                "liquidation_buffer_percentage": "1000",
723                "intraday_margin_window_measure": {
724                  "margin_window_type": "FCM_MARGIN_WINDOW_TYPE_INTRADAY",
725                  "margin_level": "MARGIN_LEVEL_TYPE_BASE",
726                  "initial_margin": "100.00",
727                  "maintenance_margin": "200.00",
728                  "liquidation_buffer_percentage": "1000",
729                  "total_hold": "100.00",
730                  "futures_buying_power": "400.00"
731                },
732                "overnight_margin_window_measure": {
733                  "margin_window_type": "FCM_MARGIN_WINDOW_TYPE_OVERNIGHT",
734                  "margin_level": "MARGIN_LEVEL_TYPE_BASE",
735                  "initial_margin": "300.00",
736                  "maintenance_margin": "200.00",
737                  "liquidation_buffer_percentage": "1000",
738                  "total_hold": "-30.00",
739                  "futures_buying_power": "2000.00"
740                }
741              }
742            }
743          ]
744        }"#;
745        let msg: CoinbaseWsMessage = serde_json::from_str(json).unwrap();
746
747        match msg {
748            CoinbaseWsMessage::FuturesBalanceSummary { events, .. } => {
749                assert_eq!(events.len(), 1);
750                assert_eq!(events[0].event_type, WsEventType::Snapshot);
751                let summary = &events[0].fcm_balance_summary;
752                assert_eq!(summary.futures_buying_power, Decimal::from(100));
753                assert_eq!(summary.daily_realized_pnl, Decimal::ZERO);
754                assert_eq!(
755                    summary.intraday_margin_window_measure.margin_window_type,
756                    CoinbaseMarginWindowType::Intraday
757                );
758                assert_eq!(
759                    summary.overnight_margin_window_measure.margin_level,
760                    CoinbaseMarginLevel::Base
761                );
762                assert_eq!(
763                    summary.overnight_margin_window_measure.total_hold,
764                    "-30.00".parse::<Decimal>().unwrap()
765                );
766            }
767            other => panic!("Expected FuturesBalanceSummary, was {other:?}"),
768        }
769    }
770
771    #[rstest]
772    fn test_deserialize_subscriptions() {
773        let json = load_test_fixture("ws_subscriptions.json");
774        let msg: CoinbaseWsMessage = serde_json::from_str(&json).unwrap();
775
776        match msg {
777            CoinbaseWsMessage::Subscriptions { events, .. } => {
778                assert_eq!(events.len(), 1);
779                assert_eq!(
780                    events[0].subscriptions.get(&CoinbaseWsChannel::Level2),
781                    Some(&vec![Ustr::from("BTC-USD")])
782                );
783                assert_eq!(
784                    events[0]
785                        .subscriptions
786                        .get(&CoinbaseWsChannel::MarketTrades),
787                    Some(&vec![Ustr::from("BTC-USD"), Ustr::from("ETH-USD")])
788                );
789            }
790            other => panic!("Expected Subscriptions, was {other:?}"),
791        }
792    }
793
794    #[rstest]
795    fn test_serialize_subscribe_request_with_jwt() {
796        let sub = CoinbaseWsSubscription {
797            msg_type: CoinbaseWsAction::Subscribe,
798            product_ids: vec![Ustr::from("BTC-USD")],
799            channel: CoinbaseWsChannel::User,
800            jwt: Some(SecretString::from("test-jwt-token")),
801        };
802
803        let json = serde_json::to_value(&sub).unwrap();
804        assert_eq!(json["type"], "subscribe");
805        assert_eq!(json["channel"], "user");
806        assert_eq!(json["product_ids"][0], "BTC-USD");
807        assert_eq!(json["jwt"], "test-jwt-token");
808    }
809
810    #[rstest]
811    fn test_serialize_subscribe_request_public_omits_jwt() {
812        let sub = CoinbaseWsSubscription {
813            msg_type: CoinbaseWsAction::Subscribe,
814            product_ids: vec![Ustr::from("BTC-USD")],
815            channel: CoinbaseWsChannel::Level2,
816            jwt: None,
817        };
818
819        let json = serde_json::to_value(&sub).unwrap();
820        assert_eq!(json["type"], "subscribe");
821        assert_eq!(json["channel"], "level2");
822        assert!(json.get("jwt").is_none());
823    }
824
825    #[rstest]
826    fn test_serialize_unsubscribe_request() {
827        let sub = CoinbaseWsSubscription {
828            msg_type: CoinbaseWsAction::Unsubscribe,
829            product_ids: vec![Ustr::from("ETH-USD")],
830            channel: CoinbaseWsChannel::MarketTrades,
831            jwt: None,
832        };
833
834        let json = serde_json::to_value(&sub).unwrap();
835        assert_eq!(json["type"], "unsubscribe");
836        assert_eq!(json["channel"], "market_trades");
837        assert!(json.get("jwt").is_none());
838    }
839
840    #[rstest]
841    fn test_serialize_channel_level_subscription_omits_product_ids() {
842        let sub = CoinbaseWsSubscription {
843            msg_type: CoinbaseWsAction::Subscribe,
844            product_ids: vec![],
845            channel: CoinbaseWsChannel::Heartbeats,
846            jwt: None,
847        };
848
849        let json = serde_json::to_value(&sub).unwrap();
850        assert_eq!(json["type"], "subscribe");
851        assert_eq!(json["channel"], "heartbeats");
852        assert!(json.get("product_ids").is_none());
853        assert!(json.get("jwt").is_none());
854    }
855
856    #[rstest]
857    fn test_ws_event_type_values() {
858        let snapshot: WsEventType = serde_json::from_str("\"snapshot\"").unwrap();
859        assert_eq!(snapshot, WsEventType::Snapshot);
860
861        let update: WsEventType = serde_json::from_str("\"update\"").unwrap();
862        assert_eq!(update, WsEventType::Update);
863    }
864
865    #[rstest]
866    fn test_ws_book_side_values() {
867        let bid: WsBookSide = serde_json::from_str("\"bid\"").unwrap();
868        assert_eq!(bid, WsBookSide::Bid);
869
870        let offer: WsBookSide = serde_json::from_str("\"offer\"").unwrap();
871        assert_eq!(offer, WsBookSide::Offer);
872    }
873}
nautilus_coinbase/websocket/messages.rs

nautilus_coinbase/websocket/
messages.rs
