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

nautilus_coinbase/websocket/
messages.rs
