1use super::super::book::BinanceLevel;
2use crate::{
3 Identifier, SnapshotFetcher,
4 books::{OrderBook, OrderBookTimes},
5 error::DataError,
6 event::{MarketEvent, MarketIter},
7 exchange::{
8 Connector,
9 binance::{
10 book::l2::{BinanceOrderBookL2Meta, BinanceOrderBookL2Snapshot},
11 futures::BinanceFuturesUsd,
12 market::BinanceMarket,
13 },
14 },
15 instrument::InstrumentData,
16 subscription::{
17 Map, Subscription,
18 book::{OrderBookEvent, OrderBooksL2},
19 },
20 transformer::ExchangeTransformer,
21};
22use chrono::{DateTime, Utc};
23use futures_util::future::try_join_all;
24use rustrade_instrument::exchange::ExchangeId;
25use rustrade_integration::{
26 Transformer, error::SocketError, protocol::websocket::WsMessage, subscription::SubscriptionId,
27};
28use serde::Deserialize;
29use std::future::Future;
30use tokio::sync::mpsc::UnboundedSender;
31
32pub const HTTP_BOOK_L2_SNAPSHOT_URL_BINANCE_FUTURES_USD: &str =
36 "https://fapi.binance.com/fapi/v1/depth";
37
38#[derive(Debug)]
39pub struct BinanceFuturesUsdOrderBooksL2SnapshotFetcher;
40
41impl SnapshotFetcher<BinanceFuturesUsd, OrderBooksL2>
42 for BinanceFuturesUsdOrderBooksL2SnapshotFetcher
43{
44 fn fetch_snapshots<Instrument>(
45 subscriptions: &[Subscription<BinanceFuturesUsd, Instrument, OrderBooksL2>],
46 ) -> impl Future<Output = Result<Vec<MarketEvent<Instrument::Key, OrderBookEvent>>, SocketError>>
47 + Send
48 where
49 Instrument: InstrumentData,
50 Subscription<BinanceFuturesUsd, Instrument, OrderBooksL2>: Identifier<BinanceMarket>,
51 {
52 let l2_snapshot_futures = subscriptions.iter().map(|sub| {
53 let market = sub.id();
55 let snapshot_url = format!(
56 "{}?symbol={}&limit=100",
57 HTTP_BOOK_L2_SNAPSHOT_URL_BINANCE_FUTURES_USD,
58 market.as_ref(),
59 );
60
61 async move {
62 let snapshot = reqwest::get(snapshot_url)
64 .await
65 .map_err(SocketError::Http)?
66 .json::<BinanceOrderBookL2Snapshot>()
67 .await
68 .map_err(SocketError::Http)?;
69
70 Ok(MarketEvent::from((
71 ExchangeId::BinanceFuturesUsd,
72 sub.instrument.key().clone(),
73 snapshot,
74 )))
75 }
76 });
77
78 try_join_all(l2_snapshot_futures)
79 }
80}
81
82#[derive(Debug)]
83pub struct BinanceFuturesUsdOrderBooksL2Transformer<InstrumentKey> {
84 instrument_map:
85 Map<BinanceOrderBookL2Meta<InstrumentKey, BinanceFuturesUsdOrderBookL2Sequencer>>,
86}
87
88impl<InstrumentKey> ExchangeTransformer<BinanceFuturesUsd, InstrumentKey, OrderBooksL2>
89 for BinanceFuturesUsdOrderBooksL2Transformer<InstrumentKey>
90where
91 InstrumentKey: Clone + PartialEq + Send + Sync,
92{
93 async fn init(
94 instrument_map: Map<InstrumentKey>,
95 initial_snapshots: &[MarketEvent<InstrumentKey, OrderBookEvent>],
96 _: UnboundedSender<WsMessage>,
97 ) -> Result<Self, DataError> {
98 let instrument_map = instrument_map
99 .0
100 .into_iter()
101 .map(|(sub_id, instrument_key)| {
102 let snapshot = initial_snapshots
103 .iter()
104 .find(|snapshot| snapshot.instrument == instrument_key)
105 .ok_or_else(|| DataError::InitialSnapshotMissing(sub_id.clone()))?;
106
107 let OrderBookEvent::Snapshot(snapshot) = &snapshot.kind else {
108 return Err(DataError::InitialSnapshotInvalid(String::from(
109 "expected OrderBookEvent::Snapshot but found OrderBookEvent::Update",
110 )));
111 };
112
113 let sequencer = BinanceFuturesUsdOrderBookL2Sequencer {
114 updates_processed: 0,
115 last_update_id: snapshot.sequence(),
116 };
117
118 Ok((
119 sub_id,
120 BinanceOrderBookL2Meta::new(instrument_key, sequencer),
121 ))
122 })
123 .collect::<Result<Map<_>, _>>()?;
124
125 Ok(Self { instrument_map })
126 }
127}
128
129impl<InstrumentKey> Transformer for BinanceFuturesUsdOrderBooksL2Transformer<InstrumentKey>
130where
131 InstrumentKey: Clone,
132{
133 type Error = DataError;
134 type Input = BinanceFuturesOrderBookL2Update;
135 type Output = MarketEvent<InstrumentKey, OrderBookEvent>;
136 type OutputIter = Vec<Result<Self::Output, Self::Error>>;
137
138 fn transform(&mut self, input: Self::Input) -> Self::OutputIter {
139 let subscription_id = match input.id() {
141 Some(subscription_id) => subscription_id,
142 None => return vec![],
143 };
144
145 let instrument = match self.instrument_map.find_mut(&subscription_id) {
147 Ok(instrument) => instrument,
148 Err(unidentifiable) => return vec![Err(DataError::from(unidentifiable))],
149 };
150
151 let valid_update = match instrument.sequencer.validate_sequence(input) {
153 Ok(Some(valid_update)) => valid_update,
154 Ok(None) => return vec![],
155 Err(error) => return vec![Err(error)],
156 };
157
158 MarketIter::<InstrumentKey, OrderBookEvent>::from((
159 BinanceFuturesUsd::ID,
160 instrument.key.clone(),
161 valid_update,
162 ))
163 .0
164 }
165}
166
167#[derive(Debug)]
193pub struct BinanceFuturesUsdOrderBookL2Sequencer {
194 pub updates_processed: u64,
195 pub last_update_id: u64,
196}
197
198impl BinanceFuturesUsdOrderBookL2Sequencer {
199 pub fn new(last_update_id: u64) -> Self {
201 Self {
202 updates_processed: 0,
203 last_update_id,
204 }
205 }
206
207 pub fn validate_sequence(
211 &mut self,
212 update: BinanceFuturesOrderBookL2Update,
213 ) -> Result<Option<BinanceFuturesOrderBookL2Update>, DataError> {
214 if update.last_update_id < self.last_update_id {
216 return Ok(None);
217 }
218
219 if self.is_first_update() {
220 self.validate_first_update(&update)?;
222 } else {
223 self.validate_next_update(&update)?;
225 }
226
227 self.updates_processed += 1;
229 self.last_update_id = update.last_update_id;
230
231 Ok(Some(update))
232 }
233
234 pub fn is_first_update(&self) -> bool {
239 self.updates_processed == 0
240 }
241
242 pub fn validate_first_update(
247 &self,
248 update: &BinanceFuturesOrderBookL2Update,
249 ) -> Result<(), DataError> {
250 if update.first_update_id <= self.last_update_id
251 && update.last_update_id >= self.last_update_id
252 {
253 Ok(())
254 } else {
255 Err(DataError::InvalidSequence {
256 prev_last_update_id: self.last_update_id,
257 first_update_id: update.first_update_id,
258 })
259 }
260 }
261
262 pub fn validate_next_update(
268 &self,
269 update: &BinanceFuturesOrderBookL2Update,
270 ) -> Result<(), DataError> {
271 if update.prev_last_update_id == self.last_update_id {
272 Ok(())
273 } else {
274 Err(DataError::InvalidSequence {
275 prev_last_update_id: self.last_update_id,
276 first_update_id: update.first_update_id,
277 })
278 }
279 }
280}
281
282#[derive(Clone, PartialEq, PartialOrd, Debug, Deserialize)]
307pub struct BinanceFuturesOrderBookL2Update {
308 #[serde(
309 alias = "s",
310 deserialize_with = "super::super::book::l2::de_ob_l2_subscription_id"
311 )]
312 pub subscription_id: SubscriptionId,
313 #[serde(
314 alias = "E",
315 deserialize_with = "rustrade_integration::serde::de::de_u64_epoch_ms_as_datetime_utc"
316 )]
317 pub time_exchange: DateTime<Utc>,
318 #[serde(
319 alias = "T",
320 deserialize_with = "rustrade_integration::serde::de::de_u64_epoch_ms_as_datetime_utc"
321 )]
322 pub time_engine: DateTime<Utc>,
323 #[serde(alias = "U")]
324 pub first_update_id: u64,
325 #[serde(alias = "u")]
326 pub last_update_id: u64,
327 #[serde(alias = "pu")]
328 pub prev_last_update_id: u64,
329 #[serde(alias = "b")]
330 pub bids: Vec<BinanceLevel>,
331 #[serde(alias = "a")]
332 pub asks: Vec<BinanceLevel>,
333}
334
335impl Identifier<Option<SubscriptionId>> for BinanceFuturesOrderBookL2Update {
336 fn id(&self) -> Option<SubscriptionId> {
337 Some(self.subscription_id.clone())
338 }
339}
340
341impl<InstrumentKey> From<(ExchangeId, InstrumentKey, BinanceFuturesOrderBookL2Update)>
342 for MarketIter<InstrumentKey, OrderBookEvent>
343{
344 fn from(
345 (exchange, instrument, update): (
346 ExchangeId,
347 InstrumentKey,
348 BinanceFuturesOrderBookL2Update,
349 ),
350 ) -> Self {
351 let time_received = Utc::now();
352 Self(vec![Ok(MarketEvent {
353 time_exchange: update.time_exchange,
354 time_received,
355 exchange,
356 instrument,
357 kind: OrderBookEvent::Update(OrderBook::new(
358 update.last_update_id,
359 OrderBookTimes {
361 time_engine: Some(update.time_engine),
362 time_exchange: Some(update.time_exchange),
363 time_received,
364 },
365 update.bids,
366 update.asks,
367 )),
368 })])
369 }
370}
371
372#[cfg(test)]
373#[allow(clippy::unwrap_used)] mod tests {
375 use super::*;
376 use crate::books::Level;
377 use rust_decimal_macros::dec;
378
379 #[test]
380 fn test_futures_order_book_times_distinct_not_transposed() {
381 let e = DateTime::from_timestamp_millis(1571889248277).unwrap(); let t = DateTime::from_timestamp_millis(1571889248276).unwrap(); assert_ne!(e, t);
387
388 let update = BinanceFuturesOrderBookL2Update {
389 subscription_id: SubscriptionId::from("@depth@100ms|BTCUSDT"),
390 time_exchange: e,
391 time_engine: t,
392 first_update_id: 157,
393 last_update_id: 160,
394 prev_last_update_id: 149,
395 bids: vec![BinanceLevel {
396 price: dec!(100),
397 amount: dec!(1),
398 }],
399 asks: vec![],
400 };
401
402 let MarketIter(mut events) = MarketIter::<&str, OrderBookEvent>::from((
403 ExchangeId::BinanceFuturesUsd,
404 "inst",
405 update,
406 ));
407 let OrderBookEvent::Update(book) = events.remove(0).unwrap().kind else {
408 panic!("expected Update");
409 };
410 assert_eq!(book.time_exchange(), Some(e), "time_exchange must be \"E\"");
411 assert_eq!(book.time_engine(), Some(t), "time_engine must be \"T\"");
412 }
413
414 #[test]
415 fn test_de_binance_futures_order_book_l2_update() {
416 let input = r#"
417 {
418 "e": "depthUpdate",
419 "E": 1571889248277,
420 "T": 1571889248276,
421 "s": "BTCUSDT",
422 "U": 157,
423 "u": 160,
424 "pu": 149,
425 "b": [
426 [
427 "0.0024",
428 "10"
429 ]
430 ],
431 "a": [
432 [
433 "0.0026",
434 "100"
435 ]
436 ]
437 }
438 "#;
439
440 assert_eq!(
441 serde_json::from_str::<BinanceFuturesOrderBookL2Update>(input).unwrap(),
442 BinanceFuturesOrderBookL2Update {
443 subscription_id: SubscriptionId::from("@depth@100ms|BTCUSDT"),
444 time_exchange: DateTime::from_timestamp_millis(1571889248277).unwrap(),
445 time_engine: DateTime::from_timestamp_millis(1571889248276).unwrap(),
446 first_update_id: 157,
447 last_update_id: 160,
448 prev_last_update_id: 149,
449 bids: vec![BinanceLevel {
450 price: dec!(0.0024),
451 amount: dec!(10.0)
452 },],
453 asks: vec![BinanceLevel {
454 price: dec!(0.0026),
455 amount: dec!(100.0)
456 },]
457 }
458 );
459 }
460
461 #[test]
462 fn test_sequencer_is_first_update() {
463 struct TestCase {
464 sequencer: BinanceFuturesUsdOrderBookL2Sequencer,
465 expected: bool,
466 }
467
468 let tests = vec![
469 TestCase {
470 sequencer: BinanceFuturesUsdOrderBookL2Sequencer::new(10),
472 expected: true,
473 },
474 TestCase {
475 sequencer: BinanceFuturesUsdOrderBookL2Sequencer {
477 updates_processed: 10,
478 last_update_id: 100,
479 },
480 expected: false,
481 },
482 ];
483
484 for (index, test) in tests.into_iter().enumerate() {
485 assert_eq!(
486 test.sequencer.is_first_update(),
487 test.expected,
488 "TC{} failed",
489 index
490 );
491 }
492 }
493
494 #[test]
495 fn test_sequencer_validate_first_update() {
496 struct TestCase {
497 updater: BinanceFuturesUsdOrderBookL2Sequencer,
498 input: BinanceFuturesOrderBookL2Update,
499 expected: Result<(), DataError>,
500 }
501
502 let tests = vec![
503 TestCase {
504 updater: BinanceFuturesUsdOrderBookL2Sequencer {
506 updates_processed: 0,
507 last_update_id: 100,
508 },
509 input: BinanceFuturesOrderBookL2Update {
510 subscription_id: SubscriptionId::from("subscription_id"),
511 time_exchange: Default::default(),
512 time_engine: Default::default(),
513 first_update_id: 100,
514 last_update_id: 110,
515 prev_last_update_id: 90,
516 bids: vec![],
517 asks: vec![],
518 },
519 expected: Ok(()),
520 },
521 TestCase {
522 updater: BinanceFuturesUsdOrderBookL2Sequencer {
524 updates_processed: 0,
525 last_update_id: 100,
526 },
527 input: BinanceFuturesOrderBookL2Update {
528 subscription_id: SubscriptionId::from("subscription_id"),
529 time_exchange: Default::default(),
530 time_engine: Default::default(),
531 first_update_id: 100,
532 last_update_id: 90,
533 prev_last_update_id: 90,
534 bids: vec![],
535 asks: vec![],
536 },
537 expected: Err(DataError::InvalidSequence {
538 prev_last_update_id: 100,
539 first_update_id: 100,
540 }),
541 },
542 TestCase {
543 updater: BinanceFuturesUsdOrderBookL2Sequencer {
545 updates_processed: 0,
546 last_update_id: 100,
547 },
548 input: BinanceFuturesOrderBookL2Update {
549 subscription_id: SubscriptionId::from("subscription_id"),
550 time_exchange: Default::default(),
551 time_engine: Default::default(),
552 first_update_id: 110,
553 last_update_id: 120,
554 prev_last_update_id: 90,
555 bids: vec![],
556 asks: vec![],
557 },
558 expected: Err(DataError::InvalidSequence {
559 prev_last_update_id: 100,
560 first_update_id: 110,
561 }),
562 },
563 TestCase {
564 updater: BinanceFuturesUsdOrderBookL2Sequencer {
566 updates_processed: 0,
567 last_update_id: 100,
568 },
569 input: BinanceFuturesOrderBookL2Update {
570 subscription_id: SubscriptionId::from("subscription_id"),
571 time_exchange: Default::default(),
572 time_engine: Default::default(),
573 first_update_id: 110,
574 last_update_id: 90,
575 prev_last_update_id: 90,
576 bids: vec![],
577 asks: vec![],
578 },
579 expected: Err(DataError::InvalidSequence {
580 prev_last_update_id: 100,
581 first_update_id: 110,
582 }),
583 },
584 ];
585
586 for (index, test) in tests.into_iter().enumerate() {
587 let actual = test.updater.validate_first_update(&test.input);
588 match (actual, test.expected) {
589 (Ok(_), Ok(_)) => {
590 }
592 (Err(actual), Err(expected)) => {
593 assert_eq!(actual, expected, "TC{index} error variant mismatch");
594 }
595 (actual, expected) => {
596 panic!(
598 "TC{index} failed because actual != expected. \nActual: {actual:?}\nExpected: {expected:?}\n"
599 );
600 }
601 }
602 }
603 }
604
605 #[test]
606 fn test_sequencer_validate_next_update() {
607 struct TestCase {
608 updater: BinanceFuturesUsdOrderBookL2Sequencer,
609 input: BinanceFuturesOrderBookL2Update,
610 expected: Result<(), DataError>,
611 }
612
613 let tests = vec![
614 TestCase {
615 updater: BinanceFuturesUsdOrderBookL2Sequencer {
617 updates_processed: 100,
618 last_update_id: 100,
619 },
620 input: BinanceFuturesOrderBookL2Update {
621 subscription_id: SubscriptionId::from("subscription_id"),
622 time_exchange: Default::default(),
623 time_engine: Default::default(),
624 first_update_id: 101,
625 last_update_id: 110,
626 prev_last_update_id: 100,
627 bids: vec![],
628 asks: vec![],
629 },
630 expected: Ok(()),
631 },
632 TestCase {
633 updater: BinanceFuturesUsdOrderBookL2Sequencer {
635 updates_processed: 100,
636 last_update_id: 100,
637 },
638 input: BinanceFuturesOrderBookL2Update {
639 subscription_id: SubscriptionId::from("subscription_id"),
640 time_exchange: Default::default(),
641 time_engine: Default::default(),
642 first_update_id: 100,
643 last_update_id: 90,
644 prev_last_update_id: 90,
645 bids: vec![],
646 asks: vec![],
647 },
648 expected: Err(DataError::InvalidSequence {
649 prev_last_update_id: 100,
650 first_update_id: 100,
651 }),
652 },
653 ];
654
655 for (index, test) in tests.into_iter().enumerate() {
656 let actual = test.updater.validate_next_update(&test.input);
657 match (actual, test.expected) {
658 (Ok(_), Ok(_)) => {
659 }
661 (Err(actual), Err(expected)) => {
662 assert_eq!(actual, expected, "TC{index} error variant mismatch");
663 }
664 (actual, expected) => {
665 panic!(
667 "TC{index} failed because actual != expected. \nActual: {actual:?}\nExpected: {expected:?}\n"
668 );
669 }
670 }
671 }
672 }
673
674 #[test]
675 fn test_update_rustrade_order_book_with_sequenced_updates() {
676 struct TestCase {
677 sequencer: BinanceFuturesUsdOrderBookL2Sequencer,
678 book: OrderBook,
679 input_update: BinanceFuturesOrderBookL2Update,
680 expected: OrderBook,
681 }
682
683 let tests = vec![
684 TestCase {
685 sequencer: BinanceFuturesUsdOrderBookL2Sequencer {
687 updates_processed: 100,
688 last_update_id: 100,
689 },
690 book: OrderBook::new(
691 0,
692 OrderBookTimes::default(),
693 vec![Level::new(50, 1)],
694 vec![Level::new(100, 1)],
695 ),
696 input_update: BinanceFuturesOrderBookL2Update {
697 subscription_id: SubscriptionId::from("subscription_id"),
698 time_exchange: Default::default(),
699 time_engine: Default::default(),
700 first_update_id: 0,
701 last_update_id: 0,
702 prev_last_update_id: 0,
703 bids: vec![],
704 asks: vec![],
705 },
706 expected: OrderBook::new(
707 0,
708 OrderBookTimes::default(),
709 vec![Level::new(50, 1)],
710 vec![Level::new(100, 1)],
711 ),
712 },
713 TestCase {
714 sequencer: BinanceFuturesUsdOrderBookL2Sequencer {
716 updates_processed: 100,
717 last_update_id: 100,
718 },
719 book: OrderBook::new(
720 100,
721 OrderBookTimes::default(),
722 vec![Level::new(80, 1), Level::new(100, 1), Level::new(90, 1)],
723 vec![Level::new(150, 1), Level::new(110, 1), Level::new(120, 1)],
724 ),
725 input_update: BinanceFuturesOrderBookL2Update {
726 subscription_id: SubscriptionId::from("subscription_id"),
727 time_exchange: Default::default(),
728 time_engine: Default::default(),
729 first_update_id: 101,
730 last_update_id: 110,
731 prev_last_update_id: 100,
732 bids: vec![
733 BinanceLevel {
735 price: dec!(80),
736 amount: dec!(0),
737 },
738 BinanceLevel {
740 price: dec!(90),
741 amount: dec!(10),
742 },
743 ],
744 asks: vec![
745 BinanceLevel {
747 price: dec!(200),
748 amount: dec!(1),
749 },
750 BinanceLevel {
752 price: dec!(500),
753 amount: dec!(0),
754 },
755 ],
756 },
757 expected: OrderBook::new(
758 110,
759 OrderBookTimes::default(),
760 vec![Level::new(100, 1), Level::new(90, 10)],
761 vec![
762 Level::new(110, 1),
763 Level::new(120, 1),
764 Level::new(150, 1),
765 Level::new(200, 1),
766 ],
767 ),
768 },
769 ];
770
771 for (index, mut test) in tests.into_iter().enumerate() {
772 if let Some(valid_update) = test.sequencer.validate_sequence(test.input_update).unwrap()
773 {
774 let rustrade_update = OrderBookEvent::Update(OrderBook::new(
775 valid_update.last_update_id,
776 OrderBookTimes::default(),
777 valid_update.bids,
778 valid_update.asks,
779 ));
780
781 test.book.update(&rustrade_update);
782 }
783
784 assert_eq!(test.book, test.expected, "TC{index} failed");
785 }
786 }
787}