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, Serialize};
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, Serialize)]
304pub struct BinanceFuturesOrderBookL2Update {
305 #[serde(
306 alias = "s",
307 deserialize_with = "super::super::book::l2::de_ob_l2_subscription_id"
308 )]
309 pub subscription_id: SubscriptionId,
310 #[serde(
311 alias = "E",
312 deserialize_with = "rustrade_integration::serde::de::de_u64_epoch_ms_as_datetime_utc"
313 )]
314 pub time_exchange: DateTime<Utc>,
315 #[serde(
316 alias = "T",
317 deserialize_with = "rustrade_integration::serde::de::de_u64_epoch_ms_as_datetime_utc"
318 )]
319 pub time_engine: DateTime<Utc>,
320 #[serde(alias = "U")]
321 pub first_update_id: u64,
322 #[serde(alias = "u")]
323 pub last_update_id: u64,
324 #[serde(alias = "pu")]
325 pub prev_last_update_id: u64,
326 #[serde(alias = "b")]
327 pub bids: Vec<BinanceLevel>,
328 #[serde(alias = "a")]
329 pub asks: Vec<BinanceLevel>,
330}
331
332impl Identifier<Option<SubscriptionId>> for BinanceFuturesOrderBookL2Update {
333 fn id(&self) -> Option<SubscriptionId> {
334 Some(self.subscription_id.clone())
335 }
336}
337
338impl<InstrumentKey> From<(ExchangeId, InstrumentKey, BinanceFuturesOrderBookL2Update)>
339 for MarketIter<InstrumentKey, OrderBookEvent>
340{
341 fn from(
342 (exchange, instrument, update): (
343 ExchangeId,
344 InstrumentKey,
345 BinanceFuturesOrderBookL2Update,
346 ),
347 ) -> Self {
348 let time_received = Utc::now();
349 Self(vec![Ok(MarketEvent {
350 time_exchange: update.time_exchange,
351 time_received,
352 exchange,
353 instrument,
354 kind: OrderBookEvent::Update(OrderBook::new(
355 update.last_update_id,
356 OrderBookTimes {
358 time_engine: Some(update.time_engine),
359 time_exchange: Some(update.time_exchange),
360 time_received,
361 },
362 update.bids,
363 update.asks,
364 )),
365 })])
366 }
367}
368
369#[cfg(test)]
370#[allow(clippy::unwrap_used)] mod tests {
372 use super::*;
373 use crate::books::Level;
374 use rust_decimal_macros::dec;
375
376 #[test]
377 fn test_futures_order_book_times_distinct_not_transposed() {
378 let e = DateTime::from_timestamp_millis(1571889248277).unwrap(); let t = DateTime::from_timestamp_millis(1571889248276).unwrap(); assert_ne!(e, t);
384
385 let update = BinanceFuturesOrderBookL2Update {
386 subscription_id: SubscriptionId::from("@depth@100ms|BTCUSDT"),
387 time_exchange: e,
388 time_engine: t,
389 first_update_id: 157,
390 last_update_id: 160,
391 prev_last_update_id: 149,
392 bids: vec![BinanceLevel {
393 price: dec!(100),
394 amount: dec!(1),
395 }],
396 asks: vec![],
397 };
398
399 let MarketIter(mut events) = MarketIter::<&str, OrderBookEvent>::from((
400 ExchangeId::BinanceFuturesUsd,
401 "inst",
402 update,
403 ));
404 let OrderBookEvent::Update(book) = events.remove(0).unwrap().kind else {
405 panic!("expected Update");
406 };
407 assert_eq!(book.time_exchange(), Some(e), "time_exchange must be \"E\"");
408 assert_eq!(book.time_engine(), Some(t), "time_engine must be \"T\"");
409 }
410
411 #[test]
412 fn test_de_binance_futures_order_book_l2_update() {
413 let input = r#"
414 {
415 "e": "depthUpdate",
416 "E": 1571889248277,
417 "T": 1571889248276,
418 "s": "BTCUSDT",
419 "U": 157,
420 "u": 160,
421 "pu": 149,
422 "b": [
423 [
424 "0.0024",
425 "10"
426 ]
427 ],
428 "a": [
429 [
430 "0.0026",
431 "100"
432 ]
433 ]
434 }
435 "#;
436
437 assert_eq!(
438 serde_json::from_str::<BinanceFuturesOrderBookL2Update>(input).unwrap(),
439 BinanceFuturesOrderBookL2Update {
440 subscription_id: SubscriptionId::from("@depth@100ms|BTCUSDT"),
441 time_exchange: DateTime::from_timestamp_millis(1571889248277).unwrap(),
442 time_engine: DateTime::from_timestamp_millis(1571889248276).unwrap(),
443 first_update_id: 157,
444 last_update_id: 160,
445 prev_last_update_id: 149,
446 bids: vec![BinanceLevel {
447 price: dec!(0.0024),
448 amount: dec!(10.0)
449 },],
450 asks: vec![BinanceLevel {
451 price: dec!(0.0026),
452 amount: dec!(100.0)
453 },]
454 }
455 );
456 }
457
458 #[test]
459 fn test_sequencer_is_first_update() {
460 struct TestCase {
461 sequencer: BinanceFuturesUsdOrderBookL2Sequencer,
462 expected: bool,
463 }
464
465 let tests = vec![
466 TestCase {
467 sequencer: BinanceFuturesUsdOrderBookL2Sequencer::new(10),
469 expected: true,
470 },
471 TestCase {
472 sequencer: BinanceFuturesUsdOrderBookL2Sequencer {
474 updates_processed: 10,
475 last_update_id: 100,
476 },
477 expected: false,
478 },
479 ];
480
481 for (index, test) in tests.into_iter().enumerate() {
482 assert_eq!(
483 test.sequencer.is_first_update(),
484 test.expected,
485 "TC{} failed",
486 index
487 );
488 }
489 }
490
491 #[test]
492 fn test_sequencer_validate_first_update() {
493 struct TestCase {
494 updater: BinanceFuturesUsdOrderBookL2Sequencer,
495 input: BinanceFuturesOrderBookL2Update,
496 expected: Result<(), DataError>,
497 }
498
499 let tests = vec![
500 TestCase {
501 updater: BinanceFuturesUsdOrderBookL2Sequencer {
503 updates_processed: 0,
504 last_update_id: 100,
505 },
506 input: BinanceFuturesOrderBookL2Update {
507 subscription_id: SubscriptionId::from("subscription_id"),
508 time_exchange: Default::default(),
509 time_engine: Default::default(),
510 first_update_id: 100,
511 last_update_id: 110,
512 prev_last_update_id: 90,
513 bids: vec![],
514 asks: vec![],
515 },
516 expected: Ok(()),
517 },
518 TestCase {
519 updater: BinanceFuturesUsdOrderBookL2Sequencer {
521 updates_processed: 0,
522 last_update_id: 100,
523 },
524 input: BinanceFuturesOrderBookL2Update {
525 subscription_id: SubscriptionId::from("subscription_id"),
526 time_exchange: Default::default(),
527 time_engine: Default::default(),
528 first_update_id: 100,
529 last_update_id: 90,
530 prev_last_update_id: 90,
531 bids: vec![],
532 asks: vec![],
533 },
534 expected: Err(DataError::InvalidSequence {
535 prev_last_update_id: 100,
536 first_update_id: 100,
537 }),
538 },
539 TestCase {
540 updater: BinanceFuturesUsdOrderBookL2Sequencer {
542 updates_processed: 0,
543 last_update_id: 100,
544 },
545 input: BinanceFuturesOrderBookL2Update {
546 subscription_id: SubscriptionId::from("subscription_id"),
547 time_exchange: Default::default(),
548 time_engine: Default::default(),
549 first_update_id: 110,
550 last_update_id: 120,
551 prev_last_update_id: 90,
552 bids: vec![],
553 asks: vec![],
554 },
555 expected: Err(DataError::InvalidSequence {
556 prev_last_update_id: 100,
557 first_update_id: 110,
558 }),
559 },
560 TestCase {
561 updater: BinanceFuturesUsdOrderBookL2Sequencer {
563 updates_processed: 0,
564 last_update_id: 100,
565 },
566 input: BinanceFuturesOrderBookL2Update {
567 subscription_id: SubscriptionId::from("subscription_id"),
568 time_exchange: Default::default(),
569 time_engine: Default::default(),
570 first_update_id: 110,
571 last_update_id: 90,
572 prev_last_update_id: 90,
573 bids: vec![],
574 asks: vec![],
575 },
576 expected: Err(DataError::InvalidSequence {
577 prev_last_update_id: 100,
578 first_update_id: 110,
579 }),
580 },
581 ];
582
583 for (index, test) in tests.into_iter().enumerate() {
584 let actual = test.updater.validate_first_update(&test.input);
585 match (actual, test.expected) {
586 (Ok(_), Ok(_)) => {
587 }
589 (Err(actual), Err(expected)) => {
590 assert_eq!(actual, expected, "TC{index} error variant mismatch");
591 }
592 (actual, expected) => {
593 panic!(
595 "TC{index} failed because actual != expected. \nActual: {actual:?}\nExpected: {expected:?}\n"
596 );
597 }
598 }
599 }
600 }
601
602 #[test]
603 fn test_sequencer_validate_next_update() {
604 struct TestCase {
605 updater: BinanceFuturesUsdOrderBookL2Sequencer,
606 input: BinanceFuturesOrderBookL2Update,
607 expected: Result<(), DataError>,
608 }
609
610 let tests = vec![
611 TestCase {
612 updater: BinanceFuturesUsdOrderBookL2Sequencer {
614 updates_processed: 100,
615 last_update_id: 100,
616 },
617 input: BinanceFuturesOrderBookL2Update {
618 subscription_id: SubscriptionId::from("subscription_id"),
619 time_exchange: Default::default(),
620 time_engine: Default::default(),
621 first_update_id: 101,
622 last_update_id: 110,
623 prev_last_update_id: 100,
624 bids: vec![],
625 asks: vec![],
626 },
627 expected: Ok(()),
628 },
629 TestCase {
630 updater: BinanceFuturesUsdOrderBookL2Sequencer {
632 updates_processed: 100,
633 last_update_id: 100,
634 },
635 input: BinanceFuturesOrderBookL2Update {
636 subscription_id: SubscriptionId::from("subscription_id"),
637 time_exchange: Default::default(),
638 time_engine: Default::default(),
639 first_update_id: 100,
640 last_update_id: 90,
641 prev_last_update_id: 90,
642 bids: vec![],
643 asks: vec![],
644 },
645 expected: Err(DataError::InvalidSequence {
646 prev_last_update_id: 100,
647 first_update_id: 100,
648 }),
649 },
650 ];
651
652 for (index, test) in tests.into_iter().enumerate() {
653 let actual = test.updater.validate_next_update(&test.input);
654 match (actual, test.expected) {
655 (Ok(_), Ok(_)) => {
656 }
658 (Err(actual), Err(expected)) => {
659 assert_eq!(actual, expected, "TC{index} error variant mismatch");
660 }
661 (actual, expected) => {
662 panic!(
664 "TC{index} failed because actual != expected. \nActual: {actual:?}\nExpected: {expected:?}\n"
665 );
666 }
667 }
668 }
669 }
670
671 #[test]
672 fn test_update_rustrade_order_book_with_sequenced_updates() {
673 struct TestCase {
674 sequencer: BinanceFuturesUsdOrderBookL2Sequencer,
675 book: OrderBook,
676 input_update: BinanceFuturesOrderBookL2Update,
677 expected: OrderBook,
678 }
679
680 let tests = vec![
681 TestCase {
682 sequencer: BinanceFuturesUsdOrderBookL2Sequencer {
684 updates_processed: 100,
685 last_update_id: 100,
686 },
687 book: OrderBook::new(
688 0,
689 OrderBookTimes::default(),
690 vec![Level::new(50, 1)],
691 vec![Level::new(100, 1)],
692 ),
693 input_update: BinanceFuturesOrderBookL2Update {
694 subscription_id: SubscriptionId::from("subscription_id"),
695 time_exchange: Default::default(),
696 time_engine: Default::default(),
697 first_update_id: 0,
698 last_update_id: 0,
699 prev_last_update_id: 0,
700 bids: vec![],
701 asks: vec![],
702 },
703 expected: OrderBook::new(
704 0,
705 OrderBookTimes::default(),
706 vec![Level::new(50, 1)],
707 vec![Level::new(100, 1)],
708 ),
709 },
710 TestCase {
711 sequencer: BinanceFuturesUsdOrderBookL2Sequencer {
713 updates_processed: 100,
714 last_update_id: 100,
715 },
716 book: OrderBook::new(
717 100,
718 OrderBookTimes::default(),
719 vec![Level::new(80, 1), Level::new(100, 1), Level::new(90, 1)],
720 vec![Level::new(150, 1), Level::new(110, 1), Level::new(120, 1)],
721 ),
722 input_update: BinanceFuturesOrderBookL2Update {
723 subscription_id: SubscriptionId::from("subscription_id"),
724 time_exchange: Default::default(),
725 time_engine: Default::default(),
726 first_update_id: 101,
727 last_update_id: 110,
728 prev_last_update_id: 100,
729 bids: vec![
730 BinanceLevel {
732 price: dec!(80),
733 amount: dec!(0),
734 },
735 BinanceLevel {
737 price: dec!(90),
738 amount: dec!(10),
739 },
740 ],
741 asks: vec![
742 BinanceLevel {
744 price: dec!(200),
745 amount: dec!(1),
746 },
747 BinanceLevel {
749 price: dec!(500),
750 amount: dec!(0),
751 },
752 ],
753 },
754 expected: OrderBook::new(
755 110,
756 OrderBookTimes::default(),
757 vec![Level::new(100, 1), Level::new(90, 10)],
758 vec![
759 Level::new(110, 1),
760 Level::new(120, 1),
761 Level::new(150, 1),
762 Level::new(200, 1),
763 ],
764 ),
765 },
766 ];
767
768 for (index, mut test) in tests.into_iter().enumerate() {
769 if let Some(valid_update) = test.sequencer.validate_sequence(test.input_update).unwrap()
770 {
771 let rustrade_update = OrderBookEvent::Update(OrderBook::new(
772 valid_update.last_update_id,
773 OrderBookTimes::default(),
774 valid_update.bids,
775 valid_update.asks,
776 ));
777
778 test.book.update(&rustrade_update);
779 }
780
781 assert_eq!(test.book, test.expected, "TC{index} failed");
782 }
783 }
784}