nautilus_coinapi/websocket/
parse.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//! Conversion of CoinAPI WebSocket messages into Nautilus market data.
17//!
18//! Instrument precisions come from CoinAPI data resolutions, which can lag the stream. A value
19//! with more decimal places than its instrument precision is rejected, except for bounded
20//! numeric tails in OHLCV volume.
21
22use jiff::Timestamp;
23use nautilus_core::UnixNanos;
24use nautilus_model::{
25    data::{
26        Bar, BarSpecification, BarType, BookOrder, OrderBookDelta, OrderBookDeltas, OrderBookDepth,
27        QuoteTick, TradeTick,
28    },
29    enums::{
30        AggregationSource, AggressorSide, BarAggregation, BookAction, OrderSide, PriceType,
31        RecordFlag,
32    },
33    identifiers::TradeId,
34    instruments::{Instrument, InstrumentAny},
35    types::{Price, Quantity},
36};
37use rust_decimal::{Decimal, RoundingStrategy};
38use ustr::Ustr;
39
40use super::messages::{Book, Ohlcv, Quote, Trade};
41use crate::http::{models::BookLevel, parse::unix_nanos};
42
43/// Converts a full L2 `book` snapshot or update into deltas, taking `ts_event` from
44/// `time_exchange` and `ts_init` from `time_coinapi`.
45///
46/// A snapshot becomes a `Clear` followed by an `Add` per level, every delta flagged `F_SNAPSHOT`;
47/// an empty snapshot is a lone `Clear`. An update sets each listed level to its new size, and a
48/// zero size removes the level. The last delta carries `F_LAST`, and an update listing no levels
49/// returns `None`.
50///
51/// # Errors
52///
53/// Returns an error if a level is finer than the instrument precision, a snapshot level has zero
54/// size, or a timestamp is invalid; a single invalid level rejects the whole message.
55pub fn parse_book_deltas(
56    book: &Book,
57    instrument: &InstrumentAny,
58) -> anyhow::Result<Option<OrderBookDeltas>> {
59    let instrument_id = instrument.id();
60    let ts_event = timestamp(book.time_exchange, "time_exchange")?;
61    let ts_init = timestamp(book.time_coinapi, "time_coinapi")?;
62    let is_snapshot = book.is_snapshot == Some(true);
63    let mut deltas = Vec::with_capacity(book.bids.len() + book.asks.len() + 1);
64    let flags = if is_snapshot {
65        deltas.push(OrderBookDelta::clear(
66            instrument_id,
67            book.sequence,
68            ts_event,
69            ts_init,
70        ));
71        RecordFlag::F_SNAPSHOT as u8
72    } else {
73        0
74    };
75
76    for (side, level) in levels(book) {
77        let order = book_order(side, level, instrument)?;
78
79        let action = if is_snapshot {
80            BookAction::Add
81        } else if order.size.is_zero() {
82            BookAction::Delete
83        } else {
84            BookAction::Update
85        };
86
87        deltas.push(OrderBookDelta::new_checked(
88            instrument_id,
89            action,
90            order,
91            flags,
92            book.sequence,
93            ts_event,
94            ts_init,
95        )?);
96    }
97
98    let Some(last) = deltas.last_mut() else {
99        return Ok(None);
100    };
101
102    last.flags |= RecordFlag::F_LAST as u8;
103
104    Ok(Some(OrderBookDeltas::new(instrument_id, deltas)))
105}
106
107/// Converts a fixed-depth `book5`, `book20`, or `book50` message into its best `depth` levels
108/// per side, taking `ts_event` from `time_exchange` and `ts_init` from `time_coinapi`.
109///
110/// CoinAPI reports no order counts, so every level count is zero.
111///
112/// # Errors
113///
114/// Returns an error if a kept level is finer than the instrument precision or a timestamp is
115/// invalid.
116pub fn parse_book_depth(
117    book: &Book,
118    depth: usize,
119    instrument: &InstrumentAny,
120) -> anyhow::Result<OrderBookDepth> {
121    let side = |side: OrderSide, levels: &[BookLevel]| {
122        levels
123            .iter()
124            .take(depth)
125            .map(|level| book_order(side, level, instrument))
126            .collect::<anyhow::Result<Vec<BookOrder>>>()
127    };
128
129    let bids = side(OrderSide::Buy, &book.bids)?;
130    let asks = side(OrderSide::Sell, &book.asks)?;
131    let bid_counts = vec![0; bids.len()];
132    let ask_counts = vec![0; asks.len()];
133
134    OrderBookDepth::new_checked(
135        instrument.id(),
136        bids,
137        asks,
138        bid_counts,
139        ask_counts,
140        RecordFlag::F_SNAPSHOT as u8,
141        book.sequence,
142        timestamp(book.time_exchange, "time_exchange")?,
143        timestamp(book.time_coinapi, "time_coinapi")?,
144    )
145}
146
147/// Converts a best bid and ask, taking `ts_event` from `time_exchange` and `ts_init` from
148/// `time_coinapi`.
149///
150/// # Errors
151///
152/// Returns an error if a value is finer than the instrument precision or a timestamp is invalid.
153pub fn parse_quote_tick(quote: &Quote, instrument: &InstrumentAny) -> anyhow::Result<QuoteTick> {
154    QuoteTick::new_checked(
155        instrument.id(),
156        price(quote.bid_price, instrument, "bid_price")?,
157        price(quote.ask_price, instrument, "ask_price")?,
158        quantity(quote.bid_size, instrument, "bid_size")?,
159        quantity(quote.ask_size, instrument, "ask_size")?,
160        timestamp(quote.time_exchange, "time_exchange")?,
161        timestamp(quote.time_coinapi, "time_coinapi")?,
162    )
163}
164
165/// Converts a trade, taking `ts_event` from `time_exchange` and `ts_init` from `time_coinapi`.
166///
167/// Only exchange-reported taker sides set the aggressor; CoinAPI estimates and unknown sides
168/// become [`AggressorSide::NoAggressor`].
169///
170/// # Errors
171///
172/// Returns an error if a value is finer than the instrument precision, the size is not positive, or
173/// a timestamp or the trade identifier is invalid.
174pub fn parse_trade_tick(trade: &Trade, instrument: &InstrumentAny) -> anyhow::Result<TradeTick> {
175    TradeTick::new_checked(
176        instrument.id(),
177        price(trade.price, instrument, "price")?,
178        quantity(trade.size, instrument, "size")?,
179        parse_aggressor_side(trade.taker_side),
180        TradeId::new_checked(&trade.uuid)?,
181        timestamp(trade.time_exchange, "time_exchange")?,
182        timestamp(trade.time_coinapi, "time_coinapi")?,
183    )
184}
185
186/// Converts a completed OHLCV period into an external last-price bar closing at
187/// `time_period_end`.
188///
189/// Rounds volume to the instrument's size precision when the adjustment is at most both
190/// `1e-15` times the reported volume and `1e-6` times one unit at that precision.
191///
192/// # Errors
193///
194/// Returns an error if the period is unsupported, a price is finer than the instrument precision,
195/// volume needs rounding beyond the noise tolerance, the prices are inconsistent, or a timestamp is
196/// invalid.
197pub fn parse_bar(
198    ohlcv: &Ohlcv,
199    instrument: &InstrumentAny,
200    ts_init: UnixNanos,
201) -> anyhow::Result<Bar> {
202    let spec = period_id_to_bar_spec(ohlcv.period_id)
203        .ok_or_else(|| anyhow::anyhow!("unsupported period_id {}", ohlcv.period_id))?;
204
205    Bar::new_checked(
206        BarType::new(instrument.id(), spec, AggregationSource::External),
207        price(ohlcv.price_open, instrument, "price_open")?,
208        price(ohlcv.price_high, instrument, "price_high")?,
209        price(ohlcv.price_low, instrument, "price_low")?,
210        price(ohlcv.price_close, instrument, "price_close")?,
211        volume(ohlcv.volume_traded, instrument)?,
212        timestamp(ohlcv.time_period_end, "time_period_end")?,
213        ts_init,
214    )
215}
216
217/// Maps an exchange taker side to the aggressor side.
218#[must_use]
219pub fn parse_aggressor_side(taker_side: Ustr) -> AggressorSide {
220    match taker_side.as_str() {
221        "BUY" => AggressorSide::Buy,
222        "SELL" => AggressorSide::Sell,
223        "BUY_ESTIMATED" | "SELL_ESTIMATED" | "UNKNOWN" => AggressorSide::NoAggressor,
224        other => {
225            log::warn!("Unknown CoinAPI taker_side {other}, using no aggressor");
226            AggressorSide::NoAggressor
227        }
228    }
229}
230
231/// Returns the CoinAPI streaming `period_id` for a last-price bar specification.
232///
233/// CoinAPI streams OHLCV periods from one second to one day.
234///
235/// # Errors
236///
237/// Returns an error if the price type is not last or CoinAPI has no matching period.
238pub fn bar_spec_to_period_id(spec: BarSpecification) -> anyhow::Result<Ustr> {
239    anyhow::ensure!(
240        spec.price_type == PriceType::Last,
241        "CoinAPI OHLCV bars use last prices, was {}",
242        spec.price_type
243    );
244
245    let step = spec.step.get();
246    let unit = match spec.aggregation {
247        BarAggregation::Second if SUB_HOUR_STEPS.contains(&step) => "SEC",
248        BarAggregation::Minute if SUB_HOUR_STEPS.contains(&step) => "MIN",
249        BarAggregation::Hour if HOUR_STEPS.contains(&step) => "HRS",
250        BarAggregation::Day if step == 1 => "DAY",
251        _ => anyhow::bail!("CoinAPI has no streaming OHLCV period for {spec}"),
252    };
253
254    Ok(Ustr::from(&format!("{step}{unit}")))
255}
256
257/// Returns the last-price bar specification for a CoinAPI streaming `period_id`.
258#[must_use]
259pub fn period_id_to_bar_spec(period_id: Ustr) -> Option<BarSpecification> {
260    let period = period_id.as_str();
261    let split = period.find(|c: char| !c.is_ascii_digit())?;
262    let (step, unit) = period.split_at(split);
263
264    let aggregation = match unit {
265        "SEC" => BarAggregation::Second,
266        "MIN" => BarAggregation::Minute,
267        "HRS" => BarAggregation::Hour,
268        "DAY" => BarAggregation::Day,
269        _ => return None,
270    };
271
272    let spec =
273        BarSpecification::new_checked(step.parse().ok()?, aggregation, PriceType::Last).ok()?;
274
275    // Rejects leading zeros and steps CoinAPI does not stream
276    (bar_spec_to_period_id(spec).ok()? == period_id).then_some(spec)
277}
278
279fn levels(book: &Book) -> impl Iterator<Item = (OrderSide, &BookLevel)> {
280    let bids = book.bids.iter().map(|level| (OrderSide::Buy, level));
281    let asks = book.asks.iter().map(|level| (OrderSide::Sell, level));
282    bids.chain(asks)
283}
284
285fn book_order(
286    side: OrderSide,
287    level: &BookLevel,
288    instrument: &InstrumentAny,
289) -> anyhow::Result<BookOrder> {
290    Ok(BookOrder::new(
291        side,
292        price(level.price, instrument, "price")?,
293        quantity(level.size, instrument, "size")?,
294        0,
295    ))
296}
297
298const SUB_HOUR_STEPS: [usize; 10] = [1, 2, 3, 4, 5, 6, 10, 15, 20, 30];
299const HOUR_STEPS: [usize; 7] = [1, 2, 3, 4, 6, 8, 12];
300
301pub(crate) fn price(
302    value: Decimal,
303    instrument: &InstrumentAny,
304    field: &str,
305) -> anyhow::Result<Price> {
306    price_at(value, instrument.price_precision(), field)
307}
308
309pub(crate) fn quantity(
310    value: Decimal,
311    instrument: &InstrumentAny,
312    field: &str,
313) -> anyhow::Result<Quantity> {
314    quantity_at(value, instrument.size_precision(), field)
315}
316
317pub(crate) fn volume(value: Decimal, instrument: &InstrumentAny) -> anyhow::Result<Quantity> {
318    let precision = u32::from(instrument.size_precision());
319    let rounded = value.round_dp_with_strategy(precision, RoundingStrategy::MidpointNearestEven);
320    let adjustment = (value - rounded).abs();
321
322    // The size-unit cap rules out meaningful rounding even for coarse resolutions. Check it
323    // before scaling the adjustment, which keeps the relative comparison exact and bounded.
324    let normalized = if value >= Decimal::ZERO
325        && adjustment <= Decimal::new(1, precision + 6)
326        && adjustment * Decimal::new(1_000_000_000_000_000, 0) <= value
327    {
328        rounded
329    } else {
330        value
331    };
332
333    quantity_at(normalized, instrument.size_precision(), "volume_traded")
334}
335
336/// Converts `value` at `precision`, rejecting a value that would need rounding.
337pub(crate) fn price_at(value: Decimal, precision: u8, field: &str) -> anyhow::Result<Price> {
338    check_scale(value, precision, field)?;
339    Ok(Price::from_decimal_dp(value, precision)?)
340}
341
342/// Converts `value` at `precision`, rejecting a value that would need rounding.
343pub(crate) fn quantity_at(value: Decimal, precision: u8, field: &str) -> anyhow::Result<Quantity> {
344    check_scale(value, precision, field)?;
345    Ok(Quantity::from_decimal_dp(value, precision)?)
346}
347
348fn check_scale(value: Decimal, precision: u8, field: &str) -> anyhow::Result<()> {
349    anyhow::ensure!(
350        value.normalize().scale() <= u32::from(precision),
351        "{field} {value} is finer than instrument precision {precision}"
352    );
353    Ok(())
354}
355
356pub(crate) fn timestamp(value: Timestamp, field: &str) -> anyhow::Result<UnixNanos> {
357    unix_nanos(value).map_err(|e| anyhow::anyhow!("invalid {field}: {e}"))
358}
359
360#[cfg(test)]
361mod tests {
362    use nautilus_model::{identifiers::InstrumentId, instruments::CurrencyPair, types::Currency};
363    use rstest::rstest;
364
365    use super::*;
366
367    fn instrument(price_precision: u8, size_precision: u8) -> InstrumentAny {
368        InstrumentAny::CurrencyPair(
369            CurrencyPair::builder()
370                .instrument_id(InstrumentId::from("BTC-USD.COINBASE"))
371                .raw_symbol("BTC-USD".into())
372                .base_currency(Currency::BTC())
373                .quote_currency(Currency::USD())
374                .price_precision(price_precision)
375                .size_precision(size_precision)
376                .price_increment(Price::new(
377                    10f64.powi(-i32::from(price_precision)),
378                    price_precision,
379                ))
380                .size_increment(Quantity::new(
381                    10f64.powi(-i32::from(size_precision)),
382                    size_precision,
383                ))
384                .ts_event(UnixNanos::default())
385                .ts_init(UnixNanos::default())
386                .build()
387                .unwrap(),
388        )
389    }
390
391    fn quote() -> Quote {
392        serde_json::from_str(include_str!("../../test_data/ws_quote.json")).unwrap()
393    }
394
395    fn trade() -> Trade {
396        serde_json::from_str(include_str!("../../test_data/ws_trade.json")).unwrap()
397    }
398
399    fn bar_spec(spec: &str) -> BarSpecification {
400        BarType::from(format!("BTC-USD.COINBASE-{spec}-EXTERNAL").as_str()).spec()
401    }
402
403    fn ohlcv() -> Ohlcv {
404        serde_json::from_str(include_str!("../../test_data/ws_ohlcv.json")).unwrap()
405    }
406
407    #[rstest]
408    fn quote_maps_every_field_and_ignores_trailing_zero_padding() {
409        let tick = parse_quote_tick(&quote(), &instrument(2, 0)).unwrap();
410
411        assert_eq!(
412            tick,
413            QuoteTick::new(
414                InstrumentId::from("BTC-USD.COINBASE"),
415                Price::from("760.00"),
416                Price::from("770.00"),
417                Quantity::from("124"),
418                Quantity::from("3252"),
419                UnixNanos::from(1_380_408_050_000_000_000),
420                UnixNanos::from(1_489_876_941_376_334_200),
421            )
422        );
423    }
424
425    #[rstest]
426    fn trade_maps_every_field_and_both_timestamps() {
427        let tick = parse_trade_tick(&trade(), &instrument(2, 8)).unwrap();
428
429        assert_eq!(
430            tick,
431            TradeTick::new(
432                InstrumentId::from("BTC-USD.COINBASE"),
433                Price::from("84520.02"),
434                Quantity::from("0.00000004"),
435                AggressorSide::Sell,
436                TradeId::new("c2519dfb-79a6-40e9-a1c8-a95d94209399"),
437                UnixNanos::from(1_790_299_746_065_618_000),
438                UnixNanos::from(1_790_299_746_143_827_200),
439            )
440        );
441    }
442
443    #[rstest]
444    fn bar_closes_at_period_end_with_given_ts_init() {
445        let ts_init = UnixNanos::from(1_790_299_803_000_000_000);
446        let bar = parse_bar(&ohlcv(), &instrument(2, 8), ts_init).unwrap();
447
448        assert_eq!(
449            bar,
450            Bar::new(
451                BarType::from("BTC-USD.COINBASE-1-SECOND-LAST-EXTERNAL"),
452                Price::from("84523.69"),
453                Price::from("84523.69"),
454                Price::from("84523.68"),
455                Price::from("84523.69"),
456                Quantity::from("0.00468951"),
457                UnixNanos::from(1_790_299_802_000_000_000),
458                ts_init,
459            )
460        );
461    }
462
463    #[rstest]
464    #[case("7.087936829999999", 8, "7.08793683")]
465    #[case("1.2582318399999999", 8, "1.25823184")]
466    #[case("28119.500000000004", 1, "28119.5")]
467    #[case("0.9707556800000001", 8, "0.97075568")]
468    #[case("1.000000000000001", 8, "1.00000000")]
469    #[case("100.00000000000001", 8, "100.00000000")]
470    #[case("0.000000001", 9, "0.000000001")]
471    #[case("0", 8, "0.00000000")]
472    fn bar_normalizes_volume_noise(
473        #[case] volume: &str,
474        #[case] size_precision: u8,
475        #[case] expected_volume: &str,
476    ) {
477        let mut ohlcv = ohlcv();
478        ohlcv.volume_traded = Decimal::from_str_exact(volume).unwrap();
479        let ts_init = UnixNanos::from(1_790_299_803_000_000_000);
480
481        let bar = parse_bar(&ohlcv, &instrument(2, size_precision), ts_init).unwrap();
482
483        assert_eq!(
484            bar,
485            Bar::new(
486                BarType::from("BTC-USD.COINBASE-1-SECOND-LAST-EXTERNAL"),
487                Price::from("84523.69"),
488                Price::from("84523.69"),
489                Price::from("84523.68"),
490                Price::from("84523.69"),
491                Quantity::from(expected_volume),
492                UnixNanos::from(1_790_299_802_000_000_000),
493                ts_init,
494            )
495        );
496        assert_eq!(bar.volume.precision, size_precision);
497    }
498
499    #[rstest]
500    #[case("1.000000000000002")]
501    #[case("100.00000000000002")]
502    #[case("1000000000.000000001")]
503    #[case("0.00000000000000000001")]
504    #[case("-0.0000000000000001")]
505    #[case("0.004689515")]
506    #[case("0.999999999999999")]
507    fn bar_rejects_volume_outside_noise_tolerance(#[case] volume: &str) {
508        let mut ohlcv = ohlcv();
509        ohlcv.volume_traded = Decimal::from_str_exact(volume).unwrap();
510
511        let error = parse_bar(&ohlcv, &instrument(2, 8), UnixNanos::default()).unwrap_err();
512
513        assert_eq!(
514            error.to_string(),
515            format!("volume_traded {volume} is finer than instrument precision 8")
516        );
517    }
518
519    #[cfg(feature = "high-precision")]
520    #[rstest]
521    #[case("0.0000000000000600000000000001", None)]
522    #[case("0.0000000000000999999999999999", None)]
523    #[case("0.0000000000001000000000000001", Some("0.0000000000001000"))]
524    fn volume_relative_bound_at_high_precision(
525        #[case] source: &str,
526        #[case] expected: Option<&str>,
527    ) {
528        let result = volume(Decimal::from_str_exact(source).unwrap(), &instrument(2, 16));
529        if let Some(expected) = expected {
530            let quantity = result.unwrap();
531            assert_eq!(quantity, Quantity::from(expected));
532            assert_eq!(quantity.precision, 16);
533        } else {
534            assert_eq!(
535                result.unwrap_err().to_string(),
536                format!("volume_traded {source} is finer than instrument precision 16")
537            );
538        }
539    }
540
541    #[rstest]
542    fn quote_finer_than_instrument_precision_is_rejected() {
543        let mut quote = quote();
544        quote.bid_price = Decimal::from_str_exact("760.005").unwrap();
545
546        let error = parse_quote_tick(&quote, &instrument(2, 0)).unwrap_err();
547
548        assert_eq!(
549            error.to_string(),
550            "bid_price 760.005 is finer than instrument precision 2"
551        );
552    }
553
554    #[rstest]
555    #[case(1, 8, "price 84520.02 is finer than instrument precision 1")]
556    #[case(2, 7, "size 0.00000004 is finer than instrument precision 7")]
557    fn trade_finer_than_instrument_precision_is_rejected(
558        #[case] price_precision: u8,
559        #[case] size_precision: u8,
560        #[case] expected: &str,
561    ) {
562        let error =
563            parse_trade_tick(&trade(), &instrument(price_precision, size_precision)).unwrap_err();
564
565        assert_eq!(error.to_string(), expected);
566    }
567
568    #[rstest]
569    fn bar_finer_than_instrument_precision_is_rejected() {
570        let error = parse_bar(&ohlcv(), &instrument(2, 7), UnixNanos::default()).unwrap_err();
571
572        assert_eq!(
573            error.to_string(),
574            "volume_traded 0.00468951 is finer than instrument precision 7"
575        );
576    }
577
578    fn book(is_snapshot: Option<bool>, bids: &[(&str, &str)], asks: &[(&str, &str)]) -> Book {
579        let level = |(price, size): &(&str, &str)| BookLevel {
580            id: None,
581            price: Decimal::from_str_exact(price).unwrap(),
582            size: Decimal::from_str_exact(size).unwrap(),
583        };
584
585        let mut book: Book =
586            serde_json::from_str(include_str!("../../test_data/ws_book.json")).unwrap();
587        book.is_snapshot = is_snapshot;
588        book.bids = bids.iter().map(level).collect();
589        book.asks = asks.iter().map(level).collect();
590        book
591    }
592
593    fn delta(
594        action: BookAction,
595        side: OrderSide,
596        price: &str,
597        size: &str,
598        flags: u8,
599    ) -> OrderBookDelta {
600        OrderBookDelta::new(
601            InstrumentId::from("BTC-USD.COINBASE"),
602            action,
603            BookOrder::new(side, Price::from(price), Quantity::from(size), 0),
604            flags,
605            2_323_346,
606            UnixNanos::from(1_380_408_050_000_000_000),
607            UnixNanos::from(1_489_876_941_376_334_200),
608        )
609    }
610
611    const SNAPSHOT: u8 = RecordFlag::F_SNAPSHOT as u8;
612    const LAST: u8 = RecordFlag::F_LAST as u8;
613
614    #[rstest]
615    fn book_snapshot_clears_then_adds_every_level() {
616        let deltas = parse_book_deltas(
617            &book(
618                Some(true),
619                &[("456.10", "42")],
620                &[("456.35", "123"), ("456.36", "23")],
621            ),
622            &instrument(2, 0),
623        )
624        .unwrap()
625        .unwrap();
626
627        let mut clear = OrderBookDelta::clear(
628            InstrumentId::from("BTC-USD.COINBASE"),
629            2_323_346,
630            UnixNanos::from(1_380_408_050_000_000_000),
631            UnixNanos::from(1_489_876_941_376_334_200),
632        );
633        clear.flags = SNAPSHOT;
634
635        assert_eq!(
636            deltas.deltas,
637            vec![
638                clear,
639                delta(BookAction::Add, OrderSide::Buy, "456.10", "42", SNAPSHOT),
640                delta(BookAction::Add, OrderSide::Sell, "456.35", "123", SNAPSHOT),
641                delta(
642                    BookAction::Add,
643                    OrderSide::Sell,
644                    "456.36",
645                    "23",
646                    SNAPSHOT | LAST
647                ),
648            ]
649        );
650    }
651
652    #[rstest]
653    fn empty_book_snapshot_is_a_lone_clear() {
654        let deltas = parse_book_deltas(&book(Some(true), &[], &[]), &instrument(2, 0))
655            .unwrap()
656            .unwrap();
657
658        assert_eq!(deltas.deltas.len(), 1);
659        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
660        assert_eq!(deltas.deltas[0].flags, SNAPSHOT | LAST);
661    }
662
663    #[rstest]
664    fn book_update_sets_levels_and_removes_zero_sizes() {
665        let deltas = parse_book_deltas(
666            &book(Some(false), &[("456.10", "0")], &[("456.35", "7")]),
667            &instrument(2, 0),
668        )
669        .unwrap()
670        .unwrap();
671
672        assert_eq!(
673            deltas.deltas,
674            vec![
675                delta(BookAction::Delete, OrderSide::Buy, "456.10", "0", 0),
676                delta(BookAction::Update, OrderSide::Sell, "456.35", "7", LAST),
677            ]
678        );
679    }
680
681    #[rstest]
682    fn book_update_without_levels_emits_nothing() {
683        let deltas = parse_book_deltas(&book(Some(false), &[], &[]), &instrument(2, 0)).unwrap();
684
685        assert_eq!(deltas, None);
686    }
687
688    #[rstest]
689    fn book_update_rejects_whole_message_for_one_finer_level() {
690        let error = parse_book_deltas(
691            &book(Some(false), &[("456.10", "1")], &[("456.355", "1")]),
692            &instrument(2, 0),
693        )
694        .unwrap_err();
695
696        assert_eq!(
697            error.to_string(),
698            "price 456.355 is finer than instrument precision 2"
699        );
700    }
701
702    #[rstest]
703    fn book_depth_keeps_best_levels_per_side() {
704        let mut fixed = book(
705            None,
706            &[("456.10", "1"), ("456.09", "2"), ("456.08", "3")],
707            &[("456.35", "4"), ("456.36", "5")],
708        );
709        fixed.data_type = super::super::messages::DataType::Book5;
710
711        let depth = parse_book_depth(&fixed, 2, &instrument(2, 0)).unwrap();
712
713        assert_eq!(
714            depth,
715            OrderBookDepth::new(
716                InstrumentId::from("BTC-USD.COINBASE"),
717                vec![
718                    BookOrder::new(
719                        OrderSide::Buy,
720                        Price::from("456.10"),
721                        Quantity::from("1"),
722                        0
723                    ),
724                    BookOrder::new(
725                        OrderSide::Buy,
726                        Price::from("456.09"),
727                        Quantity::from("2"),
728                        0
729                    ),
730                ],
731                vec![
732                    BookOrder::new(
733                        OrderSide::Sell,
734                        Price::from("456.35"),
735                        Quantity::from("4"),
736                        0
737                    ),
738                    BookOrder::new(
739                        OrderSide::Sell,
740                        Price::from("456.36"),
741                        Quantity::from("5"),
742                        0
743                    ),
744                ],
745                vec![0, 0],
746                vec![0, 0],
747                SNAPSHOT,
748                2_323_346,
749                UnixNanos::from(1_380_408_050_000_000_000),
750                UnixNanos::from(1_489_876_941_376_334_200),
751            )
752        );
753    }
754
755    #[rstest]
756    #[case("BUY", AggressorSide::Buy)]
757    #[case("SELL", AggressorSide::Sell)]
758    #[case("BUY_ESTIMATED", AggressorSide::NoAggressor)]
759    #[case("SELL_ESTIMATED", AggressorSide::NoAggressor)]
760    #[case("UNKNOWN", AggressorSide::NoAggressor)]
761    #[case("NEW_SIDE", AggressorSide::NoAggressor)]
762    fn taker_side_maps_only_reported_sides(#[case] wire: &str, #[case] expected: AggressorSide) {
763        assert_eq!(parse_aggressor_side(Ustr::from(wire)), expected);
764    }
765
766    #[rstest]
767    #[case("1-SECOND-LAST", "1SEC")]
768    #[case("30-SECOND-LAST", "30SEC")]
769    #[case("1-MINUTE-LAST", "1MIN")]
770    #[case("15-MINUTE-LAST", "15MIN")]
771    #[case("1-HOUR-LAST", "1HRS")]
772    #[case("12-HOUR-LAST", "12HRS")]
773    #[case("1-DAY-LAST", "1DAY")]
774    fn bar_spec_maps_to_period_id_and_back(#[case] spec: &str, #[case] period_id: &str) {
775        let spec = bar_spec(spec);
776
777        assert_eq!(bar_spec_to_period_id(spec).unwrap(), period_id);
778        assert_eq!(period_id_to_bar_spec(Ustr::from(period_id)), Some(spec));
779    }
780
781    #[rstest]
782    #[case("12-SECOND-LAST")]
783    #[case("12-MINUTE-LAST")]
784    #[case("2-DAY-LAST")]
785    #[case("1-WEEK-LAST")]
786    #[case("1-MINUTE-MID")]
787    #[case("100-TICK-LAST")]
788    fn bar_spec_without_streaming_period_is_rejected(#[case] spec: &str) {
789        assert!(bar_spec_to_period_id(bar_spec(spec)).is_err());
790    }
791
792    #[rstest]
793    #[case("01MIN")]
794    #[case("7SEC")]
795    #[case("1MTH")]
796    #[case("MIN")]
797    #[case("1")]
798    #[case("")]
799    fn unsupported_period_id_has_no_bar_spec(#[case] period_id: &str) {
800        assert_eq!(period_id_to_bar_spec(Ustr::from(period_id)), None);
801    }
802}
nautilus_coinapi/websocket/parse.rs

nautilus_coinapi/websocket/
parse.rs
