nautilus_coinbase_intx/websocket/
messages.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2025 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
16use chrono::{DateTime, Utc};
17use nautilus_model::{
18    data::{Data, IndexPriceUpdate, MarkPriceUpdate, OrderBookDeltas},
19    events::OrderEventAny,
20    instruments::InstrumentAny,
21};
22use serde::{Deserialize, Serialize};
23use ustr::Ustr;
24
25use super::enums::{CoinbaseIntxWsChannel, WsMessageType, WsOperation};
26use crate::common::enums::{CoinbaseIntxInstrumentType, CoinbaseIntxSide};
27
28#[derive(Debug, Clone)]
29pub enum NautilusWsMessage {
30    Data(Data),
31    DataVec(Vec<Data>),
32    Deltas(OrderBookDeltas),
33    Instrument(InstrumentAny),
34    OrderEvent(OrderEventAny),
35    MarkPrice(MarkPriceUpdate),
36    IndexPrice(IndexPriceUpdate),
37    MarkAndIndex((MarkPriceUpdate, IndexPriceUpdate)),
38}
39
40#[derive(Debug, Serialize)]
41pub struct CoinbaseIntxSubscription {
42    #[serde(rename = "type")]
43    pub op: WsOperation,
44    #[serde(skip_serializing_if = "Option::is_none")]
45    pub product_ids: Option<Vec<Ustr>>,
46    pub channels: Vec<CoinbaseIntxWsChannel>,
47    pub time: String,
48    pub key: Ustr,
49    pub passphrase: Ustr,
50    pub signature: String,
51}
52
53#[derive(Debug, Deserialize)]
54#[serde(untagged)]
55#[allow(clippy::large_enum_variant)]
56pub enum CoinbaseIntxWsMessage {
57    Reject(CoinbaseIntxWsRejectMsg),
58    Confirmation(CoinbaseIntxWsConfirmationMsg),
59    Instrument(CoinbaseIntxWsInstrumentMsg),
60    Funding(CoinbaseIntxWsFundingMsg),
61    Risk(CoinbaseIntxWsRiskMsg),
62    BookSnapshot(CoinbaseIntxWsOrderBookSnapshotMsg),
63    BookUpdate(CoinbaseIntxWsOrderBookUpdateMsg),
64    Quote(CoinbaseIntxWsQuoteMsg),
65    Trade(CoinbaseIntxWsTradeMsg),
66    CandleSnapshot(CoinbaseIntxWsCandleSnapshotMsg),
67    CandleUpdate(CoinbaseIntxWsCandleUpdateMsg),
68}
69
70#[derive(Debug, Deserialize)]
71pub struct CoinbaseIntxWsRejectMsg {
72    pub message: String,
73    pub reason: String,
74    pub channel: CoinbaseIntxWsChannel,
75}
76
77#[derive(Debug, Deserialize)]
78pub struct CoinbaseIntxWsConfirmationMsg {
79    pub channels: Vec<CoinbaseIntxWsChannelDetails>,
80    pub authenticated: bool,
81    pub channel: CoinbaseIntxWsChannel,
82    pub time: DateTime<Utc>,
83}
84
85#[derive(Debug, Deserialize)]
86pub struct CoinbaseIntxWsChannelDetails {
87    pub name: CoinbaseIntxWsChannel,
88    pub product_ids: Option<Vec<Ustr>>,
89}
90
91#[derive(Debug, Deserialize)]
92pub struct CoinbaseIntxWsInstrumentMsg {
93    #[serde(rename = "type")]
94    pub message_type: WsMessageType,
95    pub channel: CoinbaseIntxWsChannel,
96    pub product_id: Ustr,
97    pub instrument_type: CoinbaseIntxInstrumentType,
98    pub instrument_mode: String,
99    pub base_asset_name: String,
100    pub quote_asset_name: String,
101    pub base_increment: String,
102    pub quote_increment: String,
103    pub avg_daily_quantity: String,
104    pub avg_daily_volume: String,
105    pub total30_day_quantity: String,
106    pub total30_day_volume: String,
107    pub total24_hour_quantity: String,
108    pub total24_hour_volume: String,
109    pub base_imf: String,
110    pub min_quantity: String,
111    pub position_size_limit: Option<String>,
112    pub position_notional_limit: Option<String>,
113    pub funding_interval: Option<String>,
114    pub trading_state: String,
115    pub last_updated_time: DateTime<Utc>,
116    pub default_initial_margin: Option<String>,
117    pub base_asset_multiplier: String,
118    pub underlying_type: CoinbaseIntxInstrumentType,
119    pub sequence: u64,
120    pub time: DateTime<Utc>,
121}
122
123#[derive(Debug, Deserialize)]
124pub struct CoinbaseIntxWsFundingMsg {
125    #[serde(rename = "type")]
126    pub message_type: WsMessageType,
127    pub channel: CoinbaseIntxWsChannel,
128    pub product_id: Ustr,
129    pub funding_rate: String,
130    pub is_final: bool,
131    pub sequence: u64,
132    pub time: DateTime<Utc>,
133}
134
135#[derive(Debug, Deserialize)]
136pub struct CoinbaseIntxWsRiskMsg {
137    #[serde(rename = "type")]
138    pub message_type: WsMessageType,
139    pub channel: CoinbaseIntxWsChannel,
140    pub product_id: Ustr,
141    pub limit_up: String,
142    pub limit_down: String,
143    pub index_price: String,
144    pub mark_price: String,
145    pub settlement_price: String,
146    pub open_interest: String,
147    pub sequence: u64,
148    pub time: DateTime<Utc>,
149}
150
151#[derive(Debug, Deserialize)]
152pub struct CoinbaseIntxWsOrderBookSnapshotMsg {
153    #[serde(rename = "type")]
154    pub message_type: WsMessageType,
155    pub channel: CoinbaseIntxWsChannel,
156    pub product_id: Ustr,
157    pub bids: Vec<[String; 2]>, // [price, size]
158    pub asks: Vec<[String; 2]>, // [price, size]
159    pub sequence: u64,
160    pub time: DateTime<Utc>,
161}
162
163#[derive(Debug, Deserialize)]
164pub struct CoinbaseIntxWsOrderBookUpdateMsg {
165    #[serde(rename = "type")]
166    pub message_type: WsMessageType,
167    pub channel: CoinbaseIntxWsChannel,
168    pub product_id: Ustr,
169    pub changes: Vec<[String; 3]>, // [side, price, size]
170    pub sequence: u64,
171    pub time: DateTime<Utc>,
172}
173
174#[derive(Debug, Deserialize)]
175pub struct CoinbaseIntxWsQuoteMsg {
176    #[serde(rename = "type")]
177    pub message_type: WsMessageType,
178    pub channel: CoinbaseIntxWsChannel,
179    pub product_id: Ustr,
180    pub bid_price: String,
181    pub bid_qty: String,
182    pub ask_price: String,
183    pub ask_qty: String,
184    pub sequence: u64,
185    pub time: DateTime<Utc>,
186}
187
188#[derive(Debug, Deserialize)]
189pub struct CoinbaseIntxWsTradeMsg {
190    #[serde(rename = "type")]
191    pub message_type: WsMessageType,
192    pub channel: CoinbaseIntxWsChannel,
193    pub product_id: Ustr,
194    pub match_id: String,
195    pub trade_price: String,
196    pub trade_qty: String,
197    pub aggressor_side: CoinbaseIntxSide,
198    pub sequence: u64,
199    pub time: DateTime<Utc>,
200}
201
202#[derive(Debug, Deserialize)]
203pub struct CoinbaseIntxWsCandle {
204    pub start: DateTime<Utc>,
205    pub open: String,
206    pub high: String,
207    pub low: String,
208    pub close: String,
209    pub volume: String,
210}
211
212#[derive(Debug, Deserialize)]
213pub struct CoinbaseIntxWsCandleSnapshotMsg {
214    #[serde(rename = "type")]
215    pub message_type: WsMessageType,
216    pub channel: CoinbaseIntxWsChannel,
217    pub product_id: Ustr,
218    pub granularity: Ustr,
219    pub candles: Vec<CoinbaseIntxWsCandle>,
220    pub sequence: u64,
221}
222
223#[derive(Debug, Deserialize)]
224pub struct CoinbaseIntxWsCandleUpdateMsg {
225    #[serde(rename = "type")]
226    pub message_type: WsMessageType,
227    pub channel: CoinbaseIntxWsChannel,
228    pub product_id: Ustr,
229    pub start: DateTime<Utc>,
230    #[serde(default)]
231    pub open: Option<String>,
232    #[serde(default)]
233    pub high: Option<String>,
234    #[serde(default)]
235    pub low: Option<String>,
236    #[serde(default)]
237    pub close: Option<String>,
238    #[serde(default)]
239    pub volume: Option<String>,
240    pub sequence: u64,
241}
242
243#[cfg(test)]
244mod tests {
245    use rstest::rstest;
246
247    use super::*;
248    use crate::common::testing::load_test_json;
249
250    #[rstest]
251    fn test_parse_asset_model() {
252        let json_data = load_test_json("ws_instruments.json");
253        let parsed: CoinbaseIntxWsInstrumentMsg = serde_json::from_str(&json_data).unwrap();
254
255        assert_eq!(parsed.product_id, "ETH-PERP");
256        assert_eq!(parsed.message_type, WsMessageType::Snapshot);
257        assert_eq!(parsed.channel, CoinbaseIntxWsChannel::Instruments);
258        assert_eq!(parsed.instrument_type, CoinbaseIntxInstrumentType::Perp);
259        assert_eq!(parsed.instrument_mode, "standard");
260        assert_eq!(parsed.base_asset_name, "ETH");
261        assert_eq!(parsed.quote_asset_name, "USDC");
262        assert_eq!(parsed.base_increment, "0.0001");
263        assert_eq!(parsed.quote_increment, "0.01");
264        assert_eq!(parsed.avg_daily_quantity, "229061.15400333333");
265        assert_eq!(parsed.avg_daily_volume, "5.33931093731498E8");
266        assert_eq!(parsed.total30_day_quantity, "6871834.6201");
267        assert_eq!(parsed.total30_day_volume, "1.601793281194494E10");
268        assert_eq!(parsed.total24_hour_quantity, "116705.0261");
269        assert_eq!(parsed.total24_hour_volume, "2.22252453944151E8");
270        assert_eq!(parsed.base_imf, "0.05");
271        assert_eq!(parsed.min_quantity, "0.0001");
272        assert_eq!(parsed.position_size_limit, Some("5841.0594".to_string()));
273        assert_eq!(parsed.position_notional_limit, Some("70000000".to_string()));
274        assert_eq!(parsed.funding_interval, Some("3600000000000".to_string()));
275        assert_eq!(parsed.trading_state, "trading");
276        assert_eq!(
277            parsed.last_updated_time.to_rfc3339(),
278            "2025-03-14T22:00:00+00:00"
279        );
280        assert_eq!(parsed.default_initial_margin, Some("0.2".to_string()));
281        assert_eq!(parsed.base_asset_multiplier, "1.0");
282        assert_eq!(parsed.underlying_type, CoinbaseIntxInstrumentType::Spot);
283        assert_eq!(parsed.sequence, 0);
284        assert_eq!(parsed.time.to_rfc3339(), "2025-03-14T22:59:53.373+00:00");
285    }
286
287    #[rstest]
288    fn test_parse_ws_trade_msg() {
289        let json_data = load_test_json("ws_match.json");
290        let parsed: CoinbaseIntxWsTradeMsg = serde_json::from_str(&json_data).unwrap();
291
292        assert_eq!(parsed.product_id, "BTC-PERP");
293        assert_eq!(parsed.message_type, WsMessageType::Update);
294        assert_eq!(parsed.channel, CoinbaseIntxWsChannel::Match);
295        assert_eq!(parsed.match_id, "423596942694547460");
296        assert_eq!(parsed.trade_price, "84374");
297        assert_eq!(parsed.trade_qty, "0.0213");
298        assert_eq!(parsed.aggressor_side, CoinbaseIntxSide::Buy);
299        assert_eq!(parsed.sequence, 0);
300        assert_eq!(parsed.time.to_rfc3339(), "2025-03-14T23:03:01.189+00:00");
301    }
302
303    #[rstest]
304    fn test_parse_ws_quote_msg() {
305        let json_data = load_test_json("ws_quote.json");
306        let parsed: CoinbaseIntxWsQuoteMsg = serde_json::from_str(&json_data).unwrap();
307
308        assert_eq!(parsed.product_id, "BTC-PERP");
309        assert_eq!(parsed.message_type, WsMessageType::Snapshot);
310        assert_eq!(parsed.channel, CoinbaseIntxWsChannel::Level1);
311        assert_eq!(parsed.bid_price, "84368.5");
312        assert_eq!(parsed.bid_qty, "2.608");
313        assert_eq!(parsed.ask_price, "84368.6");
314        assert_eq!(parsed.ask_qty, "2.9453");
315        assert_eq!(parsed.sequence, 0);
316        assert_eq!(parsed.time.to_rfc3339(), "2025-03-14T23:05:39.533+00:00");
317    }
318
319    #[rstest]
320    fn test_parse_ws_order_book_snapshot_msg() {
321        let json_data = load_test_json("ws_book_snapshot.json");
322        let parsed: CoinbaseIntxWsOrderBookSnapshotMsg = serde_json::from_str(&json_data).unwrap();
323
324        assert_eq!(parsed.product_id, "BTC-PERP");
325        assert_eq!(parsed.message_type, WsMessageType::Snapshot);
326        assert_eq!(parsed.channel, CoinbaseIntxWsChannel::Level2);
327        assert_eq!(parsed.sequence, 0);
328        assert_eq!(parsed.time.to_rfc3339(), "2025-03-14T23:09:43.993+00:00");
329
330        assert_eq!(parsed.bids.len(), 50);
331        assert_eq!(parsed.asks.len(), 50);
332
333        assert_eq!(parsed.bids[0][0], "84323.6");
334        assert_eq!(parsed.bids[0][1], "4.9466");
335
336        assert_eq!(parsed.bids[49][0], "84296.2");
337        assert_eq!(parsed.bids[49][1], "0.0237");
338
339        assert_eq!(parsed.asks[0][0], "84323.7");
340        assert_eq!(parsed.asks[0][1], "2.6944");
341
342        assert_eq!(parsed.asks[49][0], "84346.9");
343        assert_eq!(parsed.asks[49][1], "0.3257");
344    }
345
346    #[rstest]
347    fn test_parse_ws_order_book_update_msg() {
348        let json_data = load_test_json("ws_book_update.json");
349        let parsed: CoinbaseIntxWsOrderBookUpdateMsg = serde_json::from_str(&json_data).unwrap();
350
351        assert_eq!(parsed.product_id, "BTC-PERP");
352        assert_eq!(parsed.message_type, WsMessageType::Update);
353        assert_eq!(parsed.channel, CoinbaseIntxWsChannel::Level2);
354        assert_eq!(parsed.sequence, 1);
355        assert_eq!(parsed.time.to_rfc3339(), "2025-03-14T23:09:44.095+00:00");
356
357        assert_eq!(parsed.changes.len(), 2);
358
359        assert_eq!(parsed.changes[0][0], "BUY"); // side
360        assert_eq!(parsed.changes[0][1], "84296.2"); // price
361        assert_eq!(parsed.changes[0][2], "0"); // size (0 means delete)
362
363        assert_eq!(parsed.changes[1][0], "BUY"); // side
364        assert_eq!(parsed.changes[1][1], "84296.3"); // price
365        assert_eq!(parsed.changes[1][2], "0.1779"); // size
366    }
367
368    #[rstest]
369    fn test_parse_ws_candle_snapshot_msg() {
370        let json_data = load_test_json("ws_candles.json");
371        let parsed: CoinbaseIntxWsCandleSnapshotMsg = serde_json::from_str(&json_data).unwrap();
372
373        assert_eq!(parsed.granularity, "ONE_MINUTE");
374        assert_eq!(parsed.sequence, 0);
375        assert_eq!(parsed.candles.len(), 1);
376
377        let candle = &parsed.candles[0];
378        assert_eq!(candle.start.to_rfc3339(), "2025-03-14T23:14:00+00:00");
379        assert_eq!(candle.open, "1921.71");
380        assert_eq!(candle.high, "1921.71");
381        assert_eq!(candle.low, "1919.87");
382        assert_eq!(candle.close, "1919.87");
383        assert_eq!(candle.volume, "11.2803");
384    }
385}
nautilus_coinbase_intx/websocket/messages.rs

nautilus_coinbase_intx/websocket/
messages.rs
