nautilus_coinapi/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//! Raw CoinAPI WebSocket V1 frames, decoded directly from JSON bytes to preserve decimal tokens.
17
18use jiff::Timestamp;
19use nautilus_core::serialization::decimal;
20use rust_decimal::Decimal;
21use serde::{Deserialize, Serialize};
22use ustr::Ustr;
23
24use super::error::Error;
25use crate::http::models::BookLevel;
26
27/// A market data stream type accepted in `subscribe_data_type`.
28#[derive(
29    Debug,
30    Clone,
31    Copy,
32    PartialEq,
33    Eq,
34    Hash,
35    Serialize,
36    Deserialize,
37    strum::AsRefStr,
38    strum::EnumString,
39)]
40#[serde(rename_all = "snake_case")]
41#[strum(serialize_all = "snake_case")]
42pub enum DataType {
43    Book,
44    Book5,
45    Book20,
46    Book50,
47    BookL3,
48    Quote,
49    Trade,
50    Ohlcv,
51}
52
53impl DataType {
54    /// Returns whether this is an order book type.
55    ///
56    /// A session scope can list only one book type; CoinAPI rejects a second with a terminal
57    /// `error` frame.
58    #[must_use]
59    pub fn is_book(self) -> bool {
60        matches!(
61            self,
62            Self::Book | Self::Book5 | Self::Book20 | Self::Book50 | Self::BookL3
63        )
64    }
65}
66
67/// A data type, plus its period for OHLCV.
68#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
69pub struct Channel {
70    pub data_type: DataType,
71    /// Required for and only valid with [`DataType::Ohlcv`], for example `1MIN`.
72    pub period_id: Option<Ustr>,
73}
74
75impl Channel {
76    pub(super) fn validate(&self) -> Result<(), Error> {
77        match (self.data_type, self.period_id) {
78            (DataType::Ohlcv, Some(period))
79                if !period.is_empty() && period.bytes().all(|b| b.is_ascii_alphanumeric()) =>
80            {
81                Ok(())
82            }
83            (DataType::Ohlcv, _) => Err(Error::Validation(
84                "OHLCV requires an alphanumeric period_id".into(),
85            )),
86            (_, Some(_)) => Err(Error::Validation(
87                "period_id is only valid for OHLCV".into(),
88            )),
89            (_, None) => Ok(()),
90        }
91    }
92}
93
94/// One symbol on one channel.
95#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
96pub struct Subscription {
97    pub channel: Channel,
98    /// Full CoinAPI `symbol_id`, matched exactly.
99    pub symbol_id: Ustr,
100}
101
102impl Subscription {
103    pub(super) fn validate(&self) -> Result<(), Error> {
104        self.channel.validate()?;
105
106        if self.symbol_id.is_empty()
107            || !self
108                .symbol_id
109                .bytes()
110                .all(|b| (33..=126).contains(&b) && b != b'$')
111        {
112            return Err(Error::Validation(
113                "symbol_id must contain visible ASCII characters other than '$'".into(),
114            ));
115        }
116
117        Ok(())
118    }
119}
120
121/// A change to the session scope, applied in order with others that share one `hello`.
122#[derive(Debug, Clone, Copy, PartialEq, Eq)]
123pub enum ScopeChange {
124    Subscribe(Subscription),
125    Unsubscribe(Subscription),
126}
127
128impl ScopeChange {
129    /// Returns the pair the change adds or removes.
130    #[must_use]
131    pub fn subscription(&self) -> &Subscription {
132        match self {
133            Self::Subscribe(subscription) | Self::Unsubscribe(subscription) => subscription,
134        }
135    }
136}
137
138/// A `hello` request that replaces the whole session scope.
139///
140/// CoinAPI combines every listed data type, symbol, and period, so the scope can include
141/// pairs nobody subscribed to. Authentication uses the `X-CoinAPI-Key` upgrade header, so this
142/// frame carries no credential.
143#[derive(Debug, Serialize)]
144pub(super) struct Hello {
145    r#type: &'static str,
146    heartbeat: bool,
147    subscribe_data_type: Vec<DataType>,
148    subscribe_filter_symbol_id: Vec<String>,
149    #[serde(skip_serializing_if = "Vec::is_empty")]
150    subscribe_filter_period_id: Vec<Ustr>,
151}
152
153impl Hello {
154    /// Builds the smallest exact-symbol scope covering every subscription, in sorted order.
155    pub(super) fn new(subscriptions: &[Subscription]) -> Self {
156        let mut data_types: Vec<DataType> =
157            subscriptions.iter().map(|s| s.channel.data_type).collect();
158        data_types.sort_unstable_by(|a, b| a.as_ref().cmp(b.as_ref()));
159        data_types.dedup();
160
161        let mut symbols: Vec<&str> = subscriptions.iter().map(|s| s.symbol_id.as_str()).collect();
162        symbols.sort_unstable();
163        symbols.dedup();
164
165        let mut periods: Vec<Ustr> = subscriptions
166            .iter()
167            .filter_map(|s| s.channel.period_id)
168            .collect();
169        periods.sort_unstable_by(|a, b| a.as_str().cmp(b.as_str()));
170        periods.dedup();
171
172        Self {
173            r#type: "hello",
174            heartbeat: true,
175            subscribe_data_type: data_types,
176            subscribe_filter_symbol_id: symbols.iter().map(|s| format!("{s}$")).collect(),
177            subscribe_filter_period_id: periods,
178        }
179    }
180}
181
182/// An L2 book message from `book`, `book5`, `book20`, or `book50`.
183///
184/// Only `book` carries `is_snapshot`; the fixed-depth types always list their top levels.
185#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
186pub struct Book {
187    #[serde(rename = "type")]
188    pub data_type: DataType,
189    pub symbol_id: Ustr,
190    pub sequence: u64,
191    pub time_exchange: Timestamp,
192    pub time_coinapi: Timestamp,
193    pub is_snapshot: Option<bool>,
194    pub asks: Vec<BookLevel>,
195    pub bids: Vec<BookLevel>,
196}
197
198/// An order-by-order book snapshot or update.
199#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
200pub struct BookL3 {
201    pub symbol_id: Ustr,
202    pub sequence: u64,
203    pub time_exchange: Timestamp,
204    pub time_coinapi: Timestamp,
205    pub is_snapshot: bool,
206    pub asks: Vec<BookL3Entry>,
207    pub bids: Vec<BookL3Entry>,
208}
209
210/// An order entry; a missing identifier marks an aggregated L2 fallback entry.
211#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
212pub struct BookL3Entry {
213    pub id: Option<String>,
214    #[serde(deserialize_with = "decimal::deserialize_json")]
215    pub price: Decimal,
216    #[serde(deserialize_with = "decimal::deserialize_json")]
217    pub size: Decimal,
218    /// Absent in snapshots.
219    pub update_type: Option<BookUpdateType>,
220}
221
222/// How an L3 update changes an order.
223#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
224#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
225pub enum BookUpdateType {
226    Add,
227    Update,
228    Subtract,
229    Delete,
230    Match,
231    #[serde(other)]
232    Other,
233}
234
235/// A best bid and ask update; zero sizes mean the size is unknown.
236#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
237pub struct Quote {
238    pub symbol_id: Ustr,
239    pub sequence: u64,
240    pub time_exchange: Timestamp,
241    pub time_coinapi: Timestamp,
242    #[serde(deserialize_with = "decimal::deserialize_json")]
243    pub ask_price: Decimal,
244    #[serde(deserialize_with = "decimal::deserialize_json")]
245    pub ask_size: Decimal,
246    #[serde(deserialize_with = "decimal::deserialize_json")]
247    pub bid_price: Decimal,
248    #[serde(deserialize_with = "decimal::deserialize_json")]
249    pub bid_size: Decimal,
250}
251
252/// An executed trade.
253#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
254pub struct Trade {
255    pub symbol_id: Ustr,
256    pub sequence: u64,
257    pub time_exchange: Timestamp,
258    pub time_coinapi: Timestamp,
259    pub uuid: String,
260    #[serde(deserialize_with = "decimal::deserialize_json")]
261    pub price: Decimal,
262    #[serde(deserialize_with = "decimal::deserialize_json")]
263    pub size: Decimal,
264    pub taker_side: Ustr,
265}
266
267/// An OHLCV period update, sent on close or while the period is still open.
268#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
269pub struct Ohlcv {
270    pub symbol_id: Ustr,
271    pub sequence: u64,
272    pub period_id: Ustr,
273    pub time_period_start: Timestamp,
274    pub time_period_end: Timestamp,
275    pub time_open: Timestamp,
276    pub time_close: Timestamp,
277    #[serde(deserialize_with = "decimal::deserialize_json")]
278    pub price_open: Decimal,
279    #[serde(deserialize_with = "decimal::deserialize_json")]
280    pub price_high: Decimal,
281    #[serde(deserialize_with = "decimal::deserialize_json")]
282    pub price_low: Decimal,
283    #[serde(deserialize_with = "decimal::deserialize_json")]
284    pub price_close: Decimal,
285    #[serde(deserialize_with = "decimal::deserialize_json")]
286    pub volume_traded: Decimal,
287    pub trades_count: u64,
288}
289
290/// A server notice that the connection closes before `before_time`.
291#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
292pub struct Reconnect {
293    pub within_seconds: u64,
294    pub before_time: Timestamp,
295}
296
297/// A terminal venue error; the server closes the connection after sending it.
298#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
299pub struct VenueError {
300    pub message: String,
301}
302
303/// Messages emitted by [`CoinApiWebSocketClient`](super::client::CoinApiWebSocketClient).
304#[derive(Debug, Clone, PartialEq, Eq)]
305pub enum CoinApiWsMessage {
306    Book(Book),
307    BookL3(BookL3),
308    Quote(Quote),
309    Trade(Trade),
310    Ohlcv(Ohlcv),
311    /// The venue rejected the session; its connection and subscriptions are retired.
312    Error(String),
313    /// A replacement connection replayed the subscriptions.
314    ///
315    /// Sequence numbers restart per connection, so consumers must reset stream state such as books.
316    Reconnected,
317}
318
319impl CoinApiWsMessage {
320    /// Returns the channel and symbol a data message belongs to.
321    pub(super) fn subscription(&self) -> Option<Subscription> {
322        let (data_type, period_id, symbol_id) = match self {
323            Self::Book(book) => (book.data_type, None, book.symbol_id),
324            Self::BookL3(book) => (DataType::BookL3, None, book.symbol_id),
325            Self::Quote(quote) => (DataType::Quote, None, quote.symbol_id),
326            Self::Trade(trade) => (DataType::Trade, None, trade.symbol_id),
327            Self::Ohlcv(ohlcv) => (DataType::Ohlcv, Some(ohlcv.period_id), ohlcv.symbol_id),
328            Self::Error(_) | Self::Reconnected => return None,
329        };
330
331        Some(Subscription {
332            channel: Channel {
333                data_type,
334                period_id,
335            },
336            symbol_id,
337        })
338    }
339}
340
341/// One decoded inbound text frame.
342#[derive(Debug, Clone, PartialEq, Eq)]
343pub enum Frame {
344    Data(CoinApiWsMessage),
345    Heartbeat,
346    Reconnect(Reconnect),
347    Error(VenueError),
348    Unknown(String),
349}
350
351#[derive(Deserialize)]
352struct Envelope<'a> {
353    #[serde(borrow)]
354    r#type: &'a str,
355}
356
357impl Frame {
358    /// Decodes a text frame by its `type` field.
359    ///
360    /// Reads the tag in a first pass and then decodes the matching struct from the original bytes:
361    /// Serde tagged enums buffer numbers and lose raw decimal tokens, and CoinAPI sends `type` last.
362    ///
363    /// # Errors
364    ///
365    /// Returns an error if the frame is not JSON, has no string `type`, or a known type's payload
366    /// does not decode.
367    pub fn decode(bytes: &[u8]) -> Result<Self, serde_json::Error> {
368        let envelope: Envelope = serde_json::from_slice(bytes)?;
369        Ok(match envelope.r#type {
370            "book" | "book5" | "book20" | "book50" => {
371                Self::Data(CoinApiWsMessage::Book(serde_json::from_slice(bytes)?))
372            }
373            "book_l3" => Self::Data(CoinApiWsMessage::BookL3(serde_json::from_slice(bytes)?)),
374            "quote" => Self::Data(CoinApiWsMessage::Quote(serde_json::from_slice(bytes)?)),
375            "trade" => Self::Data(CoinApiWsMessage::Trade(serde_json::from_slice(bytes)?)),
376            "ohlcv" => Self::Data(CoinApiWsMessage::Ohlcv(serde_json::from_slice(bytes)?)),
377
378            // CoinAPI spells the heartbeat type this way on the wire
379            "hearbeat" => Self::Heartbeat,
380            "reconnect" => Self::Reconnect(serde_json::from_slice(bytes)?),
381            "error" => Self::Error(serde_json::from_slice(bytes)?),
382            other => Self::Unknown(other.to_owned()),
383        })
384    }
385}
386
387#[cfg(test)]
388mod tests {
389    use rstest::rstest;
390
391    use super::*;
392
393    fn decimal(text: &str) -> Decimal {
394        Decimal::from_str_exact(text).unwrap()
395    }
396
397    fn time(text: &str) -> Timestamp {
398        text.parse().unwrap()
399    }
400
401    fn decode(bytes: &str) -> Frame {
402        Frame::decode(bytes.as_bytes()).unwrap()
403    }
404
405    fn level(price: &str, size: &str) -> BookLevel {
406        BookLevel {
407            id: None,
408            price: decimal(price),
409            size: decimal(size),
410        }
411    }
412
413    #[rstest]
414    fn full_book_carries_snapshot_flag() {
415        assert_eq!(
416            decode(include_str!("../../test_data/ws_book.json")),
417            Frame::Data(CoinApiWsMessage::Book(Book {
418                data_type: DataType::Book,
419                symbol_id: "BITSTAMP_SPOT_BTC_USD".into(),
420                sequence: 2_323_346,
421                time_exchange: time("2013-09-28T22:40:50.0000000Z"),
422                time_coinapi: time("2017-03-18T22:42:21.3763342Z"),
423                is_snapshot: Some(true),
424                asks: vec![level("456.35", "123"), level("456.36", "23")],
425                bids: vec![level("456.10", "42"), level("456.09", "5")],
426            }))
427        );
428    }
429
430    #[rstest]
431    fn fixed_depth_book_keeps_its_type_without_snapshot_flag() {
432        assert_eq!(
433            decode(include_str!("../../test_data/ws_book20.json")),
434            Frame::Data(CoinApiWsMessage::Book(Book {
435                data_type: DataType::Book20,
436                symbol_id: "BITSTAMP_SPOT_BTC_USD".into(),
437                sequence: 2_323_346,
438                time_exchange: time("2013-09-28T22:40:50.0000000Z"),
439                time_coinapi: time("2017-03-18T22:42:21.3763342Z"),
440                is_snapshot: None,
441                asks: vec![level("456.35", "123")],
442                bids: vec![level("456.10", "42")],
443            }))
444        );
445    }
446
447    #[rstest]
448    fn book_l3_preserves_order_identity_and_update_types() {
449        let entry = |id: Option<&str>, price: &str, size: &str, update_type| BookL3Entry {
450            id: id.map(Into::into),
451            price: decimal(price),
452            size: decimal(size),
453            update_type: Some(update_type),
454        };
455
456        assert_eq!(
457            decode(include_str!("../../test_data/ws_book_l3.json")),
458            Frame::Data(CoinApiWsMessage::BookL3(BookL3 {
459                symbol_id: "COINBASE_SPOT_BCH_USD".into(),
460                sequence: 2_323_346,
461                time_exchange: time("2020-08-26T12:36:04.3464706Z"),
462                time_coinapi: time("2020-08-26T12:36:04.2807750Z"),
463                is_snapshot: false,
464                asks: vec![
465                    entry(
466                        Some("520269d4-c085-47ed-8825-6d787af90800"),
467                        "273.01",
468                        "1.81940502",
469                        BookUpdateType::Add
470                    ),
471                    entry(
472                        Some("0ac891c7-8360-462c-8d9a-d8b217d3fca2"),
473                        "273.02",
474                        "1.0",
475                        BookUpdateType::Delete
476                    ),
477                ],
478                bids: vec![
479                    entry(
480                        Some("6340ec1e-8be8-404e-b9a4-a14120ec179e"),
481                        "272.94",
482                        "1.81940502",
483                        BookUpdateType::Update
484                    ),
485                    entry(None, "272.93", "4.5", BookUpdateType::Match),
486                ],
487            }))
488        );
489    }
490
491    #[rstest]
492    #[case("SUBTRACT", BookUpdateType::Subtract)]
493    #[case("REPLACE", BookUpdateType::Other)]
494    fn book_update_type_classifies_open_values(
495        #[case] wire: &str,
496        #[case] expected: BookUpdateType,
497    ) {
498        assert_eq!(
499            serde_json::from_str::<BookUpdateType>(&format!("\"{wire}\"")).unwrap(),
500            expected
501        );
502    }
503
504    #[rstest]
505    fn quote_preserves_every_field_and_decimal_scale() {
506        let Frame::Data(CoinApiWsMessage::Quote(quote)) =
507            decode(include_str!("../../test_data/ws_quote.json"))
508        else {
509            panic!("expected quote");
510        };
511
512        assert_eq!(
513            quote,
514            Quote {
515                symbol_id: "BITSTAMP_SPOT_BTC_USD".into(),
516                sequence: 2_323_346,
517                time_exchange: time("2013-09-28T22:40:50.0000000Z"),
518                time_coinapi: time("2017-03-18T22:42:21.3763342Z"),
519                ask_price: decimal("770.000000000"),
520                ask_size: decimal("3252"),
521                bid_price: decimal("760"),
522                bid_size: decimal("124"),
523            }
524        );
525        assert_eq!(quote.ask_price.scale(), 9);
526    }
527
528    #[rstest]
529    fn trade_preserves_every_field_and_decimal_scale() {
530        let Frame::Data(CoinApiWsMessage::Trade(trade)) =
531            decode(include_str!("../../test_data/ws_trade.json"))
532        else {
533            panic!("expected trade");
534        };
535
536        assert_eq!(
537            trade,
538            Trade {
539                symbol_id: "COINBASE_SPOT_BTC_USD".into(),
540                sequence: 1507,
541                time_exchange: time("2026-09-25T01:29:06.0656180Z"),
542                time_coinapi: time("2026-09-25T01:29:06.1438272Z"),
543                uuid: "c2519dfb-79a6-40e9-a1c8-a95d94209399".into(),
544                price: decimal("84520.02"),
545                size: decimal("0.00000004"),
546                taker_side: "SELL".into(),
547            }
548        );
549        assert_eq!((trade.price.scale(), trade.size.scale()), (2, 8));
550    }
551
552    #[rstest]
553    fn ohlcv_preserves_period_and_every_field() {
554        assert_eq!(
555            decode(include_str!("../../test_data/ws_ohlcv.json")),
556            Frame::Data(CoinApiWsMessage::Ohlcv(Ohlcv {
557                symbol_id: "COINBASE_SPOT_BTC_USD".into(),
558                sequence: 520_348,
559                period_id: "1SEC".into(),
560                time_period_start: time("2026-09-25T01:30:01.0000000Z"),
561                time_period_end: time("2026-09-25T01:30:02.0000000Z"),
562                time_open: time("2026-09-25T01:30:01.0173140Z"),
563                time_close: time("2026-09-25T01:30:01.7668490Z"),
564                price_open: decimal("84523.69"),
565                price_high: decimal("84523.69"),
566                price_low: decimal("84523.68"),
567                price_close: decimal("84523.69"),
568                volume_traded: decimal("0.00468951"),
569                trades_count: 5,
570            }))
571        );
572    }
573
574    #[rstest]
575    #[case(include_str!("../../test_data/ws_heartbeat.json"), Frame::Heartbeat)]
576    #[case(
577        include_str!("../../test_data/ws_reconnect.json"),
578        Frame::Reconnect(Reconnect {
579            within_seconds: 10,
580            before_time: "2020-08-06T19:19:09.7035429Z".parse().unwrap(),
581        })
582    )]
583    #[case(
584        include_str!("../../test_data/ws_error.json"),
585        Frame::Error(VenueError {
586            message: "Subscribed data type 'nonsense' is invalid. Allowed values: book5, book20, \
587                      book50, book, book_l3, quote, trade, ohlcv, exrate, asset, exchange, \
588                      symbol, connectivity_status."
589                .into(),
590        })
591    )]
592    #[case(r#"{"type":"heartbeat"}"#, Frame::Unknown("heartbeat".into()))]
593    #[case(r#"{"asset_id":"BTC","type":"asset"}"#, Frame::Unknown("asset".into()))]
594    fn control_frames_decode_by_type(#[case] bytes: &str, #[case] expected: Frame) {
595        assert_eq!(decode(bytes), expected);
596    }
597
598    #[rstest]
599    #[case(include_str!("../../test_data/malformed.txt"))]
600    #[case(r#"{"symbol_id":"TEST"}"#)]
601    #[case(r#"{"type":"trade","symbol_id":"TEST"}"#)]
602    #[case(r#"{"type":"error"}"#)]
603    fn malformed_frames_fail_decoding(#[case] bytes: &str) {
604        assert!(Frame::decode(bytes.as_bytes()).is_err());
605    }
606
607    fn sub(data_type: DataType, period_id: Option<&str>, symbol_id: &str) -> Subscription {
608        Subscription {
609            channel: Channel {
610                data_type,
611                period_id: period_id.map(Ustr::from),
612            },
613            symbol_id: symbol_id.into(),
614        }
615    }
616
617    #[rstest]
618    #[case(
619        vec![sub(DataType::Trade, None, "B"), sub(DataType::Trade, None, "A")],
620        r#"{"type":"hello","heartbeat":true,"subscribe_data_type":["trade"],"subscribe_filter_symbol_id":["A$","B$"]}"#
621    )]
622    #[case(
623        vec![sub(DataType::Trade, None, "A"), sub(DataType::BookL3, None, "B"), sub(DataType::Book20, None, "A")],
624        r#"{"type":"hello","heartbeat":true,"subscribe_data_type":["book20","book_l3","trade"],"subscribe_filter_symbol_id":["A$","B$"]}"#
625    )]
626    #[case(
627        vec![sub(DataType::Ohlcv, Some("1SEC"), "A"), sub(DataType::Ohlcv, Some("1MIN"), "A"), sub(DataType::Quote, None, "A")],
628        r#"{"type":"hello","heartbeat":true,"subscribe_data_type":["ohlcv","quote"],"subscribe_filter_symbol_id":["A$"],"subscribe_filter_period_id":["1MIN","1SEC"]}"#
629    )]
630    fn hello_serializes_sorted_union_scope(
631        #[case] subscriptions: Vec<Subscription>,
632        #[case] expected: &str,
633    ) {
634        assert_eq!(
635            serde_json::to_string(&Hello::new(&subscriptions)).unwrap(),
636            expected
637        );
638    }
639
640    #[rstest]
641    #[case(DataType::Book, "book")]
642    #[case(DataType::Book5, "book5")]
643    #[case(DataType::Book20, "book20")]
644    #[case(DataType::Book50, "book50")]
645    #[case(DataType::BookL3, "book_l3")]
646    #[case(DataType::Quote, "quote")]
647    #[case(DataType::Trade, "trade")]
648    #[case(DataType::Ohlcv, "ohlcv")]
649    fn data_type_topic_name_matches_wire_name(#[case] data_type: DataType, #[case] wire: &str) {
650        assert_eq!(data_type.is_book(), wire.starts_with("book"));
651        assert_eq!(data_type.as_ref(), wire);
652        assert_eq!(wire.parse::<DataType>().unwrap(), data_type);
653        assert_eq!(
654            serde_json::to_string(&data_type).unwrap(),
655            format!("\"{wire}\"")
656        );
657    }
658
659    #[rstest]
660    #[case(DataType::Trade, None, "COINBASE_SPOT_BTC_USD", true)]
661    #[case(DataType::Ohlcv, Some("1MIN"), "COINBASE_SPOT_BTC_USD", true)]
662    #[case(DataType::Ohlcv, None, "COINBASE_SPOT_BTC_USD", false)]
663    #[case(DataType::Ohlcv, Some(""), "COINBASE_SPOT_BTC_USD", false)]
664    #[case(DataType::Ohlcv, Some("1:MIN"), "COINBASE_SPOT_BTC_USD", false)]
665    #[case(DataType::Trade, Some("1MIN"), "COINBASE_SPOT_BTC_USD", false)]
666    #[case(DataType::Trade, None, "", false)]
667    #[case(DataType::Trade, None, "COINBASE_SPOT_BTC_USD$", false)]
668    #[case(DataType::Trade, None, "COINBASE SPOT", false)]
669    fn subscription_validation(
670        #[case] data_type: DataType,
671        #[case] period_id: Option<&str>,
672        #[case] symbol_id: &str,
673        #[case] valid: bool,
674    ) {
675        let subscription = Subscription {
676            channel: Channel {
677                data_type,
678                period_id: period_id.map(Ustr::from),
679            },
680            symbol_id: symbol_id.into(),
681        };
682
683        assert_eq!(subscription.validate().is_ok(), valid);
684    }
685}
nautilus_coinapi/websocket/messages.rs

nautilus_coinapi/websocket/
messages.rs
