1use super::cancel_replace::{
40 CANCEL_REPLACE_PATH, CancelReplaceRequest, CancelTarget, SignedRest, cancel_replace_query,
41 open_refused_legs, read_legs, refused_legs,
42};
43use super::commission::{CommissionRates, fetch_commission_rates};
44use super::restatement::{
45 BalanceRestatement, SharedStreamClock, StreamClock, restate, restatement_time,
46};
47use super::rules::RulesLoader;
48use super::server_time::{ServerTime, SigningClock, default_server_time_sync, read_server_time};
49use super::shared::{
50 AbortOnDropStream, BINANCE_MAX_TRADES, BinanceOrderType, BinanceTimeInForce,
51 CONNECT_TIMEOUT_SECS, Disconnect, DisconnectOnDrop, ExecutionReport,
52 FILL_RECOVERY_TIMEOUT_SECS, HEARTBEAT_TIMEOUT_SECS, HandshakeRefusal, MyTradesFrom,
53 ORDER_EXECUTIONS_BUDGET, OpenOrderListing, PlacementResponse, RateLimitTracker, RequestKind,
54 SharedDedupCache, UnhandledEvents, UnrecoveredFills, UnsubscribeOnDrop, UserDataFrame,
55 WeightPool, account_position_time, answer_time, binance_filled_qty, classify_order_kind_tif,
56 classify_rest_order_error, classify_rest_query_error, classify_ws_order_error,
57 convert_ended_order, convert_execution_report, convert_open_order_listing,
58 convert_open_order_owned_symbol, dedup_key_from_event, disconnect_error, disconnect_lookback,
59 drop_after, frame_excerpt, gap_failed, gap_time, is_duplicate, is_unknown_order,
60 log_unhandled_event, log_unrecognised_frame, new_dedup_cache, parse_user_data_frame,
61 placed_order_state, recovered_order_cids, recovered_order_totals, response_decode_error,
62 rest_call_with_retry, rest_read_with_retry, retry_or_end, trailing_params, unix_ms,
63 unreadable_answer,
64};
65use crate::client::venue_clock::HostMoment;
66use crate::rules::{OrderRules, PreSendRules};
67use crate::{
68 AccountEventKind, AccountSnapshot, FillRecoveryFailure, InstrumentAccountSnapshot,
69 UnindexedAccountEvent, UnindexedAccountSnapshot,
70 balance::{AssetBalance, AssetBalanceUpdate, Balance, BalanceUpdate},
71 client::{
72 ExecutionClient, OrderStatusClient,
73 order_recovery::{
74 KnownLiveOrders, OpenListing, OrderLookup, SharedKnownLiveOrders, UncheckedOrders,
75 fetch_ended_by_key, recover_ended_orders,
76 },
77 reconnect_budget::{BackoffLimits, ReconnectBudget},
78 },
79 emit_stream_terminated,
80 error::{
81 ApiError, ConnectivityError, NotPlacedReason, OrderError, UnindexedClientError,
82 UnindexedOrderError,
83 },
84 margin::MarginEffect,
85 order::{
86 Order, OrderKey, OrderKind, TimeInForce, UnindexedInactiveOrder, UnindexedOrderKey,
87 id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
88 request::{
89 CancelReplaceAnswer, OnCancelFailure, OrderRequestCancel, OrderRequestOpen,
90 UnindexedOrderResponseCancel,
91 },
92 state::{Cancelled, Open, OrderState, UnindexedOrderState},
93 },
94 parse_env_bool,
95 position::PositionReport,
96 trade::{AssetFees, TimeSource, Trade, TradeId, TradesRead},
97};
98use binance_sdk::{
99 common::{
100 config::{ConfigurationRestApi, ConfigurationWebsocketApi},
101 constants::{SPOT_REST_API_PROD_URL, SPOT_REST_API_TESTNET_URL},
102 models::WebsocketEvent,
103 },
104 spot::{
105 SpotRestApi, SpotWsApi,
106 rest_api::{
107 GetAccountParams, GetOpenOrdersParams, GetOrderParams, MyTradesParams, RestApi,
108 },
109 websocket_api::{
110 OrderCancelParams, OrderPlaceParams, OrderPlaceSideEnum, OrderPlaceTimeInForceEnum,
111 OrderPlaceTypeEnum, UserDataStreamSubscribeSignatureParams, WebsocketApi,
112 WebsocketApiHandle,
113 },
114 },
115};
116use chrono::{DateTime, TimeDelta, TimeZone, Utc};
117use fnv::FnvHashSet;
118use futures::stream::BoxStream;
119use rust_decimal::Decimal;
120use rustrade_instrument::{
121 Side,
122 asset::name::AssetNameExchange,
123 exchange::ExchangeId,
124 instrument::{kind::InstrumentKindDiscriminant, name::InstrumentNameExchange},
125};
126use serde::Deserialize;
127use smol_str::format_smolstr;
128use std::{
129 collections::HashMap,
130 str::FromStr,
131 sync::{
132 Arc,
133 atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
134 },
135 time::Duration,
136};
137use tokio::sync::{RwLock, mpsc, oneshot};
138use tracing::{debug, error, info, trace, warn};
139
140#[derive(Clone, Deserialize)]
147pub struct BinanceSpotConfig {
148 api_key: String,
151 secret_key: String,
152
153 #[serde(default = "default_testnet")]
155 pub testnet: bool,
156
157 #[serde(default)]
161 pub pre_send_rules: PreSendRules,
162
163 #[serde(default = "crate::client::binance::server_time::default_server_time_sync")]
167 pub server_time_sync: bool,
168}
169
170fn default_testnet() -> bool {
176 true
177}
178
179impl std::fmt::Debug for BinanceSpotConfig {
181 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
182 f.debug_struct("BinanceSpotConfig")
183 .field("api_key", &"***")
184 .field("secret_key", &"***")
185 .field("testnet", &self.testnet)
186 .field("pre_send_rules", &self.pre_send_rules)
187 .field("server_time_sync", &self.server_time_sync)
188 .finish()
189 }
190}
191
192impl BinanceSpotConfig {
193 pub fn new(api_key: String, secret_key: String) -> Self {
196 Self::testnet(api_key, secret_key)
197 }
198
199 pub fn testnet(api_key: String, secret_key: String) -> Self {
201 Self {
202 api_key,
203 secret_key,
204 testnet: true,
205 pre_send_rules: PreSendRules::default(),
206 server_time_sync: default_server_time_sync(),
207 }
208 }
209
210 #[must_use]
212 pub fn with_pre_send_rules(mut self, pre_send_rules: PreSendRules) -> Self {
213 self.pre_send_rules = pre_send_rules;
214 self
215 }
216
217 #[must_use]
220 pub fn with_server_time_sync(mut self, server_time_sync: bool) -> Self {
221 self.server_time_sync = server_time_sync;
222 self
223 }
224
225 pub fn production(api_key: String, secret_key: String) -> Self {
230 Self {
231 api_key,
232 secret_key,
233 testnet: false,
234 pre_send_rules: PreSendRules::default(),
235 server_time_sync: default_server_time_sync(),
236 }
237 }
238
239 pub fn from_env() -> Result<Self, BinanceSpotConfigError> {
258 let api_key = match std::env::var("BINANCE_API_KEY") {
259 Ok(value) => value,
260 Err(std::env::VarError::NotPresent) => {
261 return Err(BinanceSpotConfigError::MissingApiKey);
262 }
263 Err(std::env::VarError::NotUnicode(_)) => {
264 return Err(BinanceSpotConfigError::InvalidApiKey);
265 }
266 };
267 let secret_key = match std::env::var("BINANCE_SECRET_KEY") {
268 Ok(value) => value,
269 Err(std::env::VarError::NotPresent) => {
270 return Err(BinanceSpotConfigError::MissingSecretKey);
271 }
272 Err(std::env::VarError::NotUnicode(_)) => {
273 return Err(BinanceSpotConfigError::InvalidSecretKey);
274 }
275 };
276 let testnet = match std::env::var("BINANCE_TESTNET") {
277 Ok(value) => {
278 parse_env_bool(&value).ok_or(BinanceSpotConfigError::InvalidTestnet(value))?
279 }
280 Err(std::env::VarError::NotPresent) => true,
281 Err(std::env::VarError::NotUnicode(value)) => {
284 return Err(BinanceSpotConfigError::InvalidTestnet(
285 value.to_string_lossy().into_owned(),
286 ));
287 }
288 };
289
290 if testnet {
291 Ok(Self::testnet(api_key, secret_key))
292 } else {
293 Ok(Self::production(api_key, secret_key))
294 }
295 }
296
297 pub fn api_key(&self) -> &str {
299 &self.api_key
300 }
301}
302
303#[derive(Debug, PartialEq, thiserror::Error)]
304pub enum BinanceSpotConfigError {
305 #[error("BINANCE_API_KEY environment variable not set")]
306 MissingApiKey,
307
308 #[error("BINANCE_API_KEY environment variable is not valid UTF-8")]
311 InvalidApiKey,
312
313 #[error("BINANCE_SECRET_KEY environment variable not set")]
314 MissingSecretKey,
315
316 #[error("BINANCE_SECRET_KEY environment variable is not valid UTF-8")]
317 InvalidSecretKey,
318
319 #[error("BINANCE_TESTNET must be true or false, got {0}")]
320 InvalidTestnet(String),
321}
322
323#[derive(Debug)]
330enum WsApiUnavailable {
331 RateLimited {
334 msg: String,
335 retry_after: Option<Duration>,
336 },
337 Connect(String),
339}
340
341impl WsApiUnavailable {
342 fn into_order_error(self) -> UnindexedOrderError {
345 match self {
346 Self::RateLimited { msg, retry_after } => {
347 debug!(%msg, "BinanceSpot WS-API rate-limited, request not sent");
350 UnindexedOrderError::Rejected(ApiError::rate_limited(retry_after))
351 }
352 Self::Connect(msg) => UnindexedOrderError::Connectivity(ConnectivityError::Socket(msg)),
353 }
354 }
355
356 fn into_open_error(self) -> UnindexedOrderError {
360 match self {
361 Self::Connect(msg) => UnindexedOrderError::NotPlaced(NotPlacedReason::NotSent(
362 ConnectivityError::Socket(msg),
363 )),
364 rate_limited @ Self::RateLimited { .. } => rate_limited.into_order_error(),
365 }
366 }
367}
368
369#[derive(Clone)]
412pub struct BinanceSpot {
413 config: Arc<BinanceSpotConfig>,
414 rest: Arc<RestApi>,
415 ws_handle: WebsocketApiHandle,
417 ws_api: Arc<RwLock<Option<DisconnectOnDrop<WebsocketApi>>>>,
423 rate_limiter: Arc<RateLimitTracker>,
425 known_live: SharedKnownLiveOrders,
428 rules: Arc<RulesLoader>,
431 signed_rest: Arc<SignedRest>,
433}
434
435impl std::fmt::Debug for BinanceSpot {
436 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
437 f.debug_struct("BinanceSpot")
438 .field("testnet", &self.config.testnet)
439 .finish_non_exhaustive()
440 }
441}
442
443impl BinanceSpot {
444 pub async fn load_rules(
453 &self,
454 instruments: &[InstrumentNameExchange],
455 ) -> Result<HashMap<InstrumentNameExchange, OrderRules>, UnindexedClientError> {
456 self.rules.load(instruments).await
457 }
458
459 pub async fn fetch_commission_rates(
471 &self,
472 instrument: &InstrumentNameExchange,
473 ) -> Result<CommissionRates, UnindexedClientError> {
474 fetch_commission_rates(&self.rest, &self.rate_limiter, instrument).await
475 }
476
477 pub fn host_clock_lead(&self) -> Option<TimeDelta> {
485 self.rate_limiter.host_clock_lead()
486 }
487
488 fn remember_live<'a>(
490 &self,
491 orders: impl IntoIterator<Item = &'a Order<ExchangeId, InstrumentNameExchange, Open>>,
492 ) {
493 let mut known = self.known_live.lock();
494 for order in orders {
495 known.live(&order.key, order.quantity, order.kind, &order.state);
496 }
497 }
498
499 #[allow(clippy::expect_used)] fn build_rest(config: &BinanceSpotConfig, signing: Option<&SigningClock>) -> Arc<RestApi> {
503 let mut builder = ConfigurationRestApi::builder()
504 .api_key(config.api_key.clone())
505 .api_secret(config.secret_key.clone());
506 if let Some(signing) = signing {
507 builder = builder.timestamp_provider(signing.provider());
508 }
509 let rest_config = builder
510 .build()
511 .expect("failed to build Binance REST configuration");
512
513 Arc::new(if config.testnet {
514 SpotRestApi::testnet(rest_config)
515 } else {
516 SpotRestApi::production(rest_config)
517 })
518 }
519
520 fn build_signed_rest(
526 config: &BinanceSpotConfig,
527 signing: Option<&SigningClock>,
528 ) -> Arc<SignedRest> {
529 let base_url = if config.testnet {
530 SPOT_REST_API_TESTNET_URL
531 } else {
532 SPOT_REST_API_PROD_URL
533 };
534 Arc::new(SignedRest::new(
535 base_url.to_string(),
536 config.api_key.clone(),
537 config.secret_key.clone(),
538 signing.map(SigningClock::provider),
539 Duration::from_secs(5),
540 ))
541 }
542
543 async fn send_cancel_replace(
546 &self,
547 cancel: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
548 open: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
549 target: CancelTarget,
550 fields: SpotOrderFields,
551 on_cancel_failure: OnCancelFailure,
552 ) -> CancelReplaceAnswer {
553 let instrument = open.key.instrument.clone();
554 let cancel_key = OrderKey::new(
555 ExchangeId::BinanceSpot,
556 instrument.clone(),
557 cancel.key.strategy.clone(),
558 cancel.key.cid.clone(),
559 );
560 let order_key = OrderKey::new(
561 ExchangeId::BinanceSpot,
562 instrument.clone(),
563 open.key.strategy.clone(),
564 open.key.cid.clone(),
565 );
566 let (side, price, quantity, kind, time_in_force) = (
567 open.state.side,
568 open.state.price,
569 open.state.quantity,
570 open.state.kind,
571 open.state.time_in_force,
572 );
573
574 let (cancelled, opened) = 'legs: {
575 if let Err(refused) = self
577 .rules
578 .check(instrument.name(), side, kind, price, quantity)
579 .await
580 {
581 let cancelled = self.cancel_order(cancel).await.state;
582 break 'legs open_refused_legs(cancelled, refused, on_cancel_failure);
583 }
584
585 let request = CancelReplaceRequest {
586 symbol: instrument.name(),
587 cancel: target,
588 cid: order_key.cid.clone(),
589 side,
590 quantity,
591 fields,
592 on_cancel_failure,
593 };
594 let query = match request.params() {
595 Ok(params) => cancel_replace_query(params),
596 Err(e) => {
597 error!(%e, "BinanceSpot failed to build cancel-replace params");
598 let cancelled = self.cancel_order(cancel).await.state;
599 let refused = OrderError::Rejected(ApiError::OrderRejected(e.to_string()));
600 break 'legs open_refused_legs(cancelled, refused, on_cancel_failure);
601 }
602 };
603
604 let answer = rest_call_with_retry(
605 &self.signed_rest,
606 &self.rate_limiter,
607 RequestKind::Order,
608 |rest| {
609 let query = query.clone();
610 Box::pin(async move {
611 let answer = rest
612 .send(reqwest::Method::POST, CANCEL_REPLACE_PATH, query)
613 .await?;
614 if answer.reports_legs() {
615 Ok(answer)
616 } else {
617 Err(answer.into_connector_error().into())
618 }
619 })
620 },
621 )
622 .await;
623 match answer {
624 Ok(answer) => read_legs(&answer, &instrument, quantity),
625 Err(e) => refused_legs(
626 classify_rest_order_error(&e, &instrument),
627 on_cancel_failure,
628 ),
629 }
630 };
631
632 {
633 let mut known_live = self.known_live.lock();
634 if cancelled.is_ok() {
635 known_live.ended(&cancel_key.cid);
636 }
637 if let Some(state) = &opened {
638 known_live.placed(&order_key, quantity, kind, state);
639 }
640 }
641
642 CancelReplaceAnswer {
643 cancel: UnindexedOrderResponseCancel {
644 key: cancel_key,
645 state: cancelled,
646 },
647 open: opened.map(|state| Order {
648 key: order_key,
649 side,
650 price,
651 quantity,
652 kind,
653 time_in_force,
654 state,
655 }),
656 }
657 }
658
659 #[allow(clippy::expect_used)] fn build_ws_handle(
663 config: &BinanceSpotConfig,
664 signing: Option<&SigningClock>,
665 ) -> WebsocketApiHandle {
666 let mut builder = ConfigurationWebsocketApi::builder()
667 .api_key(config.api_key.clone())
668 .api_secret(config.secret_key.clone());
669 if let Some(signing) = signing {
670 builder = builder.timestamp_provider(signing.provider());
671 }
672 let ws_config = builder
673 .build()
674 .expect("failed to build Binance WebSocket configuration");
675
676 if config.testnet {
677 SpotWsApi::testnet(ws_config)
678 } else {
679 SpotWsApi::production(ws_config)
680 }
681 }
682
683 async fn get_ws_api(&self) -> Result<WebsocketApi, WsApiUnavailable> {
694 if let Err(cooldown) = self.rate_limiter.wait_if_blocked(RequestKind::Order).await {
695 return Err(WsApiUnavailable::RateLimited {
696 msg: cooldown.to_string(),
697 retry_after: Some(cooldown.remaining),
698 });
699 }
700 {
702 let guard = self.ws_api.read().await;
703 if let Some(ref ws) = *guard {
704 return Ok(WebsocketApi::clone(ws));
705 }
706 }
707 let mut guard = self.ws_api.write().await;
714 if let Some(ref ws) = *guard {
716 return Ok(WebsocketApi::clone(ws));
717 }
718 if let Some(remaining) = self.rate_limiter.cooldown_remaining() {
719 return Err(WsApiUnavailable::RateLimited {
720 msg: "rate-limit cooldown active, not connecting the WS-API session".into(),
721 retry_after: Some(remaining),
722 });
723 }
724 let ws = match tokio::time::timeout(
725 Duration::from_secs(CONNECT_TIMEOUT_SECS),
726 self.ws_handle.connect(),
727 )
728 .await
729 {
730 Ok(Ok(ws)) => ws,
731 Ok(Err(e)) => {
732 let Some(refusal) = HandshakeRefusal::of(&e) else {
733 return Err(WsApiUnavailable::Connect(format!("{e:#}")));
734 };
735 warn!(error = %format!("{e:#}"), "Binance refused the WS-API handshake");
737 let retry_after = refusal.retry_after();
738 self.rate_limiter.on_rate_limited(retry_after);
739 return Err(WsApiUnavailable::RateLimited {
740 msg: format!("{e:#}"),
741 retry_after,
742 });
743 }
744 Err(_) => {
745 return Err(WsApiUnavailable::Connect(format!(
746 "BinanceSpot WS connect timed out after {CONNECT_TIMEOUT_SECS}s"
747 )));
748 }
749 };
750 *guard = Some(DisconnectOnDrop::new(
751 ws.clone(),
752 "BinanceSpot order session",
753 ));
754 Ok(ws)
755 }
756
757 async fn clear_ws_api(&self) {
760 let ws = {
772 let mut guard = self.ws_api.write().await;
773 guard.take()
774 };
775 if let Some(ws) = ws {
776 match tokio::time::timeout(Duration::from_secs(5), ws.disconnect()).await {
778 Ok(Err(e)) => warn!(%e, "BinanceSpot failed to disconnect stale WS session"),
779 Err(_) => warn!("BinanceSpot WS disconnect timed out (5s)"),
780 Ok(Ok(())) => {}
781 }
782 }
783 }
784}
785
786async fn fetch_open_orders_for_instrument(
793 rest: Arc<RestApi>,
794 rate_limiter: Arc<RateLimitTracker>,
795 instrument: InstrumentNameExchange,
796 kind: RequestKind,
797) -> Result<(InstrumentNameExchange, OpenOrderListing), UnindexedClientError> {
798 let symbol_str = instrument.name().to_string();
800 let response = rest_call_with_retry(&rest, &rate_limiter, kind, |rest| {
801 let sym = symbol_str.clone();
802 Box::pin(async move {
803 let params = GetOpenOrdersParams::builder().symbol(sym).build()?;
804 rest.get_open_orders(params).await
805 })
806 })
807 .await
808 .map_err(|e| classify_rest_query_error(&e, Some(&instrument)))?;
809
810 let orders_data = response.data().await.map_err(response_decode_error)?;
811
812 let listing =
814 convert_open_order_listing(&orders_data, ExchangeId::BinanceSpot, &instrument, &|_| {
815 None
816 });
817
818 Ok((instrument, listing))
819}
820
821async fn fetch_all_open_orders(
826 rest: Arc<RestApi>,
827 rate_limiter: Arc<RateLimitTracker>,
828) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
829 let response = rest_call_with_retry(&rest, &rate_limiter, RequestKind::Query, |rest| {
830 Box::pin(async move {
831 let params = GetOpenOrdersParams::builder().build()?;
832 rest.get_open_orders(params).await
833 })
834 })
835 .await
836 .map_err(|e| classify_rest_query_error(&e, None))?;
837
838 let orders_data = response.data().await.map_err(response_decode_error)?;
839
840 let orders = orders_data
841 .into_iter()
842 .filter_map(|o| convert_open_order_owned_symbol(&o, ExchangeId::BinanceSpot, &|_| None))
844 .collect();
845
846 Ok(orders)
847}
848
849async fn fetch_order_lookup(
851 rest: Arc<RestApi>,
852 rate_limiter: Arc<RateLimitTracker>,
853 key: UnindexedOrderKey,
854 kind: RequestKind,
855) -> Result<OrderLookup, UnindexedClientError> {
856 let symbol = key.instrument.name().to_string();
858 let cid = key.cid.0.to_string();
859 let response = match rest_call_with_retry(&rest, &rate_limiter, kind, |rest| {
860 let (symbol, cid) = (symbol.clone(), cid.clone());
861 Box::pin(async move {
862 let params = GetOrderParams::builder(symbol)
863 .orig_client_order_id(cid)
864 .build()?;
865 rest.get_order(params).await
866 })
867 })
868 .await
869 {
870 Ok(response) => response,
871 Err(e) if is_unknown_order(&e) => {
872 debug!(instrument = %key.instrument, cid = %key.cid, error = %e, "BinanceSpot does not know this order");
873 return Ok(OrderLookup::Unknown);
874 }
875 Err(e) => return Err(classify_rest_query_error(&e, Some(&key.instrument))),
876 };
877
878 let row = response.data().await.map_err(response_decode_error)?;
879
880 Ok(
882 convert_ended_order(&row, ExchangeId::BinanceSpot, &key, None)
883 .map_or(OrderLookup::NotEnded, |order| {
884 OrderLookup::Ended(Box::new(order))
885 }),
886 )
887}
888
889async fn fetch_order_cid(
893 rest: &Arc<RestApi>,
894 rate_limiter: &RateLimitTracker,
895 instrument: &InstrumentNameExchange,
896 order_id: i64,
897) -> Result<Option<ClientOrderId>, UnindexedClientError> {
898 let symbol = instrument.name().to_string();
899 let response = match rest_call_with_retry(rest, rate_limiter, RequestKind::Essential, |rest| {
900 let symbol = symbol.clone();
901 Box::pin(async move {
902 let params = GetOrderParams::builder(symbol).order_id(order_id).build()?;
903 rest.get_order(params).await
904 })
905 })
906 .await
907 {
908 Ok(response) => response,
909 Err(e) if is_unknown_order(&e) => {
910 debug!(%instrument, order_id, error = %e, "BinanceSpot does not know this order");
911 return Ok(None);
912 }
913 Err(e) => return Err(classify_rest_query_error(&e, Some(instrument))),
914 };
915 let row = response.data().await.map_err(response_decode_error)?;
916 Ok(row
917 .client_order_id
918 .filter(|cid| !cid.is_empty())
919 .map(ClientOrderId::new))
920}
921
922async fn listed_open_cids(
925 rest: Arc<RestApi>,
926 rate_limiter: Arc<RateLimitTracker>,
927 instruments: Vec<InstrumentNameExchange>,
928) -> Result<FnvHashSet<ClientOrderId>, UnindexedClientError> {
929 let mut listed = FnvHashSet::default();
930 for instrument in instruments {
931 let (_, listing) = fetch_open_orders_for_instrument(
932 rest.clone(),
933 rate_limiter.clone(),
934 instrument,
935 RequestKind::Essential,
936 )
937 .await?;
938 listed.extend(listing.orders.into_iter().map(|order| order.key.cid));
939 }
940 Ok(listed)
941}
942
943async fn recover_spot_ended_orders(
946 rest: &Arc<RestApi>,
947 rate_limiter: &Arc<RateLimitTracker>,
948 known: &SharedKnownLiveOrders,
949 unchecked: &mut UncheckedOrders,
950 unrecovered: &UnrecoveredFills,
951 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
952) {
953 recover_ended_orders(
954 ExchangeId::BinanceSpot,
955 known,
956 unchecked,
957 unrecovered,
958 tx,
959 OpenListing::PerInstrument,
960 |instruments| listed_open_cids(rest.clone(), rate_limiter.clone(), instruments),
961 |key| {
962 fetch_order_lookup(
963 rest.clone(),
964 rate_limiter.clone(),
965 key,
966 RequestKind::Essential,
967 )
968 },
969 )
970 .await;
971}
972
973async fn paginate_my_trades(
984 rest: &Arc<RestApi>,
985 rate_limiter: &Arc<RateLimitTracker>,
986 instrument: &InstrumentNameExchange,
987 from: MyTradesFrom,
988 kind: RequestKind,
989) -> Result<Vec<binance_sdk::spot::rest_api::MyTradesResponseInner>, UnindexedClientError> {
990 let symbol_str = instrument.name().to_string();
992 let mut all_pages = Vec::new();
997 let mut cursor: Option<i64> = None;
998 loop {
999 let fid = cursor; let response = rest_call_with_retry(rest, rate_limiter, kind, |rest| {
1001 let sym = symbol_str.clone();
1002 Box::pin(async move {
1003 #[allow(clippy::cast_possible_truncation)]
1005 let builder = MyTradesParams::builder(sym).limit(BINANCE_MAX_TRADES as i32);
1006 let params = match (from, fid) {
1007 (MyTradesFrom::Span { start, .. }, None) => builder.start_time(start),
1008 (MyTradesFrom::Span { .. }, Some(id)) => builder.from_id(id),
1009 (MyTradesFrom::Order(order_id), None) => builder.order_id(order_id),
1010 (MyTradesFrom::Order(order_id), Some(id)) => {
1011 builder.order_id(order_id).from_id(id)
1012 }
1013 }
1014 .build()?;
1015 rest.my_trades(params).await
1016 })
1017 })
1018 .await
1019 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
1020
1021 let mut page = response.data().await.map_err(response_decode_error)?;
1022
1023 let page_len = page.len();
1024 let last_id = page.last().and_then(|t| t.id);
1025 let past_end = match from {
1027 MyTradesFrom::Span { end, .. } => drop_after(&mut page, end),
1028 MyTradesFrom::Order(_) => false,
1029 };
1030 all_pages.extend(page);
1031
1032 if past_end || page_len < BINANCE_MAX_TRADES {
1033 break;
1034 }
1035 match last_id {
1036 Some(id) => {
1037 debug!(%instrument, "BinanceSpot paginate_my_trades: fetching next page ({page_len} results)");
1038 match id.checked_add(1) {
1039 Some(next) => cursor = Some(next),
1040 None => break, }
1042 }
1043 None => {
1044 warn!(%instrument, "BinanceSpot paginate_my_trades: trade missing ID, stopping pagination");
1045 break;
1046 }
1047 }
1048 }
1049 Ok(all_pages)
1050}
1051
1052impl ExecutionClient for BinanceSpot {
1057 const EXCHANGE: ExchangeId = ExchangeId::BinanceSpot;
1058
1059 const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] =
1062 &[InstrumentKindDiscriminant::Spot];
1063 type Config = BinanceSpotConfig;
1064 type AccountStream = BoxStream<'static, UnindexedAccountEvent>;
1065
1066 fn new(config: Self::Config) -> Self {
1071 let signing = config.server_time_sync.then(SigningClock::default);
1072 let rest = Self::build_rest(&config, signing.as_ref());
1073 let ws_handle = Self::build_ws_handle(&config, signing.as_ref());
1074 let signed_rest = Self::build_signed_rest(&config, signing.as_ref());
1075 let server_time =
1077 signing.map(|signing| ServerTime::new(signing, read_server_time(Arc::clone(&rest))));
1078 let rate_limiter =
1079 Arc::new(RateLimitTracker::new(WeightPool::Spot).with_server_time(server_time));
1080 let rules = Arc::new(RulesLoader::new(
1081 config.pre_send_rules,
1082 Arc::clone(&rest),
1083 Arc::clone(&rate_limiter),
1084 ));
1085 Self {
1086 config: Arc::new(config),
1087 rest,
1088 ws_handle,
1089 ws_api: Arc::new(RwLock::new(None)),
1090 rate_limiter,
1091 known_live: KnownLiveOrders::shared(ExchangeId::BinanceSpot),
1092 rules,
1093 signed_rest,
1094 }
1095 }
1096
1097 async fn account_snapshot(
1098 &self,
1099 assets: &[AssetNameExchange],
1100 instruments: &[InstrumentNameExchange],
1101 ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
1102 let (response, sent) =
1104 rest_read_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1105 Box::pin(async move {
1106 let params = GetAccountParams::builder().build()?;
1107 rest.get_account(params).await
1108 })
1109 })
1110 .await
1111 .map_err(|e| classify_rest_query_error(&e, None))?;
1112
1113 let account = response.data().await.map_err(response_decode_error)?;
1114
1115 let balances = filter_and_convert_balances(
1117 account.balances.unwrap_or_default(),
1118 assets,
1119 self.rate_limiter.balance_read_time(sent),
1120 );
1121
1122 use futures::{StreamExt as _, TryStreamExt};
1128 let instrument_snapshots: Vec<_> =
1129 futures::stream::iter(instruments.iter().cloned().map(|instrument| {
1130 fetch_open_orders_for_instrument(
1131 self.rest.clone(),
1132 self.rate_limiter.clone(),
1133 instrument,
1134 RequestKind::Query,
1135 )
1136 }))
1137 .buffer_unordered(8)
1138 .map(|result| {
1139 let (inst, listing) = result?;
1140 self.remember_live(&listing.orders);
1141 let wrapped = listing
1142 .orders
1143 .into_iter()
1144 .map(|o| Order {
1145 key: o.key,
1146 side: o.side,
1147 price: o.price,
1148 quantity: o.quantity,
1149 kind: o.kind,
1150 time_in_force: o.time_in_force,
1151 state: OrderState::active(o.state),
1152 })
1153 .collect();
1154 Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
1155 inst,
1156 wrapped,
1157 listing.complete,
1158 PositionReport::Unreported,
1159 None,
1160 ))
1161 })
1162 .try_collect()
1163 .await?;
1164
1165 Ok(AccountSnapshot::new(
1166 ExchangeId::BinanceSpot,
1167 balances,
1168 instrument_snapshots,
1169 ))
1170 }
1171
1172 async fn account_stream(
1253 &self,
1254 assets: &[AssetNameExchange],
1258 instruments: &[InstrumentNameExchange],
1259 ) -> Result<Self::AccountStream, UnindexedClientError> {
1260 let (tx, rx) = mpsc::unbounded_channel::<UnindexedAccountEvent>();
1275 let dedup = new_dedup_cache();
1276 let ws_handle = self.ws_handle.clone();
1277 let rest = self.rest.clone();
1278 let rate_limiter = self.rate_limiter.clone();
1279 let known_live = self.known_live.clone();
1280 let instruments = instruments.to_vec();
1281 debug_assert!(
1285 instruments.iter().all(|i| i.name().len() <= 23),
1286 "instrument name exceeds SmolStr inline capacity: {:?}",
1287 instruments.iter().find(|i| i.name().len() > 23)
1288 );
1289
1290 self.rate_limiter.before_stream_open().await?;
1292
1293 let initial_ws = ws_handle.connect().await.map_err(|e| {
1297 self.rate_limiter.stream_open_error(&e, |e| {
1298 UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
1299 })
1300 })?;
1301 let initial_ws = DisconnectOnDrop::new(initial_ws, ACCOUNT_STREAM_WS);
1303
1304 #[allow(clippy::expect_used)] let params = UserDataStreamSubscribeSignatureParams::builder()
1306 .build()
1307 .expect("UserDataStreamSubscribeSignatureParams has no required fields");
1308
1309 match initial_ws
1310 .user_data_stream_subscribe_signature(params)
1311 .await
1312 {
1313 Ok(_) => {}
1314 Err(e) => {
1315 match tokio::time::timeout(Duration::from_secs(5), initial_ws.disconnect()).await {
1319 Ok(Err(de)) => {
1320 warn!(%de, "BinanceSpot failed to disconnect WS after subscribe failure")
1321 }
1322 Err(_) => {
1323 warn!("BinanceSpot WS disconnect timed out (5s) after subscribe failure")
1324 }
1325 Ok(Ok(())) => {}
1326 }
1327 self.rate_limiter.after_failure(&e).await;
1328 return Err(self
1329 .rate_limiter
1330 .stream_open_error(&e, |e| UnindexedClientError::Internal(e.to_string())));
1331 }
1332 }
1333
1334 let cm_handle = tokio::spawn(connection_manager(
1346 tx,
1347 dedup,
1348 ws_handle,
1349 rest,
1350 rate_limiter,
1351 known_live,
1352 instruments,
1353 assets.to_vec(),
1354 Some(initial_ws),
1355 ));
1356
1357 let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
1361 let guarded_stream = AbortOnDropStream::new(rx_stream, cm_handle);
1362 Ok(futures::StreamExt::boxed(guarded_stream))
1363 }
1364
1365 async fn cancel_order(
1366 &self,
1367 request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
1368 ) -> UnindexedOrderResponseCancel {
1369 let instrument = request.key.instrument.clone();
1370 let key = OrderKey {
1371 exchange: request.key.exchange,
1372 instrument: instrument.clone(),
1373 strategy: request.key.strategy.clone(),
1374 cid: request.key.cid.clone(),
1375 };
1376
1377 let ws = match self.get_ws_api().await {
1378 Ok(ws) => ws,
1379 Err(unavailable) => {
1380 return UnindexedOrderResponseCancel {
1381 key,
1382 state: Err(unavailable.into_order_error()),
1383 };
1384 }
1385 };
1386
1387 let mut params_builder =
1390 OrderCancelParams::builder(request.key.instrument.name().to_string());
1391
1392 if let Some(order_id) = request.state.id.as_ref().and_then(VenueOrderId::assigned) {
1394 if let Ok(id) = order_id.0.parse::<i64>() {
1395 params_builder = params_builder.order_id(id);
1396 } else {
1397 error!(
1402 order_id = %order_id.0,
1403 "BinanceSpot cancel: exchange orderId not parseable as i64, falling back to clientOrderId"
1404 );
1405 params_builder = params_builder.orig_client_order_id(request.key.cid.0.to_string());
1406 }
1407 } else {
1408 params_builder = params_builder.orig_client_order_id(request.key.cid.0.to_string());
1409 }
1410
1411 let params = match params_builder.build() {
1412 Ok(p) => p,
1413 Err(e) => {
1414 error!(%e, "BinanceSpot failed to build cancel order params");
1415 return UnindexedOrderResponseCancel {
1416 key,
1417 state: Err(UnindexedOrderError::Rejected(ApiError::OrderRejected(
1418 e.to_string(),
1419 ))),
1420 };
1421 }
1422 };
1423
1424 let sent_ms = unix_ms();
1425 let result = ws.order_cancel(params).await;
1426 if let Ok(response) = &result {
1428 self.rate_limiter
1429 .observe_ws_api(response.rate_limits.as_deref(), sent_ms);
1430 }
1431 match result {
1432 Ok(response) => match response.data() {
1433 Ok(data) => {
1434 let time_exchange = answer_time(
1435 ExchangeId::BinanceSpot,
1436 &instrument,
1437 "cancel_order",
1438 data.transact_time,
1439 );
1440
1441 let exchange_order_id = match data.order_id {
1442 Some(id) => OrderId(format_smolstr!("{id}")),
1443 None => {
1444 return UnindexedOrderResponseCancel {
1445 key,
1446 state: Err(unreadable_answer(
1447 ExchangeId::BinanceSpot,
1448 "cancel answer has no orderId",
1449 )),
1450 };
1451 }
1452 };
1453
1454 let filled_qty = binance_filled_qty(
1455 ExchangeId::BinanceSpot,
1456 &exchange_order_id,
1457 data.executed_qty.as_deref(),
1458 );
1459
1460 self.known_live.lock().ended(&key.cid);
1461 UnindexedOrderResponseCancel {
1462 key,
1463 state: Ok(Cancelled::new(exchange_order_id, time_exchange, filled_qty)),
1464 }
1465 }
1466 Err(e) => UnindexedOrderResponseCancel {
1467 key,
1468 state: Err(unreadable_answer(ExchangeId::BinanceSpot, &e.to_string())),
1469 },
1470 },
1471 Err(e) => {
1472 if let Some(order_err) = classify_ws_order_error(&e, &instrument) {
1480 self.rate_limiter.after_failure(&e).await;
1482 if let OrderError::Rejected(api @ ApiError::RateLimit { .. }) = &order_err {
1486 self.rate_limiter.on_rate_limited(api.retry_after());
1487 }
1488 UnindexedOrderResponseCancel {
1489 key,
1490 state: Err(order_err),
1491 }
1492 } else {
1493 self.clear_ws_api().await;
1496 UnindexedOrderResponseCancel {
1497 key,
1498 state: Err(UnindexedOrderError::Connectivity(
1499 ConnectivityError::Socket(format!("{e:#}")),
1500 )),
1501 }
1502 }
1503 }
1504 }
1505 }
1506
1507 async fn open_order(
1508 &self,
1509 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
1510 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
1511 let instrument = request.key.instrument.clone();
1512 let side = request.state.side;
1513 let price = request.state.price;
1514 let quantity = request.state.quantity;
1515 let kind = request.state.kind;
1516 let time_in_force = request.state.time_in_force;
1517 let cid = request.key.cid.clone();
1518
1519 let order_key = OrderKey::new(
1520 ExchangeId::BinanceSpot,
1521 instrument.clone(),
1522 request.key.strategy.clone(),
1523 cid.clone(),
1524 );
1525
1526 if let Some(effect) = request
1528 .state
1529 .margin_effect
1530 .filter(|effect| *effect != MarginEffect::NoSideEffect)
1531 {
1532 return Order {
1533 key: order_key,
1534 side,
1535 price,
1536 quantity,
1537 kind,
1538 time_in_force,
1539 state: OrderState::inactive(OrderError::UnsupportedMarginEffect(effect)),
1540 };
1541 }
1542
1543 if let Err(refused) = self
1544 .rules
1545 .check(instrument.name(), side, kind, price, quantity)
1546 .await
1547 {
1548 return Order {
1549 key: order_key,
1550 side,
1551 price,
1552 quantity,
1553 kind,
1554 time_in_force,
1555 state: OrderState::inactive(refused),
1556 };
1557 }
1558
1559 let ws = match self.get_ws_api().await {
1560 Ok(ws) => ws,
1561 Err(unavailable) => {
1562 return Order {
1563 key: order_key,
1564 side,
1565 price,
1566 quantity,
1567 kind,
1568 time_in_force,
1569 state: OrderState::inactive(unavailable.into_open_error()),
1570 };
1571 }
1572 };
1573
1574 let binance_side = match side {
1575 Side::Buy => OrderPlaceSideEnum::Buy,
1576 Side::Sell => OrderPlaceSideEnum::Sell,
1577 };
1578
1579 let fields = match SpotOrderFields::new(kind, time_in_force, price) {
1580 Ok(fields) => fields,
1581 Err(refused) => {
1582 return Order {
1583 key: order_key,
1584 side,
1585 price,
1586 quantity,
1587 kind,
1588 time_in_force,
1589 state: OrderState::inactive(refused),
1590 };
1591 }
1592 };
1593 let (binance_type, binance_tif) = place_enums(fields.order_type, fields.time_in_force);
1594
1595 let mut params_builder =
1602 OrderPlaceParams::builder(instrument.name().to_string(), binance_side, binance_type)
1603 .quantity(quantity)
1604 .new_client_order_id(cid.0.to_string());
1605 if let Some(price) = fields.price {
1606 params_builder = params_builder.price(price);
1607 }
1608 if let Some(stop_price) = fields.stop_price {
1609 params_builder = params_builder.stop_price(stop_price);
1610 }
1611 if let Some(trailing_delta) = fields.trailing_delta {
1612 params_builder = params_builder.trailing_delta(trailing_delta);
1613 }
1614 if let Some(tif) = binance_tif {
1615 params_builder = params_builder.time_in_force(tif);
1616 }
1617
1618 let params = match params_builder.build() {
1619 Ok(p) => p,
1620 Err(e) => {
1621 error!(%e, "BinanceSpot failed to build new order params");
1622 return Order {
1623 key: order_key,
1624 side,
1625 price,
1626 quantity,
1627 kind,
1628 time_in_force,
1629 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
1630 e.to_string(),
1631 ))),
1632 };
1633 }
1634 };
1635
1636 let sent_ms = unix_ms();
1637 let result = ws.order_place(params).await;
1638 if let Ok(response) = &result {
1640 self.rate_limiter
1641 .observe_ws_api(response.rate_limits.as_deref(), sent_ms);
1642 }
1643 match result {
1644 Ok(response) => match response.data() {
1645 Ok(data) => {
1646 let time_exchange = answer_time(
1647 ExchangeId::BinanceSpot,
1648 &instrument,
1649 "open_order",
1650 data.transact_time,
1651 );
1652
1653 let exchange_order_id = match data.order_id {
1654 Some(id) => OrderId(format_smolstr!("{id}")),
1655 None => {
1656 return Order {
1657 key: order_key,
1658 side,
1659 price,
1660 quantity,
1661 kind,
1662 time_in_force,
1663 state: OrderState::inactive(unreadable_answer(
1664 ExchangeId::BinanceSpot,
1665 "open_order answer has no orderId",
1666 )),
1667 };
1668 }
1669 };
1670
1671 let state = placed_order_state(
1676 ExchangeId::BinanceSpot,
1677 &instrument,
1678 quantity,
1679 PlacementResponse {
1680 status: data.status.as_deref(),
1681 order_id: exchange_order_id,
1682 time_exchange,
1683 executed_qty: data.executed_qty.as_deref(),
1684 cumulative_quote_qty: data.cummulative_quote_qty.as_deref(),
1685 },
1686 );
1687 self.known_live
1688 .lock()
1689 .placed(&order_key, quantity, kind, &state);
1690
1691 Order {
1692 key: order_key,
1693 side,
1694 price,
1695 quantity,
1696 kind,
1697 time_in_force,
1698 state,
1699 }
1700 }
1701 Err(e) => Order {
1702 key: order_key,
1703 side,
1704 price,
1705 quantity,
1706 kind,
1707 time_in_force,
1708 state: OrderState::inactive(unreadable_answer(
1709 ExchangeId::BinanceSpot,
1710 &e.to_string(),
1711 )),
1712 },
1713 },
1714 Err(e) => {
1715 if let Some(order_err) = classify_ws_order_error(&e, &instrument) {
1723 self.rate_limiter.after_failure(&e).await;
1725 if let OrderError::Rejected(api @ ApiError::RateLimit { .. }) = &order_err {
1729 self.rate_limiter.on_rate_limited(api.retry_after());
1730 }
1731 Order {
1732 key: order_key,
1733 side,
1734 price,
1735 quantity,
1736 kind,
1737 time_in_force,
1738 state: OrderState::inactive(order_err),
1739 }
1740 } else {
1741 self.clear_ws_api().await;
1744 Order {
1745 key: order_key,
1746 side,
1747 price,
1748 quantity,
1749 kind,
1750 time_in_force,
1751 state: OrderState::inactive(OrderError::Connectivity(
1752 ConnectivityError::Socket(format!("{e:#}")),
1753 )),
1754 }
1755 }
1756 }
1757 }
1758 }
1759
1760 fn cancel_replace(
1779 &self,
1780 cancel: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
1781 open: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
1782 on_cancel_failure: OnCancelFailure,
1783 ) -> Option<impl Future<Output = CancelReplaceAnswer> + Send> {
1784 if cancel.key.instrument != open.key.instrument
1785 || open
1786 .state
1787 .margin_effect
1788 .is_some_and(|effect| effect != MarginEffect::NoSideEffect)
1789 {
1790 return None;
1791 }
1792 let fields =
1793 SpotOrderFields::new(open.state.kind, open.state.time_in_force, open.state.price)
1794 .ok()?;
1795 let target = CancelTarget::new(cancel.state.id.as_ref(), &cancel.key.cid)?;
1796 Some(self.send_cancel_replace(cancel, open, target, fields, on_cancel_failure))
1797 }
1798
1799 async fn fetch_balances(
1805 &self,
1806 assets: &[AssetNameExchange],
1807 ) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
1808 let (response, sent) =
1809 rest_read_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1810 Box::pin(async move {
1811 let params = GetAccountParams::builder().build()?;
1812 rest.get_account(params).await
1813 })
1814 })
1815 .await
1816 .map_err(|e| classify_rest_query_error(&e, None))?;
1817
1818 let account = response.data().await.map_err(response_decode_error)?;
1819
1820 Ok(filter_and_convert_balances(
1821 account.balances.unwrap_or_default(),
1822 assets,
1823 self.rate_limiter.balance_read_time(sent),
1824 ))
1825 }
1826
1827 async fn fetch_open_orders(
1832 &self,
1833 instruments: &[InstrumentNameExchange],
1834 ) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
1835 if instruments.is_empty() {
1838 let orders =
1839 fetch_all_open_orders(self.rest.clone(), self.rate_limiter.clone()).await?;
1840 self.remember_live(&orders);
1841 return Ok(orders);
1842 }
1843 use futures::{StreamExt as _, TryStreamExt as _};
1848 futures::stream::iter(instruments.iter().cloned().map(|instrument| {
1849 fetch_open_orders_for_instrument(
1850 self.rest.clone(),
1851 self.rate_limiter.clone(),
1852 instrument,
1853 RequestKind::Query,
1854 )
1855 }))
1856 .buffer_unordered(8)
1857 .try_fold(Vec::with_capacity(instruments.len()), |mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, (_, listing)| async move {
1858 acc.extend(listing.orders);
1859 Ok(acc)
1860 })
1861 .await
1862 .inspect(|orders| self.remember_live(orders))
1863 }
1864
1865 #[allow(clippy::redundant_iter_cloned)]
1876 async fn fetch_trades(
1877 &self,
1878 start: DateTime<Utc>,
1879 end: DateTime<Utc>,
1880 instruments: &[InstrumentNameExchange],
1881 ) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
1882 use futures::StreamExt;
1883
1884 if instruments.is_empty() {
1885 debug!(
1886 "BinanceSpot fetch_trades called with empty instruments slice — returning empty result"
1887 );
1888 return Ok(TradesRead::complete(Vec::new()));
1889 }
1890 if start > end {
1891 return Ok(TradesRead::complete(Vec::new()));
1892 }
1893 let span = MyTradesFrom::Span {
1895 start: start.timestamp_millis(),
1896 end: end.timestamp_millis(),
1897 };
1898 let mut all_trades = Vec::new();
1901
1902 let mut stream = futures::stream::iter(instruments.iter().cloned().map(|inst| {
1906 let rest = self.rest.clone();
1907 let rate_limiter = self.rate_limiter.clone();
1908 async move {
1909 let pages =
1910 paginate_my_trades(&rest, &rate_limiter, &inst, span, RequestKind::Query)
1911 .await?;
1912 Ok::<_, UnindexedClientError>((inst, pages))
1913 }
1914 }))
1915 .buffer_unordered(8);
1916 while let Some(result) = stream.next().await {
1917 let (instrument, trades_data) = result?;
1918 for t in trades_data {
1919 if let Some(trade) = convert_my_trade(&t, &instrument)
1920 && (start..=end).contains(&trade.time_exchange)
1921 {
1922 all_trades.push(trade);
1923 }
1924 }
1925 }
1926
1927 Ok(TradesRead::complete(all_trades))
1928 }
1929}
1930
1931impl OrderStatusClient for BinanceSpot {
1946 async fn fetch_ended_orders(
1947 &self,
1948 orders: &[UnindexedOrderKey],
1949 ) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError> {
1950 fetch_ended_by_key(orders, |key| {
1951 fetch_order_lookup(
1952 self.rest.clone(),
1953 self.rate_limiter.clone(),
1954 key,
1955 RequestKind::Query,
1956 )
1957 })
1958 .await
1959 }
1960}
1961
1962const ACCOUNT_STREAM_WS: &str = "BinanceSpot account stream";
1964
1965async fn connect_and_subscribe(
1969 ws_handle: &WebsocketApiHandle,
1970) -> anyhow::Result<DisconnectOnDrop<WebsocketApi>> {
1971 let ws = DisconnectOnDrop::new(ws_handle.connect().await?, ACCOUNT_STREAM_WS);
1972 #[allow(clippy::expect_used)] let params = UserDataStreamSubscribeSignatureParams::builder()
1974 .build()
1975 .expect("UserDataStreamSubscribeSignatureParams has no required fields");
1976 match ws.user_data_stream_subscribe_signature(params).await {
1977 Ok(_) => Ok(ws),
1978 Err(e) => {
1979 warn!(%e, "BinanceSpot WS subscribe failed, cleaning up connection");
1982 Err(e)
1983 }
1984 }
1985}
1986
1987#[allow(clippy::cognitive_complexity)]
2011async fn connection_manager(
2012 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
2013 dedup: SharedDedupCache,
2014 ws_handle: WebsocketApiHandle,
2015 rest: Arc<RestApi>,
2016 rate_limiter: Arc<RateLimitTracker>,
2017 known: SharedKnownLiveOrders,
2018 instruments: Vec<InstrumentNameExchange>,
2019 assets: Vec<AssetNameExchange>,
2021 initial_ws: Option<DisconnectOnDrop<WebsocketApi>>,
2022) {
2023 let mut budget = ReconnectBudget::new("Binance", BackoffLimits::ACCOUNT_STREAM);
2024 let mut disconnect: Option<Disconnect> = None;
2025 let mut unrecovered = UnrecoveredFills::default();
2027 let mut unchecked = UncheckedOrders::default();
2029 let mut restatement = BalanceRestatement::default();
2031 let clock = StreamClock::shared(rate_limiter.signing_clock());
2033 let mut current_ws = initial_ws;
2034
2035 loop {
2036 let ws = match current_ws.take() {
2038 Some(ws) => ws,
2039 None => {
2040 if let Err(reason) = rate_limiter.before_stream_connect().await {
2043 emit_stream_terminated(&tx, ExchangeId::BinanceSpot, reason);
2044 break;
2045 }
2046 match connect_and_subscribe(&ws_handle).await {
2047 Ok(ws) => ws,
2053 Err(e) => {
2054 error!(%e, "BinanceSpot WS connect/subscribe failed");
2055 rate_limiter.on_stream_refused(&e);
2057 rate_limiter.after_failure(&e).await;
2059 if let Err(reason) =
2060 retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await
2061 {
2062 emit_stream_terminated(&tx, ExchangeId::BinanceSpot, reason);
2064 break;
2065 }
2066 continue;
2067 }
2068 }
2069 }
2070 };
2071 info!("BinanceSpot account_stream connected and subscribed");
2072
2073 let (signal_tx, signal_rx) = oneshot::channel::<()>();
2079 let mut signal_tx_opt = Some(signal_tx);
2080 let heartbeat_flag = Arc::new(AtomicBool::new(false));
2084 let hb_callback = heartbeat_flag.clone();
2085 let dedup_callback = dedup.clone();
2086 let known_callback = known.clone();
2087 let clock_callback = clock.clone();
2088 let mut event_tx = Some(tx.clone());
2092 let mut event_buf = Vec::with_capacity(32);
2096
2097 let subscription = ws.subscribe_on_ws_events(move |event| {
2104 let Some(ref sender) = event_tx else { return };
2105 match event {
2106 WebsocketEvent::Message(json_str) => {
2107 hb_callback.store(true, Ordering::Release);
2110 let outcome = convert_user_data_events(&json_str, &mut event_buf);
2114 StreamClock::observe_frame(
2115 &clock_callback,
2116 outcome.event_time,
2117 &event_buf,
2118 tokio::time::Instant::now(),
2119 );
2120 let stream_terminated = outcome.terminated;
2121 for ev in event_buf.drain(..) {
2122 if let Some(key) = dedup_key_from_event(&ev)
2124 && is_duplicate(&dedup_callback, key)
2125 {
2126 trace!("BinanceSpot dedup: skipping duplicate event");
2127 continue;
2128 }
2129 let known = KnownLiveOrders::observes(&ev.kind).then(|| {
2133 let mut known = known_callback.lock();
2134 known.observe(&ev);
2135 known
2136 });
2137 if sender.send(ev).is_err() {
2138 drop(known);
2139 warn!("BinanceSpot account_stream receiver dropped, suppressing further sends");
2140 event_tx.take();
2141 if let Some(s) = signal_tx_opt.take() {
2142 let _ = s.send(());
2143 }
2144 return;
2145 }
2146 }
2147 if stream_terminated {
2150 event_tx.take();
2151 if let Some(s) = signal_tx_opt.take() {
2152 let _ = s.send(());
2153 }
2154 }
2155 }
2156 WebsocketEvent::Ping | WebsocketEvent::Pong => {
2157 hb_callback.store(true, Ordering::Release);
2160 }
2161 WebsocketEvent::Error(e) => {
2162 warn!(%e, "BinanceSpot WebSocket error, will attempt reconnect");
2166 event_tx.take();
2167 if let Some(s) = signal_tx_opt.take() {
2168 let _ = s.send(());
2169 }
2170 }
2171 WebsocketEvent::Close(code, reason) => {
2172 warn!(code, %reason, "BinanceSpot WebSocket closed");
2173 event_tx.take();
2174 if let Some(s) = signal_tx_opt.take() {
2175 let _ = s.send(());
2176 }
2177 }
2178 _ => {}
2179 }
2180 });
2181 let subscription = UnsubscribeOnDrop::new(subscription);
2183
2184 if let Some(disconnect) = disconnect.take() {
2191 let end = disconnect.end(&clock.lock().venue(), HostMoment::now());
2192 unrecovered.open(&instruments, disconnect.start, end);
2193 let held = known.lock().instruments_among(&instruments);
2195 unchecked.open(held);
2196 restatement.open();
2197 }
2198 recover_fills(&rest, &rate_limiter, &mut unrecovered, &tx, &dedup, &known).await;
2199 recover_spot_ended_orders(
2200 &rest,
2201 &rate_limiter,
2202 &known,
2203 &mut unchecked,
2204 &unrecovered,
2205 &tx,
2206 )
2207 .await;
2208 enum DisconnectReason {
2214 Signal,
2215 HeartbeatTimeout,
2216 ConsumerDropped,
2217 }
2218 let reason = {
2219 let mut signal_rx = signal_rx;
2220 let retry_gaps = async {
2226 loop {
2227 restate("BinanceSpot", &clock, &mut restatement, &tx, |floor| {
2229 read_spot_restatement(&rest, &rate_limiter, &assets, &clock, floor)
2230 })
2231 .await;
2232 let next_check = unchecked.next_due(&unrecovered);
2233 match unrecovered
2234 .next_due()
2235 .into_iter()
2236 .chain(next_check)
2237 .chain(restatement.next_due())
2238 .min()
2239 {
2240 Some(due) => tokio::time::sleep_until(due).await,
2241 None => std::future::pending::<()>().await,
2242 }
2243 recover_fills(&rest, &rate_limiter, &mut unrecovered, &tx, &dedup, &known)
2244 .await;
2245 recover_spot_ended_orders(
2246 &rest,
2247 &rate_limiter,
2248 &known,
2249 &mut unchecked,
2250 &unrecovered,
2251 &tx,
2252 )
2253 .await;
2254 }
2255 };
2256 tokio::pin!(retry_gaps);
2257 let mut grace = true;
2259 loop {
2260 tokio::select! {
2261 biased;
2263 _ = tx.closed() => {
2264 debug!("BinanceSpot account_stream consumer dropped, terminating");
2265 break DisconnectReason::ConsumerDropped;
2266 }
2267 _ = &mut signal_rx => {
2268 warn!("BinanceSpot WS disconnected, will attempt reconnect");
2269 break DisconnectReason::Signal;
2270 }
2271 _ = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
2272 let in_grace = std::mem::take(&mut grace);
2279 if heartbeat_flag.swap(false, Ordering::AcqRel) {
2280 budget.reset();
2284 continue;
2285 }
2286 if in_grace {
2287 continue;
2290 }
2291 warn!("BinanceSpot heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
2292 break DisconnectReason::HeartbeatTimeout;
2293 }
2294 () = &mut retry_gaps => {}
2296 }
2297 }
2298 };
2299 let should_reconnect = !matches!(reason, DisconnectReason::ConsumerDropped);
2300
2301 if should_reconnect {
2308 let heartbeat_timed_out = matches!(reason, DisconnectReason::HeartbeatTimeout);
2309 disconnect = Some(Disconnect::seen(
2310 &clock.lock().venue(),
2311 HostMoment::now(),
2312 disconnect_lookback(heartbeat_timed_out),
2313 ));
2314 }
2315
2316 subscription.unsubscribe();
2326 if let Err(e) = ws.disconnect().await {
2329 warn!(%e, "BinanceSpot failed to disconnect WebSocket");
2330 }
2331
2332 if !should_reconnect || tx.is_closed() {
2333 debug!("BinanceSpot connection manager exiting");
2336 break;
2337 }
2338 let heartbeat_timed_out = matches!(reason, DisconnectReason::HeartbeatTimeout);
2341 let last_error = || disconnect_error(heartbeat_timed_out);
2342 if let Err(reason) = retry_or_end(&mut budget, &rate_limiter, last_error).await {
2343 emit_stream_terminated(&tx, ExchangeId::BinanceSpot, reason);
2344 break;
2345 }
2346 }
2347}
2348
2349async fn read_spot_restatement(
2353 rest: &Arc<RestApi>,
2354 rate_limiter: &Arc<RateLimitTracker>,
2355 assets: &[AssetNameExchange],
2356 clock: &SharedStreamClock,
2357 floor: Option<DateTime<Utc>>,
2358) -> Result<Vec<UnindexedAccountEvent>, UnindexedClientError> {
2359 let (response, sent) = rest_read_with_retry(rest, rate_limiter, RequestKind::Query, |rest| {
2360 Box::pin(async move {
2361 let params = GetAccountParams::builder().build()?;
2362 rest.get_account(params).await
2363 })
2364 })
2365 .await
2366 .map_err(|e| classify_rest_query_error(&e, None))?;
2367 let account = response.data().await.map_err(response_decode_error)?;
2368 let time = restatement_time(&clock.lock(), sent, floor);
2369 Ok(
2370 filter_and_convert_balances(account.balances.unwrap_or_default(), assets, time)
2371 .into_iter()
2372 .map(|balance| {
2373 UnindexedAccountEvent::new(
2374 ExchangeId::BinanceSpot,
2375 AccountEventKind::BalanceSnapshot(
2376 rustrade_integration::collection::snapshot::Snapshot(balance),
2377 ),
2378 )
2379 })
2380 .collect(),
2381 )
2382}
2383
2384async fn recover_fills(
2403 rest: &Arc<RestApi>,
2404 rate_limiter: &Arc<RateLimitTracker>,
2405 unrecovered: &mut UnrecoveredFills,
2406 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
2407 dedup: &SharedDedupCache,
2408 known: &SharedKnownLiveOrders,
2409) {
2410 use futures::StreamExt;
2411
2412 let due = unrecovered.due(tokio::time::Instant::now());
2413 let Some(oldest) = due.iter().map(|(_, gap)| gap.start_ms).min() else {
2414 return;
2415 };
2416 info!(
2417 gaps = due.len(),
2418 oldest = %gap_time(oldest),
2419 "BinanceSpot recovering fills missed while disconnected"
2420 );
2421
2422 let mut settled = vec![false; due.len()];
2425 let started = AtomicUsize::new(0);
2426 let recovery = async {
2427 let order_executions_deadline = tokio::time::Instant::now() + ORDER_EXECUTIONS_BUDGET;
2428 let mut recovered = 0u32;
2429 let mut duplicates = 0u32;
2430
2431 let mut stream =
2437 futures::stream::iter(due.iter().cloned().enumerate().map(|(index, (inst, gap))| {
2438 started.store(index + 1, Ordering::Relaxed);
2439 let rest = rest.clone();
2440 let rl = rate_limiter.clone();
2441 async move {
2442 let span = MyTradesFrom::Span {
2443 start: gap.start_ms,
2444 end: gap.end_ms,
2445 };
2446 let raw =
2447 match paginate_my_trades(&rest, &rl, &inst, span, RequestKind::Essential)
2448 .await
2449 {
2450 Ok(pages) => pages,
2451 Err(e) => return (index, Err(e)),
2452 };
2453 let totals = recovered_order_totals(
2457 ExchangeId::BinanceSpot,
2458 &inst,
2459 &raw,
2460 order_executions_deadline,
2461 |order_id| {
2462 paginate_my_trades(
2463 &rest,
2464 &rl,
2465 &inst,
2466 MyTradesFrom::Order(order_id),
2467 RequestKind::Essential,
2468 )
2469 },
2470 )
2471 .await;
2472 let cids = recovered_order_cids(
2475 ExchangeId::BinanceSpot,
2476 &inst,
2477 &raw,
2478 order_executions_deadline,
2479 |order_id| fetch_order_cid(&rest, &rl, &inst, order_id),
2480 )
2481 .await;
2482 let trades: Vec<_> = raw
2483 .iter()
2484 .filter_map(|t| {
2485 let mut trade = convert_my_trade(t, &inst)?;
2486 trade.order_filled_quantity =
2487 t.id.and_then(|id| totals.get(&id).copied());
2488 trade.client_order_id =
2489 t.order_id.and_then(|id| cids.get(&id).cloned());
2490 Some(trade)
2491 })
2492 .collect();
2493 (index, Ok(trades))
2494 }
2495 }))
2496 .buffer_unordered(8);
2497 while let Some((index, result)) = stream.next().await {
2498 let (inst, gap) = &due[index];
2499 settled[index] = true;
2500 let trades = match result {
2501 Ok(trades) => trades,
2502 Err(e) => {
2503 gap_failed(
2504 unrecovered,
2505 tx,
2506 ExchangeId::BinanceSpot,
2507 inst,
2508 gap,
2509 FillRecoveryFailure::Request(e.to_string()),
2510 );
2511 continue;
2512 }
2513 };
2514 for trade in trades {
2515 let event = UnindexedAccountEvent::new(
2518 ExchangeId::BinanceSpot,
2519 AccountEventKind::Trade(trade),
2520 );
2521 if let Some(key) = dedup_key_from_event(&event)
2524 && is_duplicate(dedup, key)
2525 {
2526 duplicates += 1;
2527 continue;
2528 }
2529 let sent = {
2532 let mut known = known.lock();
2533 known.observe(&event);
2534 tx.send(event)
2535 };
2536 if sent.is_err() {
2537 debug!("BinanceSpot fill recovery: consumer dropped during recovery");
2540 return;
2541 }
2542 recovered += 1;
2543 }
2544 unrecovered.recovered(inst, gap);
2545 }
2546 info!(recovered, duplicates, "BinanceSpot fill recovery complete");
2547 };
2548 if tokio::time::timeout(Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS), recovery)
2552 .await
2553 .is_err()
2554 {
2555 let started = started.load(Ordering::Relaxed);
2556 for ((inst, gap), _) in due[..started]
2557 .iter()
2558 .zip(&settled)
2559 .filter(|(_, settled)| !**settled)
2560 {
2561 gap_failed(
2562 unrecovered,
2563 tx,
2564 ExchangeId::BinanceSpot,
2565 inst,
2566 gap,
2567 FillRecoveryFailure::TimedOut {
2568 timeout_secs: FILL_RECOVERY_TIMEOUT_SECS,
2569 },
2570 );
2571 }
2572 }
2573}
2574
2575fn convert_balance_entry(
2589 b: binance_sdk::spot::rest_api::GetAccountResponseBalancesInner,
2590 now: chrono::DateTime<Utc>,
2591) -> Option<AssetBalance<AssetNameExchange>> {
2592 let asset_name = AssetNameExchange::new(b.asset.as_deref()?);
2593 let free = match b.free.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2594 Some(v) => v,
2595 None => {
2596 warn!(%asset_name, "BinanceSpot balance missing/unparseable 'free' field");
2597 return None;
2598 }
2599 };
2600 let locked = match b.locked.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2601 Some(v) => v,
2602 None => {
2603 warn!(%asset_name, "BinanceSpot balance missing/unparseable 'locked' field");
2604 return None;
2605 }
2606 };
2607 Some(AssetBalance::new(
2608 asset_name,
2609 Balance::new(free + locked, free),
2610 now,
2611 ))
2612}
2613
2614fn filter_and_convert_balances(
2617 balances: Vec<binance_sdk::spot::rest_api::GetAccountResponseBalancesInner>,
2618 assets: &[AssetNameExchange],
2619 time: chrono::DateTime<Utc>,
2620) -> Vec<AssetBalance<AssetNameExchange>> {
2621 if assets.is_empty() {
2623 return balances
2624 .into_iter()
2625 .filter_map(|b| convert_balance_entry(b, time))
2626 .collect();
2627 }
2628 if assets.len() <= 16 {
2631 return balances
2632 .into_iter()
2633 .filter_map(|b| {
2634 let asset_name_str = b.asset.as_deref()?;
2635 if !assets.iter().any(|a| a.name().as_str() == asset_name_str) {
2636 return None;
2637 }
2638 convert_balance_entry(b, time)
2639 })
2640 .collect();
2641 }
2642 use std::collections::HashSet;
2643 let asset_set: HashSet<&str> = assets.iter().map(|a| a.name().as_str()).collect();
2644 balances
2645 .into_iter()
2646 .filter_map(|b| {
2647 let asset_name_str = b.asset.as_deref()?;
2648 if !asset_set.contains(asset_name_str) {
2649 return None;
2650 }
2651 convert_balance_entry(b, time)
2652 })
2653 .collect()
2654}
2655
2656fn convert_my_trade(
2658 t: &binance_sdk::spot::rest_api::MyTradesResponseInner,
2659 instrument: &InstrumentNameExchange,
2660) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
2661 let trade_id_raw = match t.id {
2662 Some(id) => id,
2663 None => {
2664 warn!(%instrument, "BinanceSpot trade missing id");
2665 return None;
2666 }
2667 };
2668 let trade_id = TradeId(format_smolstr!("{}", trade_id_raw));
2669 let order_id = match t.order_id {
2670 Some(id) => OrderId(format_smolstr!("{}", id)),
2671 None => {
2672 warn!(%instrument, trade_id = %trade_id_raw, "BinanceSpot trade missing orderId");
2673 return None;
2674 }
2675 };
2676 let side = match t.is_buyer {
2677 Some(is_buyer) => {
2678 if is_buyer {
2679 Side::Buy
2680 } else {
2681 Side::Sell
2682 }
2683 }
2684 None => {
2685 warn!(%instrument, trade_id = %trade_id_raw, "BinanceSpot trade missing isBuyer");
2686 return None;
2687 }
2688 };
2689 let price = match t.price.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2690 Some(v) => v,
2691 None => {
2692 warn!(%instrument, trade_id = %trade_id_raw, "BinanceSpot trade missing/unparseable price");
2693 return None;
2694 }
2695 };
2696 let quantity = match t.qty.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2697 Some(v) => v,
2698 None => {
2699 warn!(%instrument, trade_id = %trade_id_raw, "BinanceSpot trade missing/unparseable qty");
2700 return None;
2701 }
2702 };
2703 let commission = t
2704 .commission
2705 .as_deref()
2706 .and_then(|s| Decimal::from_str(s).ok())
2707 .unwrap_or(Decimal::ZERO);
2708 let (time_exchange, time_source) = match t
2711 .time
2712 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
2713 {
2714 Some(ts) => (ts, TimeSource::Venue),
2715 None => {
2716 warn!(%instrument, trade_id = %trade_id_raw, "BinanceSpot trade missing/unparseable time, using now");
2717 (Utc::now(), TimeSource::Host)
2718 }
2719 };
2720
2721 let fee_asset = t
2727 .commission_asset
2728 .as_deref()
2729 .map(AssetNameExchange::from)
2730 .unwrap_or_else(|| AssetNameExchange::from("UNKNOWN"));
2731
2732 Some(
2733 Trade::new(
2734 trade_id,
2735 order_id,
2736 instrument.clone(),
2737 StrategyId::unknown(), time_exchange,
2739 side,
2740 price,
2741 quantity,
2742 None,
2744 AssetFees::new(fee_asset, commission, None),
2745 )
2746 .with_time_source(time_source),
2747 )
2748}
2749
2750fn convert_user_data_events(frame: &str, buf: &mut Vec<UnindexedAccountEvent>) -> UserDataOutcome {
2776 let (event_type, event, event_time) = match parse_user_data_frame(frame) {
2777 UserDataFrame::Response => return UserDataOutcome::default(),
2778 UserDataFrame::Event {
2779 event_type,
2780 event,
2781 event_time,
2782 ..
2783 } => (event_type, event, event_time),
2784 UserDataFrame::Unrecognised => {
2785 static SEEN: AtomicU64 = AtomicU64::new(0);
2786 log_unrecognised_frame("BinanceSpot", &SEEN, frame);
2787 return UserDataOutcome::default();
2788 }
2789 };
2790 UserDataOutcome {
2791 terminated: convert_user_data_event(event_type, event, buf),
2792 event_time,
2793 }
2794}
2795
2796#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
2798struct UserDataOutcome {
2799 terminated: bool,
2801 event_time: Option<DateTime<Utc>>,
2803}
2804
2805fn convert_user_data_event(
2808 event_type: &str,
2809 event: &str,
2810 buf: &mut Vec<UnindexedAccountEvent>,
2811) -> bool {
2812 match event_type {
2814 "executionReport" => {
2815 match serde_json::from_str::<ExecutionReport>(event) {
2818 Ok(report) => {
2819 convert_execution_report(&report, ExchangeId::BinanceSpot, buf);
2820 }
2821 Err(e) => {
2822 warn!(error = %e, frame = frame_excerpt(event), "BinanceSpot: undeserializable executionReport, dropping")
2823 }
2824 }
2825 false
2826 }
2827 "outboundAccountPosition" => {
2828 match serde_json::from_str::<binance_sdk::spot::websocket_api::OutboundAccountPosition>(
2829 event,
2830 ) {
2831 Ok(position) => convert_account_position(position, buf),
2832 Err(e) => {
2833 warn!(error = %e, frame = frame_excerpt(event), "BinanceSpot: undeserializable outboundAccountPosition, dropping")
2834 }
2835 }
2836 false
2837 }
2838 "balanceUpdate" => {
2839 false
2846 }
2847 "eventStreamTerminated" => {
2848 warn!("BinanceSpot user data stream terminated by exchange, signalling reconnect");
2852 true
2853 }
2854 "listStatus" | "externalLockUpdate" => {
2855 trace!(event_type, "BinanceSpot ignoring unhandled user data event");
2859 false
2860 }
2861 _ => {
2862 static SEEN: UnhandledEvents = UnhandledEvents::new();
2863 log_unhandled_event("BinanceSpot", &SEEN, event_type, event);
2864 false
2865 }
2866 }
2867}
2868
2869fn convert_account_position(
2874 position: binance_sdk::spot::websocket_api::OutboundAccountPosition,
2875 buf: &mut Vec<UnindexedAccountEvent>,
2876) {
2877 let Some(time_exchange) =
2880 account_position_time(ExchangeId::BinanceSpot, position.u, position.e_uppercase)
2881 else {
2882 return;
2883 };
2884
2885 for b in position.b_uppercase.unwrap_or_default() {
2886 let asset = match b.a {
2887 Some(a) => AssetNameExchange::new(a),
2888 None => {
2889 warn!("BinanceSpot account position entry missing asset name");
2890 continue;
2891 }
2892 };
2893 let free = match b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2894 Some(v) => v,
2895 None => {
2896 warn!(%asset, "BinanceSpot account position missing/unparseable 'free' field");
2897 continue;
2898 }
2899 };
2900 let locked = match b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2901 Some(v) => v,
2902 None => {
2903 warn!(%asset, "BinanceSpot account position missing/unparseable 'locked' field");
2904 continue;
2905 }
2906 };
2907 let update =
2912 AssetBalanceUpdate::new(asset, BalanceUpdate::new(free, locked), time_exchange);
2913 buf.push(UnindexedAccountEvent::new(
2914 ExchangeId::BinanceSpot,
2915 AccountEventKind::BalanceStreamUpdate(
2916 rustrade_integration::collection::snapshot::Snapshot::new(update),
2917 ),
2918 ));
2919 }
2920}
2921
2922#[cfg(test)]
2953fn convert_order_kind_tif(
2954 kind: OrderKind,
2955 tif: TimeInForce,
2956) -> Option<(OrderPlaceTypeEnum, Option<OrderPlaceTimeInForceEnum>)> {
2957 let price = Some(Decimal::ONE);
2961 let fields = SpotOrderFields::new(kind, tif, price).ok()?;
2962 Some(place_enums(fields.order_type, fields.time_in_force))
2963}
2964
2965#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2968pub(super) struct SpotOrderFields {
2969 pub(super) order_type: BinanceOrderType,
2970 pub(super) time_in_force: Option<BinanceTimeInForce>,
2971 pub(super) price: Option<Decimal>,
2972 pub(super) stop_price: Option<Decimal>,
2973 pub(super) trailing_delta: Option<i32>,
2974}
2975
2976impl SpotOrderFields {
2977 pub(super) fn new(
2983 kind: OrderKind,
2984 time_in_force: TimeInForce,
2985 price: Option<Decimal>,
2986 ) -> Result<Self, UnindexedOrderError> {
2987 let Some((order_type, time_in_force)) = classify_order_kind_tif(kind, time_in_force) else {
2988 return Err(OrderError::UnsupportedOrderType(format!(
2989 "Binance Spot does not yet support OrderKind::{kind:?}"
2990 )));
2991 };
2992 let mut fields = Self {
2993 order_type,
2994 time_in_force,
2995 price: None,
2996 stop_price: None,
2997 trailing_delta: None,
2998 };
2999
3000 if let Some(trailing) =
3002 trailing_params(&kind, price).map_err(|refused| refused.into_order_error())?
3003 {
3004 fields.trailing_delta = Some(trailing.trailing_delta);
3005 fields.price = trailing.price;
3006 fields.stop_price = trailing.stop_price;
3008 }
3009
3010 match kind {
3013 OrderKind::Limit => fields.price = price,
3014 OrderKind::Stop { trigger_price } | OrderKind::TakeProfit { trigger_price } => {
3015 fields.stop_price = Some(trigger_price);
3016 }
3017 OrderKind::StopLimit { trigger_price }
3018 | OrderKind::TakeProfitLimit { trigger_price } => {
3019 fields.price = price;
3020 fields.stop_price = Some(trigger_price);
3021 }
3022 OrderKind::TrailingStop { .. }
3025 | OrderKind::TrailingStopFixedLimit { .. }
3026 | OrderKind::Market
3027 | OrderKind::TrailingStopLimit { .. }
3028 | OrderKind::ConditionalUnresolved { .. } => {}
3029 }
3030 Ok(fields)
3031 }
3032}
3033
3034fn place_enums(
3036 binance_type: BinanceOrderType,
3037 binance_tif: Option<BinanceTimeInForce>,
3038) -> (OrderPlaceTypeEnum, Option<OrderPlaceTimeInForceEnum>) {
3039 let type_enum = match binance_type {
3040 BinanceOrderType::Market => OrderPlaceTypeEnum::Market,
3041 BinanceOrderType::Limit => OrderPlaceTypeEnum::Limit,
3042 BinanceOrderType::LimitMaker => OrderPlaceTypeEnum::LimitMaker,
3043 BinanceOrderType::StopLoss => OrderPlaceTypeEnum::StopLoss,
3044 BinanceOrderType::StopLossLimit => OrderPlaceTypeEnum::StopLossLimit,
3045 BinanceOrderType::TakeProfit => OrderPlaceTypeEnum::TakeProfit,
3046 BinanceOrderType::TakeProfitLimit => OrderPlaceTypeEnum::TakeProfitLimit,
3047 };
3048 let tif_enum = binance_tif.map(|t| match t {
3049 BinanceTimeInForce::Gtc => OrderPlaceTimeInForceEnum::Gtc,
3050 BinanceTimeInForce::Ioc => OrderPlaceTimeInForceEnum::Ioc,
3051 BinanceTimeInForce::Fok => OrderPlaceTimeInForceEnum::Fok,
3052 });
3053 (type_enum, tif_enum)
3054}
3055
3056#[cfg(test)]
3057#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
3059 use super::*;
3060 use crate::client::binance::shared::*;
3061 use crate::client::dedup::{DedupEventKind, DedupKey};
3062 use crate::error::StreamTerminationReason;
3063 use crate::order::{TrailingOffsetType, id::ClientOrderId};
3064 use binance_sdk::common::errors::WebsocketError;
3065 use smol_str::SmolStr;
3066
3067 #[test]
3070 fn an_unconnected_session_fails_an_open_as_not_placed() {
3071 let open = WsApiUnavailable::Connect("refused".into()).into_open_error();
3072 assert!(matches!(
3073 &open,
3074 OrderError::NotPlaced(NotPlacedReason::NotSent(ConnectivityError::Socket(m))) if m == "refused"
3075 ));
3076 assert!(!open.is_outcome_unknown());
3077
3078 let cancel = WsApiUnavailable::Connect("refused".into()).into_order_error();
3079 assert!(matches!(cancel, OrderError::Connectivity(_)));
3080
3081 let limited = || WsApiUnavailable::RateLimited {
3082 msg: "banned".into(),
3083 retry_after: None,
3084 };
3085 assert_eq!(limited().into_open_error(), limited().into_order_error());
3086 assert!(matches!(
3087 limited().into_open_error(),
3088 OrderError::Rejected(ApiError::RateLimit { .. })
3089 ));
3090 }
3091
3092 #[tokio::test]
3095 async fn open_order_refuses_a_borrow_or_repay_effect_unsent() {
3096 use crate::{
3097 client::ExecutionClient,
3098 margin::MarginEffect,
3099 order::{request::RequestOpen, state::InactiveOrderState},
3100 };
3101
3102 let client = BinanceSpot::new(
3103 BinanceSpotConfig::testnet("key".into(), "secret".into()).with_server_time_sync(false),
3104 );
3105 let instrument = InstrumentNameExchange::new("BTCUSDT");
3106
3107 for effect in [MarginEffect::Borrow, MarginEffect::Repay] {
3108 let order = client
3109 .open_order(OrderRequestOpen {
3110 key: OrderKey::new(
3111 ExchangeId::BinanceSpot,
3112 &instrument,
3113 StrategyId::new("strategy"),
3114 ClientOrderId::new("cid"),
3115 ),
3116 state: RequestOpen::new(
3117 Side::Buy,
3118 Some(Decimal::from(50_000)),
3119 Decimal::ONE,
3120 OrderKind::Limit,
3121 TimeInForce::GoodUntilCancelled { post_only: false },
3122 )
3123 .with_margin_effect(effect),
3124 })
3125 .await;
3126
3127 assert!(
3128 matches!(
3129 order.state,
3130 OrderState::Inactive(InactiveOrderState::OpenFailed(
3131 OrderError::UnsupportedMarginEffect(refused)
3132 )) if refused == effect
3133 ),
3134 "{order:?}"
3135 );
3136 }
3137 }
3138
3139 #[tokio::test]
3141 async fn load_rules_returns_the_rules_it_read() {
3142 let server = wiremock::MockServer::start().await;
3143 wiremock::Mock::given(wiremock::matchers::path("/api/v3/exchangeInfo"))
3144 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
3145 serde_json::json!({"symbols": [{
3146 "symbol": "BTCUSDT", "status": "TRADING",
3147 "filters": [
3148 {"filterType": "LOT_SIZE", "minQty": "0.001", "maxQty": "100", "stepSize": "0.001"},
3149 ],
3150 }]}),
3151 ))
3152 .mount(&server)
3153 .await;
3154 let mut client = BinanceSpot::new(
3155 BinanceSpotConfig::testnet("key".into(), "secret".into()).with_server_time_sync(false),
3156 );
3157 client.rules = Arc::new(RulesLoader::new(
3158 PreSendRules::Enforce,
3159 Arc::new(SpotRestApi::from_config(
3160 ConfigurationRestApi::builder()
3161 .api_key("key")
3162 .api_secret("secret")
3163 .base_path(server.uri())
3164 .build()
3165 .unwrap(),
3166 )),
3167 Arc::new(RateLimitTracker::new(WeightPool::Spot)),
3168 ));
3169 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
3170
3171 let loaded = client.load_rules(std::slice::from_ref(&btcusdt)).await;
3172 let rules = loaded.ok().and_then(|mut loaded| loaded.remove(&btcusdt));
3173 assert_eq!(
3174 rules.map(|rules| rules.quantity.step),
3175 Some(Some(Decimal::new(1, 3)))
3176 );
3177 }
3178
3179 #[tokio::test]
3182 async fn open_order_refuses_an_order_breaking_the_symbols_rules_unsent() {
3183 use crate::{
3184 client::ExecutionClient,
3185 error::{Bound, BoundsViolation, OrderField},
3186 order::{request::RequestOpen, state::InactiveOrderState},
3187 };
3188
3189 let server = wiremock::MockServer::start().await;
3190 wiremock::Mock::given(wiremock::matchers::method("GET"))
3191 .and(wiremock::matchers::path("/api/v3/exchangeInfo"))
3192 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
3193 serde_json::json!({"symbols": [{
3194 "symbol": "BTCUSDT", "status": "TRADING",
3195 "filters": [
3196 {"filterType": "LOT_SIZE", "minQty": "0.001", "maxQty": "100", "stepSize": "0.001"},
3197 ],
3198 }]}),
3199 ))
3200 .expect(1)
3201 .mount(&server)
3202 .await;
3203 let mut client = BinanceSpot::new(
3204 BinanceSpotConfig::testnet("key".into(), "secret".into()).with_server_time_sync(false),
3205 );
3206 client.rules = Arc::new(RulesLoader::new(
3207 PreSendRules::Enforce,
3208 Arc::new(SpotRestApi::from_config(
3209 ConfigurationRestApi::builder()
3210 .api_key("key")
3211 .api_secret("secret")
3212 .base_path(server.uri())
3213 .build()
3214 .unwrap(),
3215 )),
3216 Arc::new(RateLimitTracker::new(WeightPool::Spot)),
3217 ));
3218 let instrument = InstrumentNameExchange::new("BTCUSDT");
3219
3220 let order = client
3221 .open_order(OrderRequestOpen {
3222 key: OrderKey::new(
3223 ExchangeId::BinanceSpot,
3224 &instrument,
3225 StrategyId::new("strategy"),
3226 ClientOrderId::new("cid"),
3227 ),
3228 state: RequestOpen::new(
3229 Side::Buy,
3230 Some(Decimal::from(50_000)),
3231 Decimal::from(101),
3232 OrderKind::Limit,
3233 TimeInForce::GoodUntilCancelled { post_only: false },
3234 ),
3235 })
3236 .await;
3237
3238 assert!(
3239 matches!(
3240 order.state,
3241 OrderState::Inactive(InactiveOrderState::OpenFailed(
3242 OrderError::OutOfBounds(violation)
3243 )) if violation == BoundsViolation::new(
3244 OrderField::Quantity,
3245 Bound::Max,
3246 Decimal::from(100),
3247 Decimal::from(101),
3248 )
3249 ),
3250 "{order:?}"
3251 );
3252 }
3253
3254 #[tokio::test]
3257 async fn open_order_refuses_a_trailing_offset_outside_its_sides_range_unsent() {
3258 use crate::{
3259 client::ExecutionClient,
3260 error::{Bound, BoundsViolation, OrderField},
3261 order::{TrailingOffsetType, request::RequestOpen, state::InactiveOrderState},
3262 };
3263
3264 let server = wiremock::MockServer::start().await;
3265 wiremock::Mock::given(wiremock::matchers::method("GET"))
3266 .and(wiremock::matchers::path("/api/v3/exchangeInfo"))
3267 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
3268 serde_json::json!({"symbols": [{
3269 "symbol": "BTCUSDT", "status": "TRADING",
3270 "filters": [
3271 {"filterType": "TRAILING_DELTA", "minTrailingAboveDelta": 10,
3272 "maxTrailingAboveDelta": 2000, "minTrailingBelowDelta": 10,
3273 "maxTrailingBelowDelta": 1000},
3274 ],
3275 }]}),
3276 ))
3277 .expect(1)
3278 .mount(&server)
3279 .await;
3280 let mut client = BinanceSpot::new(
3281 BinanceSpotConfig::testnet("key".into(), "secret".into()).with_server_time_sync(false),
3282 );
3283 client.rules = Arc::new(RulesLoader::new(
3284 PreSendRules::Enforce,
3285 Arc::new(SpotRestApi::from_config(
3286 ConfigurationRestApi::builder()
3287 .api_key("key")
3288 .api_secret("secret")
3289 .base_path(server.uri())
3290 .build()
3291 .unwrap(),
3292 )),
3293 Arc::new(RateLimitTracker::new(WeightPool::Spot)),
3294 ));
3295 let instrument = InstrumentNameExchange::new("BTCUSDT");
3296 let order = |cid: &str, side: Side, offset: Decimal| OrderRequestOpen {
3297 key: OrderKey::new(
3298 ExchangeId::BinanceSpot,
3299 &instrument,
3300 StrategyId::new("strategy"),
3301 ClientOrderId::new(cid),
3302 ),
3303 state: RequestOpen::new(
3304 side,
3305 None,
3306 Decimal::ONE,
3307 OrderKind::TrailingStop {
3308 offset,
3309 offset_type: TrailingOffsetType::BasisPoints,
3310 activation_price: None,
3311 },
3312 TimeInForce::GoodUntilCancelled { post_only: false },
3313 ),
3314 };
3315 let refused_at = |order: Order<_, _, _>| match order.state {
3316 OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::OutOfBounds(
3317 violation,
3318 ))) => Some(violation),
3319 _ => None,
3320 };
3321
3322 let buy = client
3323 .open_order(order("a", Side::Buy, Decimal::from(2500)))
3324 .await;
3325 assert_eq!(
3326 refused_at(buy),
3327 Some(BoundsViolation::new(
3328 OrderField::TrailingOffset,
3329 Bound::Max,
3330 Decimal::from(2000),
3331 Decimal::from(2500),
3332 ))
3333 );
3334 let sell = client
3335 .open_order(order("b", Side::Sell, Decimal::from(1500)))
3336 .await;
3337 assert_eq!(
3338 refused_at(sell),
3339 Some(BoundsViolation::new(
3340 OrderField::TrailingOffset,
3341 Bound::Max,
3342 Decimal::from(1000),
3343 Decimal::from(1500),
3344 ))
3345 );
3346 }
3347
3348 #[test]
3349 fn test_binance_spot_config_new_uses_testnet_by_default() {
3350 let cfg = BinanceSpotConfig::new("my_key".into(), "my_secret".into());
3351 assert!(cfg.testnet);
3352 assert_eq!(cfg.api_key(), "my_key");
3353 }
3354
3355 #[test]
3356 fn test_binance_spot_config_production_is_explicit() {
3357 let cfg = BinanceSpotConfig::production("my_key".into(), "my_secret".into());
3358 assert!(!cfg.testnet);
3359 }
3360
3361 #[test]
3362 fn test_binance_spot_config_debug_redacts_credentials() {
3363 let cfg = BinanceSpotConfig::new("my_key".into(), "my_secret".into());
3364 let debug = format!("{cfg:?}");
3365
3366 assert!(!debug.contains("my_key"), "api_key should be redacted");
3367 assert!(
3368 !debug.contains("my_secret"),
3369 "secret_key should be redacted"
3370 );
3371 assert!(debug.contains("testnet: true"));
3372 }
3373
3374 #[test]
3375 fn test_binance_spot_config_deserialize_omitted_testnet_defaults_to_testnet() {
3376 let cfg: BinanceSpotConfig = serde_json::from_str(
3377 r#"{
3378 "api_key": "my_key",
3379 "secret_key": "my_secret"
3380 }"#,
3381 )
3382 .unwrap();
3383
3384 assert!(cfg.testnet);
3385 assert_eq!(cfg.api_key(), "my_key");
3386 }
3387
3388 #[test]
3389 fn test_binance_spot_config_deserialize_testnet_true() {
3390 let cfg: BinanceSpotConfig = serde_json::from_str(
3391 r#"{
3392 "api_key": "my_key",
3393 "secret_key": "my_secret",
3394 "testnet": true
3395 }"#,
3396 )
3397 .unwrap();
3398
3399 assert!(cfg.testnet);
3400 }
3401
3402 #[test]
3403 fn test_binance_spot_config_deserialize_testnet_false() {
3404 let cfg: BinanceSpotConfig = serde_json::from_str(
3405 r#"{
3406 "api_key": "my_key",
3407 "secret_key": "my_secret",
3408 "testnet": false
3409 }"#,
3410 )
3411 .unwrap();
3412
3413 assert!(!cfg.testnet);
3414 }
3415
3416 #[test]
3417 #[serial_test::serial]
3418 fn test_binance_spot_config_from_env_defaults_to_testnet() {
3419 temp_env::with_vars(
3420 [
3421 ("BINANCE_API_KEY", Some("my_key")),
3422 ("BINANCE_SECRET_KEY", Some("my_secret")),
3423 ("BINANCE_TESTNET", None),
3424 ],
3425 || {
3426 let cfg = BinanceSpotConfig::from_env().unwrap();
3427 assert!(cfg.testnet);
3428 assert_eq!(cfg.api_key(), "my_key");
3429 },
3430 );
3431 }
3432
3433 #[test]
3434 #[serial_test::serial]
3435 fn test_binance_spot_config_from_env_accepts_explicit_production() {
3436 temp_env::with_vars(
3437 [
3438 ("BINANCE_API_KEY", Some("my_key")),
3439 ("BINANCE_SECRET_KEY", Some("my_secret")),
3440 ("BINANCE_TESTNET", Some("false")),
3441 ],
3442 || {
3443 let cfg = BinanceSpotConfig::from_env().unwrap();
3444 assert!(!cfg.testnet);
3445 },
3446 );
3447 }
3448
3449 #[test]
3450 #[serial_test::serial]
3451 fn test_binance_spot_config_from_env_rejects_invalid_testnet() {
3452 temp_env::with_vars(
3453 [
3454 ("BINANCE_API_KEY", Some("my_key")),
3455 ("BINANCE_SECRET_KEY", Some("my_secret")),
3456 ("BINANCE_TESTNET", Some("maybe")),
3457 ],
3458 || {
3459 let err = BinanceSpotConfig::from_env().unwrap_err();
3460 assert!(
3461 matches!(err, BinanceSpotConfigError::InvalidTestnet(value) if value == "maybe")
3462 );
3463 },
3464 );
3465 }
3466
3467 #[test]
3468 #[serial_test::serial]
3469 fn test_binance_spot_config_from_env_requires_credentials() {
3470 temp_env::with_vars(
3471 [
3472 ("BINANCE_API_KEY", None),
3473 ("BINANCE_SECRET_KEY", Some("my_secret")),
3474 ("BINANCE_TESTNET", None),
3475 ],
3476 || {
3477 let err = BinanceSpotConfig::from_env().unwrap_err();
3478 assert!(matches!(err, BinanceSpotConfigError::MissingApiKey));
3479 },
3480 );
3481 }
3482
3483 #[test]
3484 fn test_parse_side() {
3485 assert_eq!(parse_side("BUY"), Some(Side::Buy));
3486 assert_eq!(parse_side("SELL"), Some(Side::Sell));
3487 assert_eq!(parse_side("buy"), None);
3488 assert_eq!(parse_side("UNKNOWN"), None);
3489 }
3490
3491 #[test]
3492 fn test_parse_time_in_force() {
3493 assert_eq!(
3494 parse_time_in_force("GTC"),
3495 TimeInForce::GoodUntilCancelled { post_only: false }
3496 );
3497 assert_eq!(
3498 parse_time_in_force("GTX"),
3499 TimeInForce::GoodUntilCancelled { post_only: true }
3500 );
3501 assert_eq!(parse_time_in_force("IOC"), TimeInForce::ImmediateOrCancel);
3502 assert_eq!(parse_time_in_force("FOK"), TimeInForce::FillOrKill);
3503 assert_eq!(parse_time_in_force("GTD"), TimeInForce::GoodUntilEndOfDay);
3504 assert_eq!(
3506 parse_time_in_force("UNKNOWN"),
3507 TimeInForce::GoodUntilCancelled { post_only: false }
3508 );
3509 }
3510
3511 #[test]
3512 fn test_convert_order_kind_tif() {
3513 assert!(matches!(
3515 convert_order_kind_tif(OrderKind::Market, TimeInForce::ImmediateOrCancel),
3516 Some((OrderPlaceTypeEnum::Market, None))
3517 ));
3518 assert!(matches!(
3519 convert_order_kind_tif(
3520 OrderKind::Limit,
3521 TimeInForce::GoodUntilCancelled { post_only: false }
3522 ),
3523 Some((
3524 OrderPlaceTypeEnum::Limit,
3525 Some(OrderPlaceTimeInForceEnum::Gtc)
3526 ))
3527 ));
3528 assert!(matches!(
3529 convert_order_kind_tif(
3530 OrderKind::Limit,
3531 TimeInForce::GoodUntilCancelled { post_only: true }
3532 ),
3533 Some((OrderPlaceTypeEnum::LimitMaker, None))
3534 ));
3535 assert!(matches!(
3536 convert_order_kind_tif(OrderKind::Limit, TimeInForce::FillOrKill),
3537 Some((
3538 OrderPlaceTypeEnum::Limit,
3539 Some(OrderPlaceTimeInForceEnum::Fok)
3540 ))
3541 ));
3542 assert!(matches!(
3543 convert_order_kind_tif(OrderKind::Limit, TimeInForce::ImmediateOrCancel),
3544 Some((
3545 OrderPlaceTypeEnum::Limit,
3546 Some(OrderPlaceTimeInForceEnum::Ioc)
3547 ))
3548 ));
3549 assert!(convert_order_kind_tif(OrderKind::Limit, TimeInForce::GoodUntilEndOfDay).is_none());
3552 }
3553
3554 #[test]
3555 fn test_convert_order_kind_tif_conditional() {
3556 use rust_decimal::Decimal;
3557
3558 assert!(matches!(
3559 convert_order_kind_tif(
3560 OrderKind::Stop {
3561 trigger_price: Decimal::from(100)
3562 },
3563 TimeInForce::GoodUntilCancelled { post_only: false }
3564 ),
3565 Some((OrderPlaceTypeEnum::StopLoss, None))
3566 ));
3567 assert!(matches!(
3568 convert_order_kind_tif(
3569 OrderKind::StopLimit {
3570 trigger_price: Decimal::from(100)
3571 },
3572 TimeInForce::GoodUntilCancelled { post_only: false }
3573 ),
3574 Some((
3575 OrderPlaceTypeEnum::StopLossLimit,
3576 Some(OrderPlaceTimeInForceEnum::Gtc)
3577 ))
3578 ));
3579 assert!(matches!(
3580 convert_order_kind_tif(
3581 OrderKind::TakeProfit {
3582 trigger_price: Decimal::from(150)
3583 },
3584 TimeInForce::ImmediateOrCancel
3585 ),
3586 Some((OrderPlaceTypeEnum::TakeProfit, None))
3587 ));
3588 assert!(matches!(
3589 convert_order_kind_tif(
3590 OrderKind::TakeProfitLimit {
3591 trigger_price: Decimal::from(150)
3592 },
3593 TimeInForce::FillOrKill
3594 ),
3595 Some((
3596 OrderPlaceTypeEnum::TakeProfitLimit,
3597 Some(OrderPlaceTimeInForceEnum::Fok)
3598 ))
3599 ));
3600 }
3601
3602 #[test]
3603 fn test_convert_order_kind_tif_trailing() {
3604 use rust_decimal::Decimal;
3605
3606 assert!(matches!(
3608 convert_order_kind_tif(
3609 OrderKind::TrailingStop {
3610 offset: Decimal::from(100),
3611 offset_type: TrailingOffsetType::BasisPoints,
3612 activation_price: None,
3613 },
3614 TimeInForce::GoodUntilCancelled { post_only: false }
3615 ),
3616 Some((OrderPlaceTypeEnum::StopLoss, None))
3617 ));
3618 assert!(matches!(
3619 convert_order_kind_tif(
3620 OrderKind::TrailingStop {
3621 offset: Decimal::from(5),
3622 offset_type: TrailingOffsetType::Percentage,
3623 activation_price: None,
3624 },
3625 TimeInForce::GoodUntilCancelled { post_only: false }
3626 ),
3627 Some((OrderPlaceTypeEnum::StopLoss, None))
3628 ));
3629
3630 assert!(
3632 convert_order_kind_tif(
3633 OrderKind::TrailingStop {
3634 offset: Decimal::from(10),
3635 offset_type: TrailingOffsetType::Absolute,
3636 activation_price: None,
3637 },
3638 TimeInForce::GoodUntilCancelled { post_only: false }
3639 )
3640 .is_none()
3641 );
3642
3643 assert!(
3645 convert_order_kind_tif(
3646 OrderKind::TrailingStopLimit {
3647 offset: Decimal::from(100),
3648 offset_type: TrailingOffsetType::BasisPoints,
3649 stop_price: Decimal::from(95),
3650 limit_offset: Decimal::from(10),
3651 },
3652 TimeInForce::GoodUntilCancelled { post_only: false }
3653 )
3654 .is_none()
3655 );
3656 }
3657
3658 #[test]
3661 fn a_fixed_limit_trailing_stop_converts_to_stop_loss_limit() {
3662 let fixed_limit = |offset_type| OrderKind::TrailingStopFixedLimit {
3663 offset: Decimal::from(100),
3664 offset_type,
3665 activation_price: Some(Decimal::from(50_000)),
3666 };
3667 assert!(matches!(
3668 convert_order_kind_tif(
3669 fixed_limit(TrailingOffsetType::BasisPoints),
3670 TimeInForce::GoodUntilCancelled { post_only: false }
3671 ),
3672 Some((
3673 OrderPlaceTypeEnum::StopLossLimit,
3674 Some(OrderPlaceTimeInForceEnum::Gtc)
3675 ))
3676 ));
3677 assert!(
3678 convert_order_kind_tif(
3679 fixed_limit(TrailingOffsetType::Absolute),
3680 TimeInForce::GoodUntilCancelled { post_only: false }
3681 )
3682 .is_none()
3683 );
3684 }
3685
3686 #[test]
3687 fn test_parse_binance_api_error() {
3688 let instrument = InstrumentNameExchange::new("BTCUSDT");
3689
3690 assert!(matches!(
3691 parse_binance_api_error("Insufficient balance".into(), &instrument),
3692 ApiError::BalanceInsufficient(None, _)
3693 ));
3694 assert!(matches!(
3695 parse_binance_api_error("Not enough funds".into(), &instrument),
3696 ApiError::BalanceInsufficient(None, _)
3697 ));
3698 assert_eq!(
3699 parse_binance_api_error("Rate limit exceeded".into(), &instrument),
3700 ApiError::rate_limited(None)
3701 );
3702 assert_eq!(
3703 parse_binance_api_error("Error code -1015".into(), &instrument),
3704 ApiError::rate_limited(None)
3705 );
3706 assert_eq!(
3708 parse_binance_api_error("Error -1003: too many requests".into(), &instrument),
3709 ApiError::rate_limited(None)
3710 );
3711 assert_eq!(
3713 parse_binance_api_error("Error code -2011".into(), &instrument),
3714 ApiError::OrderAlreadyCancelled
3715 );
3716 assert!(matches!(
3718 parse_binance_api_error("Unknown order sent -2013".into(), &instrument),
3719 ApiError::OrderAlreadyCancelled
3720 ));
3721 assert!(matches!(
3723 parse_binance_api_error("Order does not exist -2013".into(), &instrument),
3724 ApiError::OrderAlreadyCancelled
3725 ));
3726 assert!(matches!(
3728 parse_binance_api_error("Unknown order encountered".into(), &instrument),
3729 ApiError::OrderRejected(_)
3730 ));
3731 assert!(matches!(
3732 parse_binance_api_error("Invalid symbol -1121".into(), &instrument),
3733 ApiError::InstrumentInvalid(_, _)
3734 ));
3735 assert!(matches!(
3737 parse_binance_api_error(
3738 "Server-side response error (code -2010): Account has insufficient balance".into(),
3739 &instrument
3740 ),
3741 ApiError::BalanceInsufficient(None, _)
3742 ));
3743 assert!(matches!(
3744 parse_binance_api_error(
3745 "Server-side response error (code -2010): Order would trigger immediately.".into(),
3746 &instrument
3747 ),
3748 ApiError::OrderRejected(_)
3749 ));
3750 assert!(matches!(
3751 parse_binance_api_error("Some other error".into(), &instrument),
3752 ApiError::OrderRejected(_)
3753 ));
3754 }
3755
3756 #[test]
3757 fn test_contains_error_code_suffix_guard() {
3758 assert!(
3760 !contains_error_code("-20130", "-2013"),
3761 "-20130 should not match -2013"
3762 );
3763 assert!(
3764 !contains_error_code("-20131", "-2013"),
3765 "-20131 should not match -2013"
3766 );
3767 assert!(contains_error_code("-2013", "-2013"), "exact match");
3769 assert!(
3770 contains_error_code("Error -2013: text", "-2013"),
3771 "match with trailing text"
3772 );
3773 }
3774
3775 #[test]
3776 fn test_contains_error_code_prefix_guard() {
3777 assert!(
3780 !contains_error_code("1-2013", "-2013"),
3781 "1-2013 should not match -2013"
3782 );
3783 assert!(
3784 !contains_error_code("error 1-2013 text", "-2013"),
3785 "embedded 1-2013 should not match"
3786 );
3787 assert!(
3789 contains_error_code("code=-2013,", "-2013"),
3790 "=-2013 prefix should match"
3791 );
3792 assert!(
3793 contains_error_code(" -2013 ", "-2013"),
3794 "space prefix should match"
3795 );
3796 }
3797
3798 #[test]
3799 fn test_contains_error_code_second_occurrence_valid() {
3800 assert!(
3803 contains_error_code("response code -20130 or -2013: rate limit", "-2013"),
3804 "second valid occurrence must match when first fails digit guard"
3805 );
3806 assert!(
3808 contains_error_code("-2013 or -20130", "-2013"),
3809 "first valid occurrence must match when second is a longer code"
3810 );
3811 }
3812
3813 #[test]
3814 fn test_dedup_cache() {
3815 let cache = new_dedup_cache();
3816 let key = DedupKey {
3817 instrument: SmolStr::from("BTCUSDT"),
3818 id: SmolStr::from("12345"),
3819 kind: DedupEventKind::Trade,
3820 };
3821
3822 assert!(!is_duplicate(&cache, key));
3824 let key = DedupKey {
3826 instrument: SmolStr::from("BTCUSDT"),
3827 id: SmolStr::from("12345"),
3828 kind: DedupEventKind::Trade,
3829 };
3830 assert!(is_duplicate(&cache, key));
3831
3832 let key2 = DedupKey {
3834 instrument: SmolStr::from("BTCUSDT"),
3835 id: SmolStr::from("12345"),
3836 kind: DedupEventKind::OrderState {
3837 filled_quantity: Decimal::ZERO,
3838 },
3839 };
3840 assert!(!is_duplicate(&cache, key2));
3841 }
3842
3843 #[test]
3847 fn dedup_distinguishes_order_states_and_still_collapses_a_replay() {
3848 let cache = new_dedup_cache();
3849 let state = |filled: &str| DedupKey {
3850 instrument: SmolStr::from("BTCUSDT"),
3851 id: SmolStr::from("12345"),
3852 kind: DedupEventKind::OrderState {
3853 filled_quantity: Decimal::from_str(filled).unwrap(),
3854 },
3855 };
3856
3857 assert!(!is_duplicate(&cache, state("0")), "ack is new");
3859 assert!(
3860 !is_duplicate(&cache, state("1")),
3861 "first fill is not the ack"
3862 );
3863 assert!(
3864 !is_duplicate(&cache, state("2")),
3865 "second fill is not the first"
3866 );
3867
3868 assert!(
3871 is_duplicate(&cache, state("0")),
3872 "replayed ack is a duplicate"
3873 );
3874 assert!(
3875 is_duplicate(&cache, state("2")),
3876 "replayed fill is a duplicate"
3877 );
3878
3879 assert!(
3881 is_duplicate(&cache, state("2.00")),
3882 "2.00 must be the same key as 2"
3883 );
3884 }
3885
3886 fn drive_ws_pipeline(frames: &[&str]) -> Vec<UnindexedAccountEvent> {
3893 use crate::client::dedup::dedup_key_from_event;
3894
3895 let cache = new_dedup_cache();
3896 let mut delivered = Vec::new();
3897 let mut buf = Vec::new();
3898 for frame in frames {
3899 let _ = convert_user_data_events(frame, &mut buf);
3900 for ev in buf.drain(..) {
3901 if let Some(key) = dedup_key_from_event(&ev)
3902 && is_duplicate(&cache, key)
3903 {
3904 continue;
3905 }
3906 delivered.push(ev);
3907 }
3908 }
3909 delivered
3910 }
3911
3912 const WS_NEW: &str = r#"{"subscriptionId":0,"event":{"e":"executionReport","s":"BTCUSDT",
3914 "i":12345,"c":"client-1","x":"NEW","X":"NEW","S":"BUY","o":"LIMIT","f":"GTC","q":"2",
3915 "p":"100","z":"0","T":1700000000000}}"#;
3916
3917 const WS_PARTIAL_FILL: &str = r#"{"subscriptionId":0,"event":{"e":"executionReport",
3918 "s":"BTCUSDT","i":12345,"c":"client-1","x":"TRADE","X":"PARTIALLY_FILLED","S":"BUY",
3919 "o":"LIMIT","f":"GTC","q":"2","p":"100","z":"1","l":"1","L":"100","t":555,"n":"0.1",
3920 "N":"USDT","T":1700000001000}}"#;
3921
3922 #[test]
3930 fn ws_fill_snapshot_survives_the_dedup_gate() {
3931 use crate::order::state::ActiveOrderState;
3932
3933 let delivered = drive_ws_pipeline(&[WS_NEW, WS_PARTIAL_FILL]);
3934
3935 assert_eq!(
3936 delivered.len(),
3937 3,
3938 "expected the ack snapshot, then the fill's execution and its snapshot, got: {delivered:?}"
3939 );
3940
3941 let snapshots: Vec<Decimal> = delivered
3942 .iter()
3943 .filter_map(|ev| match &ev.kind {
3944 AccountEventKind::OrderSnapshot(snap) => match &snap.0.state {
3945 OrderState::Active(ActiveOrderState::Open(open)) => Some(open.filled_quantity),
3946 _ => None,
3947 },
3948 _ => None,
3949 })
3950 .collect();
3951
3952 assert_eq!(
3953 snapshots,
3954 vec![Decimal::ZERO, Decimal::ONE],
3955 "the fill's snapshot must reach the consumer carrying the advanced filled quantity"
3956 );
3957 }
3958
3959 #[test]
3962 fn ws_replayed_frame_is_still_deduplicated() {
3963 let delivered = drive_ws_pipeline(&[WS_NEW, WS_PARTIAL_FILL, WS_NEW, WS_PARTIAL_FILL]);
3964
3965 assert_eq!(
3966 delivered.len(),
3967 3,
3968 "re-delivering both frames must add nothing, got: {delivered:?}"
3969 );
3970 }
3971
3972 #[test]
3973 fn test_is_rate_limit_error() {
3974 assert!(is_rate_limit_error(&anyhow::anyhow!(
3976 "Too many requests. You are being rate-limited. Please slow down."
3977 )));
3978 assert!(is_rate_limit_error(&anyhow::anyhow!(
3980 "The IP address has been banned for exceeding rate limits. Contact support."
3981 )));
3982 assert!(is_rate_limit_error(&anyhow::anyhow!(
3984 "Error -1015: too many new orders"
3985 )));
3986 assert!(is_rate_limit_error(&anyhow::anyhow!(
3987 "Error -1003: too many requests"
3988 )));
3989 assert!(!is_rate_limit_error(&anyhow::anyhow!("order 4290 failed")));
3991 assert!(!is_rate_limit_error(&anyhow::anyhow!("connection timeout")));
3992 assert!(!is_rate_limit_error(&anyhow::anyhow!("unknown error")));
3993 assert!(
3995 !is_rate_limit_error(&anyhow::anyhow!("Error -10150: some other error")),
3996 "-10150 should not match -1015"
3997 );
3998 assert!(
3999 !is_rate_limit_error(&anyhow::anyhow!("Error -10030: some other error")),
4000 "-10030 should not match -1003"
4001 );
4002 }
4003
4004 #[test]
4005 fn test_classify_ws_order_error_recognises_venue_responses_only() {
4006 let instrument = InstrumentNameExchange::new("BTCUSDT");
4007
4008 let rejection = anyhow::anyhow!(WebsocketError::ResponseError {
4010 code: -2010,
4011 message: "Account has insufficient balance for requested action.".into(),
4012 retry_after: None,
4013 });
4014 assert!(
4015 matches!(
4016 classify_ws_order_error(&rejection, &instrument),
4017 Some(OrderError::Rejected(ApiError::BalanceInsufficient(None, _)))
4018 ),
4019 "ResponseError should be classified as a venue response"
4020 );
4021
4022 let transport = anyhow::anyhow!("connection reset by peer");
4024 assert!(
4025 classify_ws_order_error(&transport, &instrument).is_none(),
4026 "plain transport error should not be classified as a venue response"
4027 );
4028
4029 let rate_limit = anyhow::anyhow!("Too many requests. You are being rate-limited.");
4031 assert!(
4032 classify_ws_order_error(&rate_limit, &instrument).is_none(),
4033 "rate-limit string error should not be classified as a venue response"
4034 );
4035 }
4036
4037 #[test]
4038 fn test_dedup_key_from_event_trade_includes_instrument() {
4039 use crate::order::id::{OrderId, StrategyId};
4040 use crate::trade::{AssetFees, Trade, TradeId};
4041 use chrono::Utc;
4042 use rust_decimal::Decimal;
4043 use rustrade_instrument::Side;
4044
4045 let instrument = InstrumentNameExchange::new("BTCUSDT");
4046 let trade = Trade::<AssetNameExchange, InstrumentNameExchange>::new(
4047 TradeId::new("9001"),
4048 OrderId::new("4242"),
4049 instrument.clone(),
4050 StrategyId::unknown(),
4051 Utc::now(),
4052 Side::Buy,
4053 Decimal::ZERO,
4054 Decimal::ZERO,
4055 None,
4056 AssetFees::new(
4057 AssetNameExchange::from("USDT"),
4058 Decimal::ZERO,
4059 Some(Decimal::ZERO),
4060 ),
4061 );
4062 let event =
4063 UnindexedAccountEvent::new(ExchangeId::BinanceSpot, AccountEventKind::Trade(trade));
4064
4065 let key = dedup_key_from_event(&event).expect("Trade should produce a DedupKey");
4066 assert_eq!(key.kind, DedupEventKind::Trade);
4067 assert_eq!(key.id.as_str(), "9001", "key.id should be the trade ID");
4068 assert_eq!(
4069 key.instrument.as_str(),
4070 "BTCUSDT",
4071 "key.instrument should be the symbol"
4072 );
4073
4074 let instrument2 = InstrumentNameExchange::new("ETHUSDT");
4076 let trade2 = Trade::<AssetNameExchange, InstrumentNameExchange>::new(
4077 TradeId::new("9001"),
4078 OrderId::new("7777"),
4079 instrument2,
4080 StrategyId::unknown(),
4081 Utc::now(),
4082 Side::Buy,
4083 Decimal::ZERO,
4084 Decimal::ZERO,
4085 None,
4086 AssetFees::new(
4087 AssetNameExchange::from("USDT"),
4088 Decimal::ZERO,
4089 Some(Decimal::ZERO),
4090 ),
4091 );
4092 let event2 =
4093 UnindexedAccountEvent::new(ExchangeId::BinanceSpot, AccountEventKind::Trade(trade2));
4094 let key2 = dedup_key_from_event(&event2).expect("Trade should produce a DedupKey");
4095 assert_ne!(
4096 key, key2,
4097 "same trade ID on different symbols must produce distinct keys"
4098 );
4099 }
4100
4101 fn make_balance_inner(
4106 asset: &str,
4107 free: &str,
4108 locked: &str,
4109 ) -> binance_sdk::spot::websocket_api::OutboundAccountPositionBInner {
4110 binance_sdk::spot::websocket_api::OutboundAccountPositionBInner {
4111 a: Some(asset.to_string()),
4112 f: Some(free.to_string()),
4113 l: Some(locked.to_string()),
4114 }
4115 }
4116
4117 #[test]
4118 fn test_convert_account_position_happy_path() {
4119 let position = binance_sdk::spot::websocket_api::OutboundAccountPosition {
4120 u: Some(1_700_000_000_000),
4121 b_uppercase: Some(vec![make_balance_inner("BTC", "1.5", "0.5")]),
4122 ..Default::default()
4123 };
4124 let mut buf = Vec::new();
4125 convert_account_position(position, &mut buf);
4126
4127 assert_eq!(buf.len(), 1);
4128 match &buf[0].kind {
4129 AccountEventKind::BalanceStreamUpdate(snap) => {
4130 let update = &snap.0;
4131 assert_eq!(update.asset.as_ref(), "BTC");
4132 let expected_free = Decimal::from_str("1.5").unwrap();
4134 let expected_locked = Decimal::from_str("0.5").unwrap();
4135 assert_eq!(update.update.free, expected_free);
4136 assert_eq!(update.update.locked, expected_locked);
4137 assert_eq!(update.update.total(), Decimal::from_str("2.0").unwrap());
4138 }
4139 other => panic!("expected BalanceStreamUpdate, got {:?}", other),
4140 }
4141 }
4142
4143 #[test]
4145 fn test_convert_account_position_is_at_u_else_at_e() {
4146 let time_of = |u, e| {
4147 let position = binance_sdk::spot::websocket_api::OutboundAccountPosition {
4148 u,
4149 e_uppercase: e,
4150 b_uppercase: Some(vec![make_balance_inner("ETH", "2.0", "0.0")]),
4151 };
4152 let mut buf = Vec::new();
4153 convert_account_position(position, &mut buf);
4154 match &buf[..] {
4155 [event] => match &event.kind {
4156 AccountEventKind::BalanceStreamUpdate(snap) => snap.0.time_exchange,
4157 other => panic!("expected BalanceStreamUpdate, got {other:?}"),
4158 },
4159 other => panic!("expected one event, got {other:?}"),
4160 }
4161 };
4162 let at = |ms| Utc.timestamp_millis_opt(ms).unwrap();
4163 assert_eq!(time_of(Some(2_000), Some(3_000)), at(2_000));
4164 assert_eq!(time_of(None, Some(3_000)), at(3_000));
4165 }
4166
4167 #[test]
4168 fn test_convert_account_position_missing_asset_name_skipped() {
4169 let position = binance_sdk::spot::websocket_api::OutboundAccountPosition {
4170 u: Some(1_700_000_000_000),
4171 b_uppercase: Some(vec![
4172 binance_sdk::spot::websocket_api::OutboundAccountPositionBInner {
4173 a: None, f: Some("1.0".to_string()),
4175 l: Some("0.0".to_string()),
4176 },
4177 make_balance_inner("USDT", "100.0", "0.0"), ]),
4179 ..Default::default()
4180 };
4181 let mut buf = Vec::new();
4182 convert_account_position(position, &mut buf);
4183 assert_eq!(buf.len(), 1);
4185 match &buf[0].kind {
4186 AccountEventKind::BalanceStreamUpdate(snap) => {
4187 assert_eq!(snap.0.asset.as_ref(), "USDT");
4188 }
4189 other => panic!("expected BalanceStreamUpdate, got {:?}", other),
4190 }
4191 }
4192
4193 #[test]
4194 fn test_convert_account_position_unparseable_free_skipped() {
4195 let position = binance_sdk::spot::websocket_api::OutboundAccountPosition {
4196 u: Some(1_700_000_000_000),
4197 b_uppercase: Some(vec![
4198 binance_sdk::spot::websocket_api::OutboundAccountPositionBInner {
4199 a: Some("BTC".to_string()),
4200 f: Some("not-a-number".to_string()),
4201 l: Some("0.0".to_string()),
4202 },
4203 make_balance_inner("ETH", "1.0", "0.0"), ]),
4205 ..Default::default()
4206 };
4207 let mut buf = Vec::new();
4208 convert_account_position(position, &mut buf);
4209 assert_eq!(buf.len(), 1);
4211 match &buf[0].kind {
4212 AccountEventKind::BalanceStreamUpdate(snap) => {
4213 assert_eq!(snap.0.asset.as_ref(), "ETH");
4214 }
4215 other => panic!("expected BalanceStreamUpdate, got {:?}", other),
4216 }
4217 }
4218
4219 #[test]
4220 fn test_convert_account_position_empty_balances() {
4221 let position = binance_sdk::spot::websocket_api::OutboundAccountPosition {
4222 u: Some(1_700_000_000_000),
4223 b_uppercase: Some(vec![]),
4224 ..Default::default()
4225 };
4226 let mut buf = Vec::new();
4227 convert_account_position(position, &mut buf);
4228 assert!(
4229 buf.is_empty(),
4230 "empty balance list should produce no events"
4231 );
4232 }
4233
4234 #[test]
4237 fn an_account_position_without_a_venue_time_is_dropped() {
4238 let converted = |u, e_uppercase| {
4239 let position = binance_sdk::spot::websocket_api::OutboundAccountPosition {
4240 u,
4241 e_uppercase,
4242 b_uppercase: Some(vec![make_balance_inner("USDT", "100.0", "0.0")]),
4243 };
4244 let mut buf = Vec::new();
4245 convert_account_position(position, &mut buf);
4246 buf.len()
4247 };
4248 assert_eq!(converted(None, None), 0);
4249 assert_eq!(converted(Some(i64::MAX), None), 0);
4250 assert_eq!(converted(Some(1_700_000_000_000), None), 1);
4251 assert_eq!(converted(None, Some(1_700_000_000_000)), 1);
4252 }
4253
4254 #[test]
4255 fn test_convert_account_position_b_field_none() {
4256 let position = binance_sdk::spot::websocket_api::OutboundAccountPosition {
4257 u: Some(1_700_000_000_000),
4258 b_uppercase: None, ..Default::default()
4260 };
4261 let mut buf = Vec::new();
4262 convert_account_position(position, &mut buf);
4263 assert!(buf.is_empty(), "None B field should produce no events");
4264 }
4265
4266 fn make_base_report() -> ExecutionReport {
4270 ExecutionReport {
4271 symbol: Some("BTCUSDT".to_string()),
4272 order_id: Some(12345),
4273 client_order_id: Some("client-1".to_string()),
4274 transaction_time: Some(1_700_000_000_000),
4275 ..Default::default()
4276 }
4277 }
4278
4279 fn convert(report: ExecutionReport) -> Vec<UnindexedAccountEvent> {
4281 let mut buf = Vec::new();
4282 convert_execution_report(&report, ExchangeId::BinanceSpot, &mut buf);
4283 buf
4284 }
4285
4286 fn sole_event(mut events: Vec<UnindexedAccountEvent>) -> Option<UnindexedAccountEvent> {
4291 assert!(
4292 events.len() <= 1,
4293 "expected at most a single event, got: {events:?}"
4294 );
4295 events.pop()
4296 }
4297
4298 #[test]
4299 fn test_convert_execution_report_new() {
4300 let report = ExecutionReport {
4301 execution_type: Some("NEW".to_string()),
4302 side: Some("BUY".to_string()),
4303 order_type: Some("LIMIT".to_string()),
4304 price: Some("50000.00".to_string()),
4305 order_quantity: Some("0.01".to_string()),
4306 time_in_force: Some("GTC".to_string()),
4307 cumulative_filled_quantity: Some("0".to_string()),
4308 ..make_base_report()
4309 };
4310
4311 let event = sole_event(convert(report)).expect("NEW event should produce Some");
4312 assert_eq!(event.exchange, ExchangeId::BinanceSpot);
4313 match &event.kind {
4314 AccountEventKind::OrderSnapshot(snap) => {
4315 let order = &snap.0;
4316 assert_eq!(order.side, Side::Buy);
4317 assert_eq!(order.kind, OrderKind::Limit);
4318 assert_eq!(order.price, Some(Decimal::from_str("50000.00").unwrap()));
4319 assert_eq!(order.quantity, Decimal::from_str("0.01").unwrap());
4320 assert_eq!(order.key.cid.0.as_str(), "client-1");
4321 assert_eq!(order.key.instrument.name().as_str(), "BTCUSDT");
4322 }
4323 other => panic!("NEW should yield OrderSnapshot, got {other:?}"),
4324 }
4325 }
4326
4327 #[test]
4330 fn a_trailing_stop_loss_limit_report_keeps_its_limit_and_activation_price() {
4331 let report = ExecutionReport {
4332 execution_type: Some("NEW".to_string()),
4333 side: Some("SELL".to_string()),
4334 order_type: Some("STOP_LOSS_LIMIT".to_string()),
4335 price: Some("49000.00".to_string()),
4336 stop_price: Some("50000.00".to_string()),
4337 trailing_delta: Some(WireInt::Int(25)),
4338 order_quantity: Some("0.01".to_string()),
4339 time_in_force: Some("GTC".to_string()),
4340 cumulative_filled_quantity: Some("0".to_string()),
4341 ..make_base_report()
4342 };
4343
4344 let event = sole_event(convert(report)).expect("NEW event should produce Some");
4345 let AccountEventKind::OrderSnapshot(snap) = &event.kind else {
4346 panic!("NEW should yield OrderSnapshot, got {:?}", event.kind);
4347 };
4348 assert_eq!(
4349 snap.0.kind,
4350 OrderKind::TrailingStopFixedLimit {
4351 offset: Decimal::from(25),
4352 offset_type: TrailingOffsetType::BasisPoints,
4353 activation_price: Some(Decimal::from(50_000)),
4354 }
4355 );
4356 assert_eq!(snap.0.price, Some(Decimal::from(49_000)));
4357 }
4358
4359 #[test]
4363 fn a_report_without_an_order_type_has_no_order_snapshot() {
4364 let new = ExecutionReport {
4365 execution_type: Some("NEW".to_string()),
4366 side: Some("BUY".to_string()),
4367 price: Some("50000.00".to_string()),
4368 order_quantity: Some("0.01".to_string()),
4369 ..make_base_report()
4370 };
4371 assert!(convert(new).is_empty());
4372
4373 let fill = ExecutionReport {
4374 execution_type: Some("TRADE".to_string()),
4375 order_status: Some("PARTIALLY_FILLED".to_string()),
4376 side: Some("BUY".to_string()),
4377 trade_id: Some(9999),
4378 last_executed_price: Some("50000.00".to_string()),
4379 last_executed_quantity: Some("0.005".to_string()),
4380 commission_amount: Some("0".to_string()),
4381 cumulative_filled_quantity: Some("0.005".to_string()),
4382 order_type: Some("LIMIT".to_string()),
4383 price: Some("50000.00".to_string()),
4384 order_quantity: Some("0.01".to_string()),
4385 ..make_base_report()
4386 };
4387 let kinds = |report| {
4388 convert(report)
4389 .into_iter()
4390 .map(|event| match event.kind {
4391 AccountEventKind::Trade(_) => "trade",
4392 AccountEventKind::OrderSnapshot(_) => "snapshot",
4393 _ => "other",
4394 })
4395 .collect::<Vec<_>>()
4396 };
4397 assert_eq!(kinds(fill.clone()), ["trade", "snapshot"]);
4398 let untyped = ExecutionReport {
4399 order_type: None,
4400 ..fill
4401 };
4402 assert_eq!(kinds(untyped), ["trade"]);
4403 }
4404
4405 #[test]
4406 fn test_convert_execution_report_trade() {
4407 let report = ExecutionReport {
4410 execution_type: Some("TRADE".to_string()),
4411 side: Some("BUY".to_string()),
4412 trade_id: Some(9999),
4413 last_executed_price: Some("50000.00".to_string()),
4414 last_executed_quantity: Some("0.01".to_string()),
4415 commission_amount: Some("0.000001".to_string()),
4416 ..make_base_report()
4417 };
4418
4419 let event = sole_event(convert(report)).expect("TRADE event should produce Some");
4420 assert_eq!(event.exchange, ExchangeId::BinanceSpot);
4421 match &event.kind {
4422 AccountEventKind::Trade(trade) => {
4423 assert_eq!(trade.side, Side::Buy);
4424 assert_eq!(trade.price, Decimal::from_str("50000.00").unwrap());
4425 assert_eq!(trade.quantity, Decimal::from_str("0.01").unwrap());
4426 assert_eq!(trade.id.0.as_str(), "9999");
4427 assert_eq!(trade.order_id.0.as_str(), "12345");
4428 assert_eq!(trade.instrument.name().as_str(), "BTCUSDT");
4429 }
4430 other => panic!("TRADE should yield Trade, got {other:?}"),
4431 }
4432 }
4433
4434 #[test]
4437 fn a_fills_time_is_its_transaction_time_then_its_event_time_then_this_hosts() {
4438 let trade = |transaction_time, event_time| {
4439 let report = ExecutionReport {
4440 execution_type: Some("TRADE".to_string()),
4441 side: Some("BUY".to_string()),
4442 trade_id: Some(9999),
4443 last_executed_price: Some("50000.00".to_string()),
4444 last_executed_quantity: Some("0.01".to_string()),
4445 transaction_time,
4446 event_time,
4447 ..make_base_report()
4448 };
4449 match sole_event(convert(report)).map(|event| event.kind) {
4450 Some(AccountEventKind::Trade(trade)) => trade,
4451 other => panic!("TRADE should yield Trade, got {other:?}"),
4452 }
4453 };
4454 let at = |ms| Utc.timestamp_millis_opt(ms).unwrap();
4455
4456 let fill = trade(Some(1_700_000_000_000), Some(1_700_000_000_005));
4457 assert_eq!(
4458 (fill.time_exchange, fill.time_source),
4459 (at(1_700_000_000_000), crate::trade::TimeSource::Venue)
4460 );
4461 for transaction_time in [None, Some(i64::MAX)] {
4462 let fill = trade(transaction_time, Some(1_700_000_000_005));
4463 assert_eq!(
4464 (fill.time_exchange, fill.time_source),
4465 (at(1_700_000_000_005), crate::trade::TimeSource::Venue),
4466 "{transaction_time:?}"
4467 );
4468 }
4469 let before = Utc::now();
4470 let fill = trade(None, None);
4471 assert_eq!(fill.time_source, crate::trade::TimeSource::Host);
4472 assert!((before..=Utc::now()).contains(&fill.time_exchange));
4473 }
4474
4475 fn trade_events(
4477 events: Vec<UnindexedAccountEvent>,
4478 ) -> (
4479 Trade<AssetNameExchange, InstrumentNameExchange>,
4480 crate::order::Order<
4481 ExchangeId,
4482 InstrumentNameExchange,
4483 OrderState<AssetNameExchange, InstrumentNameExchange>,
4484 >,
4485 ) {
4486 let [first, second]: [UnindexedAccountEvent; 2] = events
4487 .try_into()
4488 .expect("a fill report must produce an execution and an order snapshot");
4489 let AccountEventKind::Trade(trade) = first.kind else {
4490 panic!("first event must be the Trade, got {:?}", first.kind);
4491 };
4492 let AccountEventKind::OrderSnapshot(snap) = second.kind else {
4493 panic!(
4494 "second event must be the OrderSnapshot, got {:?}",
4495 second.kind
4496 );
4497 };
4498 (trade, snap.0)
4499 }
4500
4501 #[test]
4505 fn a_partially_filled_report_carries_the_cumulative_filled_quantity() {
4506 let report = ExecutionReport {
4507 execution_type: Some("TRADE".to_string()),
4508 order_status: Some("PARTIALLY_FILLED".to_string()),
4509 side: Some("BUY".to_string()),
4510 order_type: Some("LIMIT".to_string()),
4511 price: Some("50000.00".to_string()),
4512 order_quantity: Some("0.01".to_string()),
4513 time_in_force: Some("GTC".to_string()),
4514 trade_id: Some(9999),
4515 last_executed_price: Some("50000.00".to_string()),
4516 last_executed_quantity: Some("0.004".to_string()),
4517 cumulative_filled_quantity: Some("0.004".to_string()),
4518 ..make_base_report()
4519 };
4520
4521 let (trade, order) = trade_events(convert(report));
4522 assert_eq!(trade.quantity, Decimal::from_str("0.004").unwrap());
4523
4524 let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
4525 else {
4526 panic!("expected an Open snapshot, got {:?}", order.state);
4527 };
4528 assert_eq!(open.filled_quantity, Decimal::from_str("0.004").unwrap());
4529 assert_eq!(
4530 open.quantity_remaining(order.quantity),
4531 Decimal::from_str("0.006").unwrap()
4532 );
4533 }
4534
4535 #[test]
4538 fn a_filled_report_reports_nothing_left_to_fill() {
4539 let report = ExecutionReport {
4540 execution_type: Some("TRADE".to_string()),
4541 order_status: Some("FILLED".to_string()),
4542 side: Some("BUY".to_string()),
4543 order_type: Some("LIMIT".to_string()),
4544 price: Some("50000.00".to_string()),
4545 order_quantity: Some("0.01".to_string()),
4546 time_in_force: Some("GTC".to_string()),
4547 trade_id: Some(10_000),
4548 last_executed_price: Some("50000.00".to_string()),
4549 last_executed_quantity: Some("0.006".to_string()),
4550 cumulative_filled_quantity: Some("0.01".to_string()),
4551 ..make_base_report()
4552 };
4553
4554 let (_trade, order) = trade_events(convert(report));
4555 let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
4556 else {
4557 panic!("expected an Open snapshot, got {:?}", order.state);
4558 };
4559 assert!(
4560 open.quantity_remaining(order.quantity).is_zero(),
4561 "a completed order must report nothing left to fill so the engine retires it"
4562 );
4563 }
4564
4565 #[test]
4569 fn a_trade_for_an_order_no_longer_working_emits_the_execution_without_a_snapshot() {
4570 for status in ["CANCELED", "EXPIRED", "REJECTED", "PENDING_CANCEL", "NEW"] {
4571 let report = ExecutionReport {
4572 execution_type: Some("TRADE".to_string()),
4573 order_status: Some(status.to_string()),
4574 side: Some("BUY".to_string()),
4575 order_type: Some("LIMIT".to_string()),
4576 price: Some("50000.00".to_string()),
4577 order_quantity: Some("0.01".to_string()),
4578 time_in_force: Some("GTC".to_string()),
4579 trade_id: Some(9999),
4580 last_executed_price: Some("50000.00".to_string()),
4581 last_executed_quantity: Some("0.004".to_string()),
4582 cumulative_filled_quantity: Some("0.004".to_string()),
4583 ..make_base_report()
4584 };
4585 let events = convert(report);
4586 assert_eq!(
4587 events.len(),
4588 1,
4589 "status {status} must produce the execution alone, got: {events:?}"
4590 );
4591 assert!(
4592 matches!(events[0].kind, AccountEventKind::Trade(_)),
4593 "the execution must still be reported for status {status}"
4594 );
4595 }
4596 }
4597
4598 #[test]
4599 fn test_convert_execution_report_trade_missing_last_price() {
4600 let report = ExecutionReport {
4602 execution_type: Some("TRADE".to_string()),
4603 side: Some("BUY".to_string()),
4604 trade_id: Some(9999),
4605 last_executed_price: None, last_executed_quantity: Some("0.01".to_string()),
4607 ..make_base_report()
4608 };
4609 assert!(
4610 sole_event(convert(report)).is_none(),
4611 "missing last price (L) should return None"
4612 );
4613 }
4614
4615 #[test]
4616 fn test_convert_execution_report_trade_missing_last_qty() {
4617 let report = ExecutionReport {
4619 execution_type: Some("TRADE".to_string()),
4620 side: Some("BUY".to_string()),
4621 trade_id: Some(9999),
4622 last_executed_price: Some("50000.00".to_string()),
4623 last_executed_quantity: None, ..make_base_report()
4625 };
4626 assert!(
4627 sole_event(convert(report)).is_none(),
4628 "missing last qty (l) should return None"
4629 );
4630 }
4631
4632 #[test]
4633 fn test_convert_execution_report_canceled() {
4634 let report = ExecutionReport {
4635 execution_type: Some("CANCELED".to_string()),
4636 ..make_base_report()
4637 };
4638
4639 let event = sole_event(convert(report)).expect("CANCELED should produce Some");
4640 assert!(
4641 matches!(event.kind, AccountEventKind::OrderCancelled(ref r) if r.state.is_ok()),
4642 "CANCELED should yield OrderCancelled with Ok state"
4643 );
4644 }
4645
4646 #[test]
4649 fn a_cancel_report_is_keyed_by_the_cancelled_orders_id() {
4650 let cancelled_cid = |orig: Option<&str>| {
4651 let report = ExecutionReport {
4652 execution_type: Some("CANCELED".to_string()),
4653 client_order_id: Some("cancel-request".to_string()),
4654 orig_client_order_id: orig.map(str::to_string),
4655 ..make_base_report()
4656 };
4657 let event = sole_event(convert(report)).expect("CANCELED should produce Some");
4658 let AccountEventKind::OrderCancelled(response) = event.kind else {
4659 panic!("expected OrderCancelled, got {:?}", event.kind);
4660 };
4661 response.key.cid
4662 };
4663
4664 assert_eq!(
4665 cancelled_cid(Some("client-1")),
4666 ClientOrderId::new("client-1")
4667 );
4668 assert_eq!(
4669 cancelled_cid(Some("")),
4670 ClientOrderId::new("cancel-request")
4671 );
4672 assert_eq!(cancelled_cid(None), ClientOrderId::new("cancel-request"));
4673 }
4674
4675 #[test]
4678 fn only_a_cancel_report_reads_the_orders_id_from_c_upper() {
4679 for exec_type in ["NEW", "TRADE", "EXPIRED", "TRADE_PREVENTION", "REPLACED"] {
4680 let report = ExecutionReport {
4681 execution_type: Some(exec_type.to_string()),
4682 orig_client_order_id: Some("client-0".to_string()),
4683 ..make_base_report()
4684 };
4685 assert_eq!(report.order_client_id(), Some("client-1"), "{exec_type}");
4686 }
4687 }
4688
4689 #[test]
4691 fn a_cancel_report_carries_its_fill_or_none() {
4692 let cancelled = |z: Option<&str>| {
4693 let report = ExecutionReport {
4694 execution_type: Some("CANCELED".to_string()),
4695 cumulative_filled_quantity: z.map(str::to_string),
4696 ..make_base_report()
4697 };
4698 let event = sole_event(convert(report)).expect("CANCELED should produce Some");
4699 let AccountEventKind::OrderCancelled(response) = event.kind else {
4700 panic!("expected OrderCancelled, got {:?}", event.kind);
4701 };
4702 let Ok(cancelled) = response.state else {
4703 panic!("expected Ok, got {:?}", response.state);
4704 };
4705 cancelled.filled_quantity
4706 };
4707
4708 assert_eq!(cancelled(Some("0.5")), Some(Decimal::new(5, 1)));
4709 assert_eq!(cancelled(Some("0")), Some(Decimal::ZERO));
4710 assert_eq!(cancelled(None), None);
4711 assert_eq!(cancelled(Some("not a number")), None);
4712 }
4713
4714 #[test]
4715 fn test_convert_execution_report_expired() {
4716 let report = ExecutionReport {
4717 execution_type: Some("EXPIRED".to_string()),
4718 ..make_base_report()
4719 };
4720 let event = sole_event(convert(report)).expect("EXPIRED should produce Some");
4721 assert!(
4722 matches!(event.kind, AccountEventKind::OrderCancelled(ref r) if r.state.is_ok()),
4723 "EXPIRED should yield OrderCancelled with Ok state"
4724 );
4725 }
4726
4727 #[test]
4730 fn a_trade_prevention_report_removes_the_order_only_once_it_has_ended() {
4731 let prevented = |status: Option<&str>| {
4732 convert(ExecutionReport {
4733 execution_type: Some("TRADE_PREVENTION".to_string()),
4734 order_status: status.map(str::to_string),
4735 ..make_base_report()
4736 })
4737 };
4738
4739 for ended in ["EXPIRED_IN_MATCH", "EXPIRED", "FILLED"] {
4740 let event = sole_event(prevented(Some(ended))).expect(ended);
4741 let AccountEventKind::OrderCancelled(response) = event.kind else {
4742 panic!("{ended}: expected OrderCancelled, got {:?}", event.kind);
4743 };
4744 assert_eq!(response.key.cid, ClientOrderId::new("client-1"), "{ended}");
4745 let Ok(cancelled) = response.state else {
4746 panic!("{ended}: expected Ok, got {:?}", response.state);
4747 };
4748 assert_eq!(cancelled.id, OrderId::new("12345"), "{ended}");
4749 }
4750 for not_ended in [
4751 Some("NEW"),
4752 Some("PARTIALLY_FILLED"),
4753 Some("SOME_FUTURE_STATUS"),
4754 None,
4755 ] {
4756 assert!(prevented(not_ended).is_empty(), "{not_ended:?}");
4757 }
4758 }
4759
4760 #[test]
4762 fn expired_in_match_as_an_execution_type_produces_nothing() {
4763 let report = ExecutionReport {
4764 execution_type: Some("EXPIRED_IN_MATCH".to_string()),
4765 ..make_base_report()
4766 };
4767 assert!(convert(report).is_empty());
4768 }
4769
4770 #[test]
4771 fn test_convert_execution_report_rejected() {
4772 let report = ExecutionReport {
4773 execution_type: Some("REJECTED".to_string()),
4774 reject_reason: Some("INSUFFICIENT_FUNDS".to_string()),
4775 ..make_base_report()
4776 };
4777
4778 let event = sole_event(convert(report)).expect("REJECTED should produce Some");
4779 assert!(
4780 matches!(
4781 event.kind,
4782 AccountEventKind::OrderCancelled(ref r) if matches!(
4783 &r.state,
4784 Err(OrderError::Rejected(ApiError::BalanceInsufficient(None, reason)))
4785 if reason == "INSUFFICIENT_FUNDS"
4786 )
4787 ),
4788 "a REJECTED shortfall should go through the classifier, got {:?}",
4789 event.kind
4790 );
4791 }
4792
4793 #[test]
4794 fn test_convert_execution_report_rejected_for_another_reason() {
4795 let report = ExecutionReport {
4796 execution_type: Some("REJECTED".to_string()),
4797 reject_reason: Some("WOULD_MATCH_IMMEDIATELY".to_string()),
4798 ..make_base_report()
4799 };
4800
4801 let event = sole_event(convert(report)).expect("REJECTED should produce Some");
4802 assert!(
4803 matches!(
4804 event.kind,
4805 AccountEventKind::OrderCancelled(ref r) if matches!(
4806 &r.state,
4807 Err(OrderError::Rejected(ApiError::OrderRejected(reason)))
4808 if reason == "WOULD_MATCH_IMMEDIATELY"
4809 )
4810 ),
4811 "an unrecognised REJECTED reason should stay OrderRejected, got {:?}",
4812 event.kind
4813 );
4814 }
4815
4816 #[test]
4817 fn test_convert_execution_report_missing_exec_type() {
4818 let report = ExecutionReport {
4820 symbol: Some("BTCUSDT".to_string()),
4821 ..Default::default()
4822 };
4823 assert!(
4824 sole_event(convert(report)).is_none(),
4825 "missing execution type should return None"
4826 );
4827 }
4828
4829 #[test]
4830 fn test_convert_execution_report_missing_symbol() {
4831 let report = ExecutionReport {
4833 execution_type: Some("NEW".to_string()),
4834 ..Default::default()
4835 };
4836 assert!(
4837 sole_event(convert(report)).is_none(),
4838 "missing symbol should return None"
4839 );
4840 }
4841
4842 #[test]
4843 fn test_convert_execution_report_missing_order_id() {
4844 let report = ExecutionReport {
4846 execution_type: Some("NEW".to_string()),
4847 symbol: Some("BTCUSDT".to_string()),
4848 order_id: None,
4849 ..Default::default()
4850 };
4851 assert!(
4852 sole_event(convert(report)).is_none(),
4853 "missing orderId should return None"
4854 );
4855 }
4856
4857 #[test]
4858 fn test_convert_execution_report_trade_missing_trade_id() {
4859 let report = ExecutionReport {
4861 execution_type: Some("TRADE".to_string()),
4862 trade_id: None,
4863 ..make_base_report()
4864 };
4865 assert!(
4866 sole_event(convert(report)).is_none(),
4867 "TRADE event missing tradeId should return None"
4868 );
4869 }
4870
4871 #[test]
4873 fn an_amend_report_produces_nothing() {
4874 for exec_type in ["REPLACED", "REPLACE"] {
4875 let report = ExecutionReport {
4876 execution_type: Some(exec_type.to_string()),
4877 order_status: Some("NEW".to_string()),
4878 orig_client_order_id: Some("client-0".to_string()),
4879 ..make_base_report()
4880 };
4881 assert!(convert(report).is_empty(), "{exec_type}");
4882 }
4883 }
4884
4885 fn make_balance(
4890 asset: &str,
4891 free: &str,
4892 locked: &str,
4893 ) -> binance_sdk::spot::rest_api::GetAccountResponseBalancesInner {
4894 binance_sdk::spot::rest_api::GetAccountResponseBalancesInner {
4895 asset: Some(asset.to_string()),
4896 free: Some(free.to_string()),
4897 locked: Some(locked.to_string()),
4898 }
4899 }
4900
4901 #[test]
4902 fn test_filter_balances_empty_assets_returns_all() {
4903 let balances = vec![
4904 make_balance("BTC", "1.0", "0.0"),
4905 make_balance("USDT", "500.0", "50.0"),
4906 ];
4907 let result = filter_and_convert_balances(balances, &[], Utc::now());
4908 assert_eq!(result.len(), 2, "empty filter should return all balances");
4909 }
4910
4911 #[test]
4912 fn test_filter_balances_matching_asset_returned() {
4913 let balances = vec![
4914 make_balance("BTC", "1.5", "0.5"),
4915 make_balance("ETH", "10.0", "0.0"),
4916 ];
4917 let assets = vec![AssetNameExchange::new("BTC")];
4918 let result = filter_and_convert_balances(balances, &assets, Utc::now());
4919 assert_eq!(result.len(), 1);
4920 assert_eq!(result[0].asset, AssetNameExchange::new("BTC"));
4921 assert_eq!(result[0].balance.total, Decimal::from_str("2.0").unwrap());
4923 assert_eq!(result[0].balance.free, Decimal::from_str("1.5").unwrap());
4924 }
4925
4926 #[test]
4927 fn test_filter_balances_non_matching_asset_filtered_out() {
4928 let balances = vec![
4929 make_balance("BTC", "1.0", "0.0"),
4930 make_balance("ETH", "2.0", "0.0"),
4931 ];
4932 let assets = vec![AssetNameExchange::new("USDT")];
4933 let result = filter_and_convert_balances(balances, &assets, Utc::now());
4934 assert!(
4935 result.is_empty(),
4936 "non-matching asset should be filtered out"
4937 );
4938 }
4939
4940 #[test]
4941 fn test_filter_balances_missing_asset_field_skipped() {
4942 let balances = vec![
4943 binance_sdk::spot::rest_api::GetAccountResponseBalancesInner {
4944 asset: None,
4945 free: Some("1.0".to_string()),
4946 locked: Some("0.0".to_string()),
4947 },
4948 make_balance("USDT", "100.0", "0.0"),
4949 ];
4950 let result = filter_and_convert_balances(balances, &[], Utc::now());
4951 assert_eq!(
4952 result.len(),
4953 1,
4954 "entry with missing asset should be skipped"
4955 );
4956 assert_eq!(result[0].asset, AssetNameExchange::new("USDT"));
4957 }
4958
4959 #[test]
4960 fn test_filter_balances_unparseable_free_skipped() {
4961 let balances = vec![
4962 binance_sdk::spot::rest_api::GetAccountResponseBalancesInner {
4963 asset: Some("BTC".to_string()),
4964 free: Some("not-a-number".to_string()),
4965 locked: Some("0.0".to_string()),
4966 },
4967 ];
4968 let result = filter_and_convert_balances(balances, &[], Utc::now());
4969 assert!(
4970 result.is_empty(),
4971 "unparseable free field should be skipped"
4972 );
4973 }
4974
4975 #[test]
4976 fn test_filter_balances_zero_balance_included() {
4977 let balances = vec![
4980 make_balance("BTC", "0.00000000", "0.00000000"),
4981 make_balance("USDT", "100.0", "0.0"),
4982 ];
4983 let result = filter_and_convert_balances(balances, &[], Utc::now());
4984 assert_eq!(result.len(), 2, "zero-balance entries must be included");
4985 let btc = result
4986 .iter()
4987 .find(|b| b.asset == AssetNameExchange::new("BTC"))
4988 .unwrap();
4989 assert_eq!(btc.balance.total, Decimal::ZERO);
4990 assert_eq!(btc.balance.free, Decimal::ZERO);
4991 }
4992
4993 #[test]
4994 fn test_filter_balances_duplicate_assets_in_response() {
4995 let balances = vec![
4999 make_balance("BTC", "1.0", "0.0"),
5000 make_balance("BTC", "2.0", "0.0"),
5001 ];
5002 let result = filter_and_convert_balances(balances, &[], Utc::now());
5003 assert_eq!(
5004 result.len(),
5005 2,
5006 "duplicate asset entries produce two AssetBalance entries"
5007 );
5008 }
5009
5010 fn make_base_trade() -> binance_sdk::spot::rest_api::MyTradesResponseInner {
5015 binance_sdk::spot::rest_api::MyTradesResponseInner {
5016 id: Some(9001),
5017 order_id: Some(4242),
5018 price: Some("50000.00".to_string()),
5019 qty: Some("0.01".to_string()),
5020 commission: Some("0.05".to_string()),
5021 time: Some(1_700_000_000_000),
5022 is_buyer: Some(true),
5023 ..Default::default()
5024 }
5025 }
5026
5027 #[test]
5028 fn test_convert_my_trade_happy_path() {
5029 let instrument = InstrumentNameExchange::new("BTCUSDT");
5030 let trade =
5031 convert_my_trade(&make_base_trade(), &instrument).expect("valid trade should convert");
5032 assert_eq!(trade.instrument, instrument);
5033 assert_eq!(trade.side, Side::Buy);
5034 assert_eq!(trade.price, Decimal::from_str("50000.00").unwrap());
5035 assert_eq!(trade.quantity, Decimal::from_str("0.01").unwrap());
5036 }
5037
5038 #[test]
5041 fn a_rest_trade_without_a_time_is_at_this_hosts_time() {
5042 let instrument = InstrumentNameExchange::new("BTCUSDT");
5043 let trade = convert_my_trade(&make_base_trade(), &instrument).unwrap();
5044 assert_eq!(
5045 (trade.time_exchange, trade.time_source),
5046 (
5047 Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(),
5048 TimeSource::Venue
5049 )
5050 );
5051 let before = Utc::now();
5052 let t = binance_sdk::spot::rest_api::MyTradesResponseInner {
5053 time: None,
5054 ..make_base_trade()
5055 };
5056 let trade = convert_my_trade(&t, &instrument).unwrap();
5057 assert_eq!(trade.time_source, TimeSource::Host);
5058 assert!((before..=Utc::now()).contains(&trade.time_exchange));
5059 }
5060
5061 #[test]
5062 fn test_convert_my_trade_sell_side() {
5063 let instrument = InstrumentNameExchange::new("BTCUSDT");
5064 let t = binance_sdk::spot::rest_api::MyTradesResponseInner {
5065 is_buyer: Some(false),
5066 ..make_base_trade()
5067 };
5068 let trade = convert_my_trade(&t, &instrument).expect("sell-side trade should convert");
5069 assert_eq!(trade.side, Side::Sell);
5070 }
5071
5072 #[test]
5073 fn test_convert_my_trade_missing_id_returns_none() {
5074 let instrument = InstrumentNameExchange::new("BTCUSDT");
5075 let t = binance_sdk::spot::rest_api::MyTradesResponseInner {
5076 id: None,
5077 ..make_base_trade()
5078 };
5079 assert!(
5080 convert_my_trade(&t, &instrument).is_none(),
5081 "missing id should return None"
5082 );
5083 }
5084
5085 #[test]
5086 fn test_convert_my_trade_missing_order_id_returns_none() {
5087 let instrument = InstrumentNameExchange::new("BTCUSDT");
5088 let t = binance_sdk::spot::rest_api::MyTradesResponseInner {
5089 order_id: None,
5090 ..make_base_trade()
5091 };
5092 assert!(
5093 convert_my_trade(&t, &instrument).is_none(),
5094 "missing orderId should return None"
5095 );
5096 }
5097
5098 #[test]
5099 fn test_convert_my_trade_missing_is_buyer_returns_none() {
5100 let instrument = InstrumentNameExchange::new("BTCUSDT");
5101 let t = binance_sdk::spot::rest_api::MyTradesResponseInner {
5102 is_buyer: None,
5103 ..make_base_trade()
5104 };
5105 assert!(
5106 convert_my_trade(&t, &instrument).is_none(),
5107 "missing isBuyer should return None"
5108 );
5109 }
5110
5111 #[test]
5112 fn test_convert_my_trade_commission_none_defaults_to_zero() {
5113 let instrument = InstrumentNameExchange::new("BTCUSDT");
5114 let t = binance_sdk::spot::rest_api::MyTradesResponseInner {
5115 commission: None,
5116 ..make_base_trade()
5117 };
5118 let trade =
5119 convert_my_trade(&t, &instrument).expect("None commission should still convert");
5120 assert_eq!(trade.fees.fees, Decimal::ZERO);
5121 }
5122
5123 fn make_base_open_order() -> binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5128 binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5129 order_id: Some(12345),
5130 client_order_id: Some("cid-abc".to_string()),
5131 side: Some("BUY".to_string()),
5132 r#type: Some("LIMIT".to_string()),
5133 price: Some("50000.00".to_string()),
5134 orig_qty: Some("0.01".to_string()),
5135 executed_qty: Some("0.0".to_string()),
5136 time_in_force: Some("GTC".to_string()),
5137 time: Some(1_700_000_000_000),
5138 status: Some("NEW".to_string()),
5139 ..Default::default()
5140 }
5141 }
5142
5143 #[test]
5144 fn test_convert_open_order_all_orders_response_converts_identically() {
5145 let instrument = InstrumentNameExchange::new("BTCUSDT");
5152 let all_orders = binance_sdk::spot::rest_api::AllOrdersResponseInner {
5153 order_id: Some(12345),
5154 client_order_id: Some("cid-abc".to_string()),
5155 side: Some("BUY".to_string()),
5156 r#type: Some("LIMIT".to_string()),
5157 price: Some("50000.00".to_string()),
5158 orig_qty: Some("0.01".to_string()),
5159 executed_qty: Some("0.0".to_string()),
5160 time_in_force: Some("GTC".to_string()),
5161 time: Some(1_700_000_000_000),
5162 status: Some("NEW".to_string()),
5163 ..Default::default()
5164 };
5165 let from_all_orders =
5166 convert_open_order(&all_orders, ExchangeId::BinanceSpot, &instrument, &|_| None)
5167 .expect("valid order should convert");
5168 let from_open_orders = convert_open_order(
5169 &make_base_open_order(),
5170 ExchangeId::BinanceSpot,
5171 &instrument,
5172 &|_| None,
5173 )
5174 .expect("valid order should convert");
5175 assert_eq!(from_all_orders, from_open_orders);
5176 }
5177
5178 #[test]
5184 fn test_convert_open_order_refuses_an_all_orders_row_that_is_not_live() {
5185 let instrument = InstrumentNameExchange::new("BTCUSDT");
5186 for status in [
5187 "CANCELED",
5188 "FILLED",
5189 "EXPIRED",
5190 "EXPIRED_IN_MATCH",
5191 "REJECTED",
5192 "PENDING_CANCEL",
5193 "SOME_FUTURE_STATUS",
5194 ] {
5195 let row = binance_sdk::spot::rest_api::AllOrdersResponseInner {
5196 order_id: Some(12345),
5197 client_order_id: Some("cid-abc".to_string()),
5198 side: Some("BUY".to_string()),
5199 r#type: Some("LIMIT".to_string()),
5200 price: Some("50000.00".to_string()),
5201 orig_qty: Some("0.01".to_string()),
5202 executed_qty: Some("0.004".to_string()),
5203 time_in_force: Some("GTC".to_string()),
5204 time: Some(1_700_000_000_000),
5205 status: Some(status.to_string()),
5206 ..Default::default()
5207 };
5208 assert_eq!(
5209 convert_open_order(&row, ExchangeId::BinanceSpot, &instrument, &|_| None),
5210 None,
5211 "{status}"
5212 );
5213 }
5214 }
5215
5216 #[test]
5217 fn test_convert_open_order_admits_every_live_status() {
5218 let instrument = InstrumentNameExchange::new("BTCUSDT");
5219 for status in ["NEW", "PARTIALLY_FILLED", "PENDING_NEW"] {
5220 let o = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5221 status: Some(status.to_string()),
5222 ..make_base_open_order()
5223 };
5224 assert!(
5225 convert_open_order(&o, ExchangeId::BinanceSpot, &instrument, &|_| None).is_some(),
5226 "{status}"
5227 );
5228 }
5229 }
5230
5231 #[test]
5236 fn test_convert_open_order_listing_is_complete_only_when_every_row_survives_under_its_cid() {
5237 let instrument = InstrumentNameExchange::new("BTCUSDT");
5238 let second = || binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5239 order_id: Some(67890),
5240 client_order_id: Some("cid-def".to_string()),
5241 ..make_base_open_order()
5242 };
5243
5244 let listing = convert_open_order_listing(
5245 &[make_base_open_order(), second()],
5246 ExchangeId::BinanceSpot,
5247 &instrument,
5248 &|_| None,
5249 );
5250 assert_eq!(listing.orders.len(), 2);
5251 assert!(listing.complete);
5252
5253 let empty = convert_open_order_listing::<
5254 binance_sdk::spot::rest_api::GetOpenOrdersResponseInner,
5255 >(&[], ExchangeId::BinanceSpot, &instrument, &|_| None);
5256 assert!(empty.orders.is_empty());
5257 assert!(empty.complete, "no open orders is a complete answer");
5258
5259 let dropped = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5260 side: None,
5261 ..second()
5262 };
5263 let listing = convert_open_order_listing(
5264 &[make_base_open_order(), dropped],
5265 ExchangeId::BinanceSpot,
5266 &instrument,
5267 &|_| None,
5268 );
5269 assert_eq!(listing.orders.len(), 1);
5270 assert!(!listing.complete, "a dropped row may be a live order");
5271
5272 let without_cid = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5273 client_order_id: None,
5274 ..second()
5275 };
5276 let listing = convert_open_order_listing(
5277 &[make_base_open_order(), without_cid],
5278 ExchangeId::BinanceSpot,
5279 &instrument,
5280 &|_| None,
5281 );
5282 assert_eq!(listing.orders.len(), 2, "the row is still converted");
5283 assert!(
5284 !listing.complete,
5285 "an order kept under its orderId cannot be found under the cid it was placed with"
5286 );
5287 }
5288
5289 #[test]
5290 fn test_convert_open_order_missing_status_returns_none() {
5291 let instrument = InstrumentNameExchange::new("BTCUSDT");
5292 let o = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5293 status: None,
5294 ..make_base_open_order()
5295 };
5296 assert!(convert_open_order(&o, ExchangeId::BinanceSpot, &instrument, &|_| None).is_none());
5297 }
5298
5299 #[test]
5305 fn test_convert_open_order_prefers_update_time_over_creation_time() {
5306 let instrument = InstrumentNameExchange::new("BTCUSDT");
5307
5308 let updated = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5309 update_time: Some(1_700_000_005_000),
5310 ..make_base_open_order()
5311 };
5312 let order = convert_open_order(&updated, ExchangeId::BinanceSpot, &instrument, &|_| None)
5313 .expect("valid order should convert");
5314 assert_eq!(
5315 order.state.time_exchange,
5316 Utc.timestamp_millis_opt(1_700_000_005_000)
5317 .single()
5318 .unwrap(),
5319 "update_time must win over time"
5320 );
5321
5322 let order = convert_open_order(
5325 &make_base_open_order(),
5326 ExchangeId::BinanceSpot,
5327 &instrument,
5328 &|_| None,
5329 )
5330 .expect("valid order should convert");
5331 assert_eq!(
5332 order.state.time_exchange,
5333 Utc.timestamp_millis_opt(1_700_000_000_000)
5334 .single()
5335 .unwrap(),
5336 "absent update_time falls back to time"
5337 );
5338 }
5339
5340 #[test]
5341 fn test_convert_open_order_happy_path() {
5342 let instrument = InstrumentNameExchange::new("BTCUSDT");
5343 let order = convert_open_order(
5344 &make_base_open_order(),
5345 ExchangeId::BinanceSpot,
5346 &instrument,
5347 &|_| None,
5348 )
5349 .expect("valid order should convert");
5350 assert_eq!(order.key.instrument, instrument);
5351 assert_eq!(order.side, Side::Buy);
5352 assert_eq!(order.kind, OrderKind::Limit);
5353 assert_eq!(order.price, Some(Decimal::from_str("50000.00").unwrap()));
5354 assert_eq!(order.quantity, Decimal::from_str("0.01").unwrap());
5355 assert_eq!(order.state.filled_quantity, Decimal::ZERO);
5356 }
5357
5358 #[test]
5359 fn test_convert_open_order_owned_symbol_recovers_instrument() {
5360 let o = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5362 symbol: Some("ETHUSDT".to_string()),
5363 ..make_base_open_order()
5364 };
5365 let order = convert_open_order_owned_symbol(&o, ExchangeId::BinanceSpot, &|_| None)
5366 .expect("valid order should convert");
5367 assert_eq!(order.key.instrument.name().as_str(), "ETHUSDT");
5368 assert_eq!(order.side, Side::Buy);
5369 }
5370
5371 #[test]
5372 fn test_convert_open_order_owned_symbol_missing_symbol_returns_none() {
5373 let o = make_base_open_order();
5375 assert!(o.symbol.is_none());
5376 assert!(convert_open_order_owned_symbol(&o, ExchangeId::BinanceSpot, &|_| None).is_none());
5377 }
5378
5379 #[test]
5380 fn test_convert_open_order_missing_order_id_returns_none() {
5381 let instrument = InstrumentNameExchange::new("BTCUSDT");
5382 let o = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5383 order_id: None,
5384 ..make_base_open_order()
5385 };
5386 assert!(
5387 convert_open_order(&o, ExchangeId::BinanceSpot, &instrument, &|_| None).is_none(),
5388 "missing orderId should return None"
5389 );
5390 }
5391
5392 #[test]
5393 fn test_convert_open_order_missing_side_returns_none() {
5394 let instrument = InstrumentNameExchange::new("BTCUSDT");
5395 let o = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5396 side: None,
5397 ..make_base_open_order()
5398 };
5399 assert!(
5400 convert_open_order(&o, ExchangeId::BinanceSpot, &instrument, &|_| None).is_none(),
5401 "missing side should return None"
5402 );
5403 }
5404
5405 #[test]
5406 fn test_convert_open_order_missing_type_returns_none() {
5407 let instrument = InstrumentNameExchange::new("BTCUSDT");
5408 let o = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5409 r#type: None,
5410 ..make_base_open_order()
5411 };
5412 assert!(
5413 convert_open_order(&o, ExchangeId::BinanceSpot, &instrument, &|_| None).is_none(),
5414 "missing type should return None"
5415 );
5416 }
5417
5418 #[test]
5419 fn test_convert_open_order_executed_qty_none_defaults_to_zero() {
5420 let instrument = InstrumentNameExchange::new("BTCUSDT");
5421 let o = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5422 executed_qty: None,
5423 ..make_base_open_order()
5424 };
5425 let order = convert_open_order(&o, ExchangeId::BinanceSpot, &instrument, &|_| None)
5426 .expect("None executedQty should still convert");
5427 assert_eq!(
5428 order.state.filled_quantity,
5429 Decimal::ZERO,
5430 "None executedQty should default to zero"
5431 );
5432 }
5433
5434 #[test]
5435 fn test_convert_open_order_executed_qty_unparseable_defaults_to_zero() {
5436 let instrument = InstrumentNameExchange::new("BTCUSDT");
5437 let o = binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
5438 executed_qty: Some("bad-value".to_string()),
5439 ..make_base_open_order()
5440 };
5441 let order = convert_open_order(&o, ExchangeId::BinanceSpot, &instrument, &|_| None)
5442 .expect("unparseable executedQty should still convert");
5443 assert_eq!(
5444 order.state.filled_quantity,
5445 Decimal::ZERO,
5446 "unparseable executedQty should default to zero"
5447 );
5448 }
5449
5450 #[test]
5455 fn test_dedup_key_from_event_non_open_states_return_none() {
5456 use crate::order::state::{
5457 ActiveOrderState, CancelInFlight, InactiveOrderState, OpenInFlight,
5458 };
5459 use rustrade_integration::collection::snapshot::Snapshot;
5460
5461 let key = OrderKey::new(
5462 ExchangeId::BinanceSpot,
5463 InstrumentNameExchange::new("BTCUSDT"),
5464 StrategyId::unknown(),
5465 ClientOrderId::new("cid1"),
5466 );
5467
5468 let event = UnindexedAccountEvent::new(
5470 ExchangeId::BinanceSpot,
5471 AccountEventKind::OrderSnapshot(Snapshot(Order::new(
5472 key.clone(),
5473 Side::Buy,
5474 None, Decimal::ZERO,
5476 OrderKind::Market,
5477 TimeInForce::ImmediateOrCancel,
5478 OrderState::<AssetNameExchange, InstrumentNameExchange>::Active(
5479 ActiveOrderState::OpenInFlight(OpenInFlight::new(Utc::now())),
5480 ),
5481 ))),
5482 );
5483 assert!(
5484 dedup_key_from_event(&event).is_none(),
5485 "OpenInFlight should return None — dedup not meaningful before exchange ack"
5486 );
5487
5488 let event = UnindexedAccountEvent::new(
5490 ExchangeId::BinanceSpot,
5491 AccountEventKind::OrderSnapshot(Snapshot(Order::new(
5492 key.clone(),
5493 Side::Buy,
5494 None, Decimal::ZERO,
5496 OrderKind::Market,
5497 TimeInForce::ImmediateOrCancel,
5498 OrderState::<AssetNameExchange, InstrumentNameExchange>::Active(
5499 ActiveOrderState::CancelInFlight(CancelInFlight::new(None, Utc::now())),
5500 ),
5501 ))),
5502 );
5503 assert!(
5504 dedup_key_from_event(&event).is_none(),
5505 "CancelInFlight should return None"
5506 );
5507
5508 let event = UnindexedAccountEvent::new(
5510 ExchangeId::BinanceSpot,
5511 AccountEventKind::OrderSnapshot(Snapshot(Order::new(
5512 key,
5513 Side::Buy,
5514 None, Decimal::ONE,
5516 OrderKind::Market,
5517 TimeInForce::ImmediateOrCancel,
5518 OrderState::<AssetNameExchange, InstrumentNameExchange>::Inactive(
5519 InactiveOrderState::FullyFilled(crate::order::state::Filled::new(
5520 OrderId::new("123"),
5521 Utc::now(),
5522 Decimal::ONE, None,
5524 )),
5525 ),
5526 ))),
5527 );
5528 assert!(
5529 dedup_key_from_event(&event).is_none(),
5530 "Inactive state should return None"
5531 );
5532 }
5533
5534 #[test]
5535 fn test_dedup_key_from_event_cancelled_error_returns_none() {
5536 let report = ExecutionReport {
5539 execution_type: Some("REJECTED".to_string()),
5540 reject_reason: Some("INSUFFICIENT_FUNDS".to_string()),
5541 ..make_base_report()
5542 };
5543 let event = sole_event(convert(report)).expect("REJECTED report should produce Some");
5544 assert!(
5545 matches!(&event.kind, AccountEventKind::OrderCancelled(r) if r.state.is_err()),
5546 "prerequisite: event is OrderCancelled with Err"
5547 );
5548 assert!(
5549 dedup_key_from_event(&event).is_none(),
5550 "OrderCancelled with Err state should return None"
5551 );
5552 }
5553
5554 fn user_data_frame<T: serde::Serialize>(event_type: &str, event: &T) -> String {
5564 let mut value = serde_json::to_value(event).expect("event serializes to Value");
5565 value
5566 .as_object_mut()
5567 .expect("event serializes to a JSON object")
5568 .insert(
5569 "e".to_string(),
5570 serde_json::Value::String(event_type.to_string()),
5571 );
5572 serde_json::json!({ "subscriptionId": 0, "event": value }).to_string()
5573 }
5574
5575 #[test]
5576 fn test_convert_user_data_events_execution_report_pushes_to_buf() {
5577 let report = ExecutionReport {
5578 execution_type: Some("NEW".to_string()),
5579 side: Some("BUY".to_string()),
5580 order_type: Some("LIMIT".to_string()),
5581 price: Some("50000.00".to_string()),
5582 order_quantity: Some("0.01".to_string()),
5583 time_in_force: Some("GTC".to_string()),
5584 cumulative_filled_quantity: Some("0".to_string()),
5585 ..make_base_report()
5586 };
5587 let frame = user_data_frame("executionReport", &report);
5588 let mut buf = Vec::new();
5589 let terminated = convert_user_data_events(&frame, &mut buf).terminated;
5590 assert!(
5591 !terminated,
5592 "ExecutionReport should not signal stream termination"
5593 );
5594 assert_eq!(buf.len(), 1, "ExecutionReport should push one event");
5595 assert!(matches!(buf[0].kind, AccountEventKind::OrderSnapshot(_)));
5596 }
5597
5598 #[test]
5599 fn test_convert_user_data_events_account_position_pushes_to_buf() {
5600 let position = binance_sdk::spot::websocket_api::OutboundAccountPosition {
5601 u: Some(1_700_000_000_000),
5602 b_uppercase: Some(vec![make_balance_inner("BTC", "1.0", "0.0")]),
5603 ..Default::default()
5604 };
5605 let frame = user_data_frame("outboundAccountPosition", &position);
5606 let mut buf = Vec::new();
5607 let terminated = convert_user_data_events(&frame, &mut buf).terminated;
5608 assert!(
5609 !terminated,
5610 "OutboundAccountPosition should not signal stream termination"
5611 );
5612 assert_eq!(
5613 buf.len(),
5614 1,
5615 "OutboundAccountPosition should push one balance event"
5616 );
5617 }
5618
5619 #[test]
5620 fn test_convert_user_data_events_balance_update_ignored() {
5621 let update = binance_sdk::spot::websocket_api::BalanceUpdate {
5622 ..Default::default()
5623 };
5624 let frame = user_data_frame("balanceUpdate", &update);
5625 let mut buf = Vec::new();
5626 let terminated = convert_user_data_events(&frame, &mut buf).terminated;
5627 assert!(
5628 !terminated,
5629 "BalanceUpdate should not signal stream termination"
5630 );
5631 assert!(buf.is_empty(), "BalanceUpdate should push no events");
5632 }
5633
5634 #[test]
5635 fn test_convert_user_data_events_stream_terminated_signals_reconnect() {
5636 let frame =
5638 r#"{"subscriptionId":0,"event":{"e":"eventStreamTerminated","E":1700000000000}}"#;
5639 let mut buf = Vec::new();
5640 let terminated = convert_user_data_events(frame, &mut buf).terminated;
5641 assert!(
5642 terminated,
5643 "eventStreamTerminated must signal stream termination"
5644 );
5645 assert!(
5646 buf.is_empty(),
5647 "eventStreamTerminated should push no events"
5648 );
5649 }
5650
5651 #[test]
5652 fn test_convert_user_data_events_unknown_event_ignored() {
5653 for e in ["listStatus", "externalLockUpdate", "someFutureEvent"] {
5656 let frame = format!(
5657 r#"{{"subscriptionId":0,"event":{{"e":"{e}","E":1700000000000,"s":"BTCUSDT"}}}}"#
5658 );
5659 let mut buf = Vec::new();
5660 let terminated = convert_user_data_events(&frame, &mut buf).terminated;
5661 assert!(!terminated, "{e} must not signal termination");
5662 assert!(buf.is_empty(), "{e} should push no events");
5663 }
5664 }
5665
5666 #[test]
5671 fn test_convert_user_data_events_reads_the_ws_api_envelope() {
5672 let fill = r#"{"subscriptionId":3,"event":{"e":"executionReport","E":1700000001000,
5673 "s":"BTCUSDT","c":"client-1","S":"SELL","o":"MARKET","f":"GTC","q":"0.5","p":"0",
5674 "x":"TRADE","X":"FILLED","i":777,"l":"0.5","z":"0.5","L":"40000","n":"0.02",
5675 "N":"USDT","T":1700000001000,"t":888}}"#;
5676 let mut buf = Vec::new();
5677 assert!(!convert_user_data_events(fill, &mut buf).terminated);
5678 assert!(
5679 buf.iter()
5680 .any(|ev| matches!(ev.kind, AccountEventKind::Trade(_))),
5681 "the enveloped fill must produce its trade, got: {buf:?}"
5682 );
5683
5684 let bare = r#"{"e":"executionReport","s":"BTCUSDT","i":777,"x":"NEW","X":"NEW"}"#;
5686 buf.clear();
5687 assert!(!convert_user_data_events(bare, &mut buf).terminated);
5688 assert!(buf.is_empty());
5689 }
5690
5691 #[test]
5694 fn test_convert_user_data_events_returns_the_events_time() {
5695 let new_order = r#"{"subscriptionId":3,"event":{"e":"executionReport","E":1700000002000,
5696 "s":"BTCUSDT","c":"client-1","S":"BUY","o":"LIMIT","f":"GTC","q":"0.5","p":"40000",
5697 "x":"NEW","X":"NEW","i":777,"l":"0","z":"0","L":"0","n":"0","T":1700000001000}}"#;
5698 let mut buf = Vec::new();
5699 let outcome = convert_user_data_events(new_order, &mut buf);
5700 assert_eq!(
5701 outcome.event_time,
5702 DateTime::<Utc>::from_timestamp_millis(1_700_000_002_000)
5703 );
5704
5705 let ack = r#"{"id":"abc","status":200,"result":[]}"#;
5706 assert_eq!(convert_user_data_events(ack, &mut buf).event_time, None);
5707 }
5708
5709 #[test]
5710 fn test_convert_user_data_events_non_user_data_frame_ignored() {
5711 let frame = r#"{"id":"abc","status":200,"result":[]}"#;
5713 let mut buf = Vec::new();
5714 let terminated = convert_user_data_events(frame, &mut buf).terminated;
5715 assert!(
5716 !terminated,
5717 "non-user-data frame must not signal termination"
5718 );
5719 assert!(buf.is_empty(), "non-user-data frame should push no events");
5720 }
5721
5722 #[tokio::test]
5727 async fn test_rate_limit_tracker_not_blocked_initially() {
5728 tokio::time::pause();
5729 let tracker = RateLimitTracker::new(WeightPool::Spot);
5730 tokio::time::timeout(
5732 std::time::Duration::from_millis(1),
5733 tracker.wait_if_blocked(RequestKind::Order),
5734 )
5735 .await
5736 .expect("wait_if_blocked should return immediately when not blocked")
5737 .expect("no cooldown too long to sleep through");
5738 }
5739
5740 #[tokio::test]
5741 async fn test_rate_limit_tracker_blocks_until_deadline() {
5742 tokio::time::pause();
5743 let tracker = RateLimitTracker::new(WeightPool::Spot);
5744 let delay = Duration::from_secs(5);
5745 tracker.on_rate_limited(Some(delay));
5746
5747 assert!(
5749 tokio::time::timeout(
5750 Duration::from_millis(1),
5751 tracker.wait_if_blocked(RequestKind::Order)
5752 )
5753 .await
5754 .is_err(),
5755 "wait_if_blocked should block while cooldown is active"
5756 );
5757
5758 tokio::time::advance(delay + Duration::from_millis(1)).await;
5760 tokio::time::timeout(
5761 Duration::from_millis(1),
5762 tracker.wait_if_blocked(RequestKind::Order),
5763 )
5764 .await
5765 .expect("wait_if_blocked should return after cooldown expires")
5766 .expect("no cooldown too long to sleep through");
5767 }
5768
5769 #[tokio::test]
5770 async fn test_rate_limit_tracker_cooldown_extends_to_max() {
5771 tokio::time::pause();
5772 let tracker = RateLimitTracker::new(WeightPool::Spot);
5773
5774 tracker.on_rate_limited(Some(Duration::from_secs(5)));
5776 tracker.on_rate_limited(Some(Duration::from_secs(10)));
5778
5779 tokio::time::advance(Duration::from_secs(6)).await;
5781 assert!(
5782 tokio::time::timeout(
5783 Duration::from_millis(1),
5784 tracker.wait_if_blocked(RequestKind::Order)
5785 )
5786 .await
5787 .is_err(),
5788 "cooldown should have been extended to 10s"
5789 );
5790
5791 tokio::time::advance(Duration::from_secs(5)).await;
5793 tokio::time::timeout(
5794 Duration::from_millis(1),
5795 tracker.wait_if_blocked(RequestKind::Order),
5796 )
5797 .await
5798 .expect("wait_if_blocked should return after extended cooldown expires")
5799 .expect("no cooldown too long to sleep through");
5800 }
5801
5802 #[tokio::test]
5803 async fn test_rate_limit_tracker_shorter_cooldown_does_not_shorten() {
5804 tokio::time::pause();
5805 let tracker = RateLimitTracker::new(WeightPool::Spot);
5806
5807 tracker.on_rate_limited(Some(Duration::from_secs(10)));
5809 tracker.on_rate_limited(Some(Duration::from_secs(2)));
5810
5811 tokio::time::advance(Duration::from_secs(3)).await;
5813 assert!(
5814 tokio::time::timeout(
5815 Duration::from_millis(1),
5816 tracker.wait_if_blocked(RequestKind::Order)
5817 )
5818 .await
5819 .is_err(),
5820 "shorter on_rate_limited must not shorten existing cooldown"
5821 );
5822 }
5823
5824 #[test]
5829 fn test_classify_ws_order_error_with_wrapped_error_chain() {
5830 let instrument = InstrumentNameExchange::new("BTCUSDT");
5835 let raw = anyhow::anyhow!(WebsocketError::ResponseError {
5836 code: -2010,
5837 message: "insufficient balance".into(),
5838 retry_after: None,
5839 });
5840 assert!(
5842 classify_ws_order_error(&raw, &instrument).is_some(),
5843 "unwrapped ResponseError at root must be detected"
5844 );
5845 let wrapped = raw.context("outer context (e.g. SDK adds context layer)");
5847 assert!(
5848 classify_ws_order_error(&wrapped, &instrument).is_some(),
5849 "context-wrapped ResponseError must still be detected — anyhow::downcast_ref searches the full chain"
5850 );
5851 }
5852
5853 #[tokio::test]
5856 async fn rest_used_weight_near_the_limit_pauses_queries() {
5857 for (used, pauses) in [(5_399, false), (5_400, true)] {
5858 let server = wiremock::MockServer::start().await;
5859 wiremock::Mock::given(wiremock::matchers::method("GET"))
5860 .and(wiremock::matchers::path("/api/v3/openOrders"))
5861 .respond_with(
5862 wiremock::ResponseTemplate::new(200)
5863 .insert_header("x-mbx-used-weight-1m", used.to_string())
5864 .set_body_json(serde_json::json!([])),
5865 )
5866 .mount(&server)
5867 .await;
5868 let rest = Arc::new(SpotRestApi::from_config(
5869 ConfigurationRestApi::builder()
5870 .api_key("key")
5871 .api_secret("secret")
5872 .base_path(server.uri())
5873 .build()
5874 .unwrap(),
5875 ));
5876 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Spot));
5877
5878 fetch_all_open_orders(rest, Arc::clone(&tracker))
5879 .await
5880 .unwrap();
5881
5882 let query_waits = tokio::time::timeout(
5883 Duration::from_millis(50),
5884 tracker.wait_if_blocked(RequestKind::Query),
5885 )
5886 .await
5887 .is_err();
5888 assert_eq!(query_waits, pauses, "used weight {used}");
5889 tokio::time::timeout(
5890 Duration::from_millis(50),
5891 tracker.wait_if_blocked(RequestKind::Order),
5892 )
5893 .await
5894 .expect("orders never wait on the pause")
5895 .expect("no cooldown too long to sleep through");
5896 }
5897 }
5898
5899 fn client_with_ws_api(ws_url: String) -> BinanceSpot {
5904 let mut client = <BinanceSpot as ExecutionClient>::new(
5905 BinanceSpotConfig::new("key".into(), "secret".into())
5906 .with_server_time_sync(false)
5907 .with_pre_send_rules(PreSendRules::Off),
5908 );
5909 client.rest = Arc::new(SpotRestApi::from_config(
5910 ConfigurationRestApi::builder()
5911 .api_key("key")
5912 .api_secret("secret")
5913 .base_path("http://127.0.0.1:9")
5914 .build()
5915 .unwrap(),
5916 ));
5917 client.ws_handle = SpotWsApi::from_config(
5918 ConfigurationWebsocketApi::builder()
5919 .api_key("key")
5920 .api_secret("secret")
5921 .ws_url(ws_url)
5922 .build()
5923 .unwrap(),
5924 );
5925 client
5926 }
5927
5928 fn cancel_request(
5930 instrument: &InstrumentNameExchange,
5931 ) -> OrderRequestCancel<ExchangeId, &InstrumentNameExchange> {
5932 crate::order::OrderEvent {
5933 key: OrderKey {
5934 exchange: ExchangeId::BinanceSpot,
5935 instrument,
5936 strategy: StrategyId::new("strategy"),
5937 cid: ClientOrderId::new("cid"),
5938 },
5939 state: crate::order::request::RequestCancel { id: None },
5940 }
5941 }
5942
5943 async fn ws_server<F, Fut>(serve: F) -> String
5945 where
5946 F: Fn(tokio::net::TcpStream) -> Fut + Send + 'static,
5947 Fut: std::future::Future<Output = ()> + Send + 'static,
5948 {
5949 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
5950 let url = format!("ws://{}", listener.local_addr().unwrap());
5951 tokio::spawn(async move {
5952 while let Ok((stream, _)) = listener.accept().await {
5953 tokio::spawn(serve(stream));
5954 }
5955 });
5956 url
5957 }
5958
5959 #[tokio::test]
5962 async fn cancel_without_a_session_during_a_cooldown_is_rejected_unsent() {
5963 let client = <BinanceSpot as ExecutionClient>::new(
5964 BinanceSpotConfig::new("key".into(), "secret".into()).with_server_time_sync(false),
5965 );
5966 client
5967 .rate_limiter
5968 .on_rate_limited(Some(Duration::from_secs(60)));
5969 let instrument = InstrumentNameExchange::new("BTCUSDT");
5970 let response = client.cancel_order(cancel_request(&instrument)).await;
5971
5972 assert_rate_limited_for(&response.state, Duration::from_secs(60));
5973 assert!(
5974 client.ws_api.read().await.is_none(),
5975 "no session was opened"
5976 );
5977 }
5978
5979 fn assert_rate_limited_for<T: std::fmt::Debug>(
5982 state: &Result<T, UnindexedOrderError>,
5983 wait: Duration,
5984 ) {
5985 let retry_after = match state {
5986 Err(UnindexedOrderError::Rejected(api @ ApiError::RateLimit { .. })) => {
5987 api.retry_after()
5988 }
5989 other => panic!("expected a rate-limit rejection, got {other:?}"),
5990 };
5991 assert!(
5992 retry_after.is_some_and(|got| got <= wait && got > wait - Duration::from_secs(5)),
5993 "expected a wait of about {wait:?}, got {retry_after:?}"
5994 );
5995 }
5996
5997 async fn ws_api_answering(
6000 answer: fn(&serde_json::Value) -> serde_json::Value,
6001 ) -> (String, Arc<std::sync::atomic::AtomicU32>) {
6002 use futures::{SinkExt as _, StreamExt as _};
6003 use tokio_tungstenite::tungstenite::Message;
6004
6005 let requests = Arc::new(std::sync::atomic::AtomicU32::new(0));
6006 let url = ws_server({
6007 let requests = Arc::clone(&requests);
6008 move |stream| {
6009 let requests = Arc::clone(&requests);
6010 async move {
6011 let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else {
6012 return;
6013 };
6014 while let Some(Ok(message)) = ws.next().await {
6015 let Message::Text(text) = message else {
6016 continue;
6017 };
6018 requests.fetch_add(1, Ordering::SeqCst);
6019 let request: serde_json::Value = serde_json::from_str(&text).unwrap();
6020 let response = answer(&request);
6021 if ws.send(Message::text(response.to_string())).await.is_err() {
6022 return;
6023 }
6024 }
6025 }
6026 }
6027 })
6028 .await;
6029 (url, requests)
6030 }
6031
6032 fn cancel_accepted(request: &serde_json::Value) -> serde_json::Value {
6034 serde_json::json!({
6035 "id": request["id"],
6036 "status": 200,
6037 "result": {
6038 "symbol": "BTCUSDT",
6039 "origClientOrderId": "cid",
6040 "orderId": 7,
6041 "transactTime": 1_700_000_000_000_i64,
6042 "executedQty": "0",
6043 },
6044 })
6045 }
6046
6047 #[tokio::test]
6050 async fn cancel_on_an_open_session_during_a_long_cooldown_is_rejected_unsent() {
6051 let (url, requests) = ws_api_answering(cancel_accepted).await;
6052 let client = client_with_ws_api(url);
6053 let instrument = InstrumentNameExchange::new("BTCUSDT");
6054 let response = client.cancel_order(cancel_request(&instrument)).await;
6055 assert!(response.state.is_ok(), "{:?}", response.state);
6056
6057 client
6058 .rate_limiter
6059 .on_rate_limited(Some(Duration::from_secs(60)));
6060 let response = client.cancel_order(cancel_request(&instrument)).await;
6061
6062 assert_rate_limited_for(&response.state, Duration::from_secs(60));
6063 assert_eq!(
6064 requests.load(Ordering::SeqCst),
6065 1,
6066 "the second was not sent"
6067 );
6068 assert!(
6069 client.ws_api.read().await.is_some(),
6070 "the session stays open"
6071 );
6072 }
6073
6074 #[tokio::test]
6078 async fn a_ws_api_ban_reports_its_wait_and_holds_back_later_requests() {
6079 let (url, requests) = ws_api_answering(banned_for_two_minutes).await;
6080 let client = client_with_ws_api(url);
6081 let instrument = InstrumentNameExchange::new("BTCUSDT");
6082
6083 let response = client.cancel_order(cancel_request(&instrument)).await;
6084 assert_rate_limited_for(&response.state, TWO_MINUTE_BAN);
6085 assert_cooldown_about(&client, TWO_MINUTE_BAN);
6086
6087 let response = client.cancel_order(cancel_request(&instrument)).await;
6088 assert_rate_limited_for(&response.state, TWO_MINUTE_BAN);
6089 assert_eq!(
6090 requests.load(Ordering::SeqCst),
6091 1,
6092 "the second was not sent"
6093 );
6094 }
6095
6096 const TWO_MINUTE_BAN: Duration = Duration::from_millis(120_000 + CLOCK_SKEW_SLACK_MS);
6098
6099 fn banned_for_two_minutes(request: &serde_json::Value) -> serde_json::Value {
6101 let until = unix_ms() + 120_000;
6102 serde_json::json!({
6103 "id": request["id"],
6104 "status": 418,
6105 "error": {
6106 "code": -1003,
6107 "msg": format!("Way too much request weight used; IP banned until {until}."),
6108 "data": { "serverTime": unix_ms(), "retryAfter": until },
6109 },
6110 })
6111 }
6112
6113 fn assert_cooldown_about(client: &BinanceSpot, wait: Duration) {
6116 let left = client.rate_limiter.cooldown_remaining();
6117 assert!(
6118 left.is_some_and(|left| left <= wait && left > wait - Duration::from_secs(5)),
6119 "expected a cooldown of about {wait:?}, got {left:?}"
6120 );
6121 }
6122
6123 #[tokio::test]
6126 async fn a_ws_api_ban_on_an_order_holds_back_later_requests() {
6127 let (url, requests) = ws_api_answering(banned_for_two_minutes).await;
6128 let client = client_with_ws_api(url);
6129 let instrument = InstrumentNameExchange::new("BTCUSDT");
6130 let request = crate::order::OrderEvent {
6131 key: OrderKey {
6132 exchange: ExchangeId::BinanceSpot,
6133 instrument: &instrument,
6134 strategy: StrategyId::new("strategy"),
6135 cid: ClientOrderId::new("cid"),
6136 },
6137 state: crate::order::request::RequestOpen::new(
6138 Side::Buy,
6139 Some(Decimal::new(50_000, 0)),
6140 Decimal::new(1, 3),
6141 OrderKind::Limit,
6142 TimeInForce::GoodUntilCancelled { post_only: false },
6143 ),
6144 };
6145
6146 let order = client.open_order(request).await;
6147
6148 let state = match order.state {
6149 OrderState::Inactive(crate::order::state::InactiveOrderState::OpenFailed(error)) => {
6150 Err::<(), _>(error)
6151 }
6152 other => panic!("expected the open to fail, got {other:?}"),
6153 };
6154 assert_rate_limited_for(&state, TWO_MINUTE_BAN);
6155 assert_cooldown_about(&client, TWO_MINUTE_BAN);
6156 assert_eq!(requests.load(Ordering::SeqCst), 1);
6157 }
6158
6159 fn refused_timestamp(request: &serde_json::Value) -> serde_json::Value {
6161 serde_json::json!({
6162 "id": request["id"],
6163 "status": 400,
6164 "error": {
6165 "code": -1021,
6166 "msg": "Timestamp for this request was 1000ms ahead of the server's time.",
6167 },
6168 })
6169 }
6170
6171 #[tokio::test]
6174 async fn a_ws_api_refused_timestamp_reads_the_server_time_again() {
6175 let (url, _) = ws_api_answering(refused_timestamp).await;
6176 let mut client = client_with_ws_api(url);
6177 let reads = Arc::new(std::sync::atomic::AtomicU32::new(0));
6178 let read: crate::client::binance::server_time::ReadServerTime = {
6179 let reads = Arc::clone(&reads);
6180 Arc::new(move || {
6181 reads.fetch_add(1, Ordering::SeqCst);
6182 Box::pin(async { Ok(i64::try_from(unix_ms()).unwrap()) })
6183 })
6184 };
6185 client.rate_limiter = Arc::new(
6186 RateLimitTracker::new(WeightPool::Spot)
6187 .with_server_time(Some(ServerTime::new(SigningClock::default(), read))),
6188 );
6189 let instrument = InstrumentNameExchange::new("BTCUSDT");
6190 let open = crate::order::OrderEvent {
6191 key: OrderKey {
6192 exchange: ExchangeId::BinanceSpot,
6193 instrument: &instrument,
6194 strategy: StrategyId::new("strategy"),
6195 cid: ClientOrderId::new("cid"),
6196 },
6197 state: crate::order::request::RequestOpen::new(
6198 Side::Buy,
6199 Some(Decimal::new(50_000, 0)),
6200 Decimal::new(1, 3),
6201 OrderKind::Limit,
6202 TimeInForce::GoodUntilCancelled { post_only: false },
6203 ),
6204 };
6205
6206 let order = client.open_order(open).await;
6207 assert!(
6208 matches!(order.state, OrderState::Inactive(_)),
6209 "{:?}",
6210 order.state
6211 );
6212 assert_eq!(reads.load(Ordering::SeqCst), 2);
6214
6215 let cancel = client.cancel_order(cancel_request(&instrument)).await;
6216 assert!(cancel.state.is_err(), "{:?}", cancel.state);
6217 assert_eq!(reads.load(Ordering::SeqCst), 3);
6218 }
6219
6220 fn answered_without_an_order_id(request: &serde_json::Value) -> serde_json::Value {
6222 serde_json::json!({
6223 "id": request["id"],
6224 "status": 200,
6225 "result": { "symbol": "BTCUSDT", "clientOrderId": "cid" },
6226 })
6227 }
6228
6229 fn answered_undecodably(request: &serde_json::Value) -> serde_json::Value {
6231 serde_json::json!({
6232 "id": request["id"],
6233 "status": 200,
6234 "result": { "symbol": "BTCUSDT", "orderId": "not a number" },
6235 })
6236 }
6237
6238 #[tokio::test]
6241 async fn an_open_answered_unreadably_is_an_unknown_outcome() {
6242 for answer in [answered_without_an_order_id, answered_undecodably] {
6243 let (url, _) = ws_api_answering(answer).await;
6244 let client = client_with_ws_api(url);
6245 let instrument = InstrumentNameExchange::new("BTCUSDT");
6246 let request = crate::order::OrderEvent {
6247 key: OrderKey {
6248 exchange: ExchangeId::BinanceSpot,
6249 instrument: &instrument,
6250 strategy: StrategyId::new("strategy"),
6251 cid: ClientOrderId::new("cid"),
6252 },
6253 state: crate::order::request::RequestOpen::new(
6254 Side::Buy,
6255 Some(Decimal::new(50_000, 0)),
6256 Decimal::new(1, 3),
6257 OrderKind::Limit,
6258 TimeInForce::GoodUntilCancelled { post_only: false },
6259 ),
6260 };
6261
6262 let order = client.open_order(request).await;
6263
6264 assert!(
6265 matches!(
6266 order.state,
6267 OrderState::Inactive(crate::order::state::InactiveOrderState::OpenFailed(
6268 OrderError::Connectivity(_)
6269 ))
6270 ),
6271 "{:?}",
6272 order.state
6273 );
6274 }
6275 }
6276
6277 #[tokio::test]
6280 async fn a_cancel_answered_unreadably_is_an_unknown_outcome() {
6281 for answer in [answered_without_an_order_id, answered_undecodably] {
6282 let (url, _) = ws_api_answering(answer).await;
6283 let client = client_with_ws_api(url);
6284 let instrument = InstrumentNameExchange::new("BTCUSDT");
6285
6286 let response = client.cancel_order(cancel_request(&instrument)).await;
6287
6288 assert!(
6289 matches!(response.state, Err(OrderError::Connectivity(_))),
6290 "{:?}",
6291 response.state
6292 );
6293 }
6294 }
6295
6296 fn subscribed(request: &serde_json::Value) -> serde_json::Value {
6298 serde_json::json!({
6299 "id": request["id"],
6300 "status": 200,
6301 "result": { "subscriptionId": 0 },
6302 })
6303 }
6304
6305 fn rate_limited_wait<T>(opened: Result<T, UnindexedClientError>) -> Option<Duration> {
6307 match opened {
6308 Err(UnindexedClientError::Api(api @ ApiError::RateLimit { .. })) => api.retry_after(),
6309 Err(other) => panic!("expected a rate limit, got {other:?}"),
6310 Ok(_) => panic!("expected the open to fail"),
6311 }
6312 }
6313
6314 #[tokio::test]
6317 async fn opening_the_account_stream_during_a_long_cooldown_fails_unsent() {
6318 let (url, requests) = ws_api_answering(subscribed).await;
6319 let client = client_with_ws_api(url);
6320 client
6321 .rate_limiter
6322 .on_rate_limited(Some(Duration::from_secs(60)));
6323
6324 let wait = rate_limited_wait(client.account_stream(&[], &[]).await);
6325
6326 assert!(
6327 wait.is_some_and(
6328 |wait| wait <= Duration::from_secs(60) && wait > Duration::from_secs(55)
6329 ),
6330 "{wait:?}"
6331 );
6332 assert_eq!(requests.load(Ordering::SeqCst), 0);
6333 }
6334
6335 #[tokio::test]
6338 async fn a_ban_on_the_account_stream_subscribe_starts_its_cooldown() {
6339 let (url, _) = ws_api_answering(banned_for_two_minutes).await;
6340 let client = client_with_ws_api(url);
6341
6342 let wait = rate_limited_wait(client.account_stream(&[], &[]).await);
6343
6344 assert!(
6345 wait.is_some_and(
6346 |wait| wait <= TWO_MINUTE_BAN && wait > TWO_MINUTE_BAN - Duration::from_secs(5)
6347 ),
6348 "{wait:?}"
6349 );
6350 assert_cooldown_about(&client, TWO_MINUTE_BAN);
6351 }
6352
6353 #[tokio::test]
6356 async fn a_refused_account_stream_handshake_starts_a_cooldown() {
6357 use tokio_tungstenite::tungstenite::http;
6358
6359 let url = ws_server(|stream| async move {
6360 #[allow(clippy::result_large_err)]
6362 let refuse = |_: &http::Request<()>, _| {
6363 Err(http::Response::builder()
6364 .status(http::StatusCode::IM_A_TEAPOT)
6365 .body(None)
6366 .unwrap())
6367 };
6368 let _ = tokio_tungstenite::accept_hdr_async(stream, refuse).await;
6369 })
6370 .await;
6371 let client = client_with_ws_api(url);
6372
6373 let wait = rate_limited_wait(client.account_stream(&[], &[]).await);
6374
6375 assert!(
6377 wait.is_some_and(
6378 |wait| wait <= MIN_IP_BAN && wait > MIN_IP_BAN - Duration::from_secs(5)
6379 ),
6380 "{wait:?}"
6381 );
6382 assert_cooldown_about(&client, MIN_IP_BAN);
6383 }
6384
6385 #[tokio::test]
6388 async fn a_drop_during_a_long_cooldown_ends_the_account_stream_rate_limited() {
6389 use futures::{SinkExt as _, StreamExt as _};
6390 use tokio_tungstenite::tungstenite::Message;
6391
6392 let subscribes = Arc::new(std::sync::atomic::AtomicU32::new(0));
6393 let url = ws_server({
6394 let subscribes = Arc::clone(&subscribes);
6395 move |stream| {
6396 let subscribes = Arc::clone(&subscribes);
6397 async move {
6398 let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else {
6399 return;
6400 };
6401 while let Some(Ok(message)) = ws.next().await {
6404 let Message::Text(text) = message else {
6405 continue;
6406 };
6407 subscribes.fetch_add(1, Ordering::SeqCst);
6408 let request: serde_json::Value = serde_json::from_str(&text).unwrap();
6409 let answer = subscribed(&request).to_string();
6410 if ws.send(Message::text(answer)).await.is_err() {
6411 return;
6412 }
6413 tokio::time::sleep(Duration::from_secs(1)).await;
6414 let _ = ws.close(None).await;
6415 return;
6416 }
6417 }
6418 }
6419 })
6420 .await;
6421 let client = client_with_ws_api(url);
6422 let mut stream = client.account_stream(&[], &[]).await.unwrap();
6423 client
6424 .rate_limiter
6425 .on_rate_limited(Some(Duration::from_secs(60)));
6426
6427 let event = tokio::time::timeout(Duration::from_secs(5), stream.next())
6428 .await
6429 .expect("the stream ends rate-limited within the timeout")
6430 .expect("it reports why before it ends");
6431
6432 match event.kind {
6433 AccountEventKind::StreamTerminated(StreamTerminationReason::RateLimited {
6434 retry_after,
6435 }) => assert!(
6436 retry_after <= Duration::from_secs(60) && retry_after > Duration::from_secs(55),
6437 "{retry_after:?}"
6438 ),
6439 other => panic!("expected a rate-limited termination, got {other:?}"),
6440 }
6441 assert!(stream.next().await.is_none(), "the stream has ended");
6442 assert_eq!(subscribes.load(Ordering::SeqCst), 1);
6443 }
6444
6445 #[tokio::test]
6448 async fn a_reconnect_waits_out_a_short_cooldown() {
6449 use futures::{SinkExt as _, StreamExt as _};
6450 use tokio_tungstenite::tungstenite::Message;
6451
6452 let subscribed_at = Arc::new(parking_lot::Mutex::new(Vec::new()));
6453 let url = ws_server({
6454 let subscribed_at = Arc::clone(&subscribed_at);
6455 move |stream| {
6456 let subscribed_at = Arc::clone(&subscribed_at);
6457 async move {
6458 let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else {
6459 return;
6460 };
6461 while let Some(Ok(message)) = ws.next().await {
6462 let Message::Text(text) = message else {
6463 continue;
6464 };
6465 subscribed_at.lock().push(tokio::time::Instant::now());
6466 let request: serde_json::Value = serde_json::from_str(&text).unwrap();
6467 if ws
6468 .send(Message::text(subscribed(&request).to_string()))
6469 .await
6470 .is_err()
6471 {
6472 return;
6473 }
6474 if subscribed_at.lock().len() == 1 {
6477 tokio::time::sleep(Duration::from_millis(500)).await;
6478 let _ = ws.close(None).await;
6479 return;
6480 }
6481 }
6482 }
6483 }
6484 })
6485 .await;
6486 let client = client_with_ws_api(url);
6487 let _stream = client.account_stream(&[], &[]).await.unwrap();
6488 let cooldown = Duration::from_secs(3);
6489 let cooled = tokio::time::Instant::now() + cooldown;
6491 client.rate_limiter.on_rate_limited(Some(cooldown));
6492
6493 let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
6494 while subscribed_at.lock().len() < 2 {
6495 assert!(tokio::time::Instant::now() < deadline, "no reconnect");
6496 tokio::time::sleep(Duration::from_millis(20)).await;
6497 }
6498
6499 assert!(subscribed_at.lock()[1] >= cooled);
6501 }
6502
6503 #[tokio::test(start_paused = true)]
6508 async fn the_reconnect_budget_resets_only_once_a_connection_proves_stable() {
6509 async fn run(pinging: bool) {
6510 let lifetime = if pinging {
6511 test_ws_api::dropped_then_pinging
6512 } else {
6513 test_ws_api::dropped_then_silent
6514 };
6515 let (url, requests) = test_ws_api::serve(subscribed, lifetime).await;
6516 let client = client_with_ws_api(url);
6517 let _stream = client.account_stream(&[], &[]).await.unwrap();
6518 if pinging {
6519 test_ws_api::assert_reset_once_pinged(&requests, "spot").await;
6520 } else {
6521 test_ws_api::assert_no_reset_when_silent(&requests, "spot").await;
6522 }
6523 }
6524
6525 tokio::join!(run(true), run(false));
6526 }
6527
6528 #[tokio::test]
6531 async fn a_reconnect_restates_the_balances() {
6532 fn dropped_once(n: usize) -> test_ws_api::Lifetime {
6533 match n {
6534 0 => test_ws_api::Lifetime::Silent(Duration::from_millis(100)),
6535 _ => test_ws_api::Lifetime::Open,
6536 }
6537 }
6538 let venue = wiremock::MockServer::start().await;
6539 wiremock::Mock::given(wiremock::matchers::method("GET"))
6540 .and(wiremock::matchers::path("/api/v3/account"))
6541 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
6542 serde_json::json!({ "balances": [
6543 { "asset": "BTC", "free": "1", "locked": "0" },
6544 { "asset": "ETH", "free": "2", "locked": "0" },
6545 ] }),
6546 ))
6547 .mount(&venue)
6548 .await;
6549 let (url, _) = test_ws_api::serve(subscribed, dropped_once).await;
6550 let mut client = client_with_ws_api(url);
6551 client.rest = rest_at(&venue);
6552 let btc = AssetNameExchange::new("BTC");
6553 let mut stream = client
6554 .account_stream(std::slice::from_ref(&btc), &[])
6555 .await
6556 .unwrap();
6557
6558 let restated = tokio::time::timeout(Duration::from_secs(10), async {
6559 loop {
6560 let event = futures::StreamExt::next(&mut stream).await.unwrap();
6561 if let AccountEventKind::BalanceSnapshot(balance) = event.kind {
6562 break balance.0;
6563 }
6564 }
6565 })
6566 .await
6567 .expect("the balances are restated after the reconnect");
6568 assert_eq!(restated.asset, btc);
6569 assert_eq!(restated.balance, Balance::new(Decimal::ONE, Decimal::ONE));
6570 }
6571
6572 #[tokio::test]
6577 async fn a_restatement_is_stamped_no_earlier_than_the_last_balance_delivered() {
6578 use crate::client::binance::shared::REST_BALANCE_LAG;
6579 fn ahead_ms() -> i64 {
6580 (Utc::now() + chrono::Duration::seconds(60)).timestamp_millis()
6581 }
6582 fn position() -> String {
6583 let time = ahead_ms();
6584 serde_json::json!({
6585 "subscriptionId": 0,
6586 "event": {
6587 "e": "outboundAccountPosition", "E": time, "u": time,
6588 "B": [{ "a": "BTC", "f": "1", "l": "0" }],
6589 },
6590 })
6591 .to_string()
6592 }
6593 fn pushed_then_open(n: usize) -> test_ws_api::Lifetime {
6594 match n {
6595 0 => test_ws_api::Lifetime::Pushing(position, Duration::from_millis(300)),
6596 _ => test_ws_api::Lifetime::Open,
6597 }
6598 }
6599 let venue = wiremock::MockServer::start().await;
6600 wiremock::Mock::given(wiremock::matchers::method("GET"))
6601 .and(wiremock::matchers::path("/api/v3/account"))
6602 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
6603 serde_json::json!({ "balances": [{ "asset": "BTC", "free": "1", "locked": "0" }] }),
6604 ))
6605 .mount(&venue)
6606 .await;
6607 let (url, _) = test_ws_api::serve(subscribed, pushed_then_open).await;
6608 let mut client = client_with_ws_api(url);
6609 client.rest = rest_at(&venue);
6610 let mut stream = client.account_stream(&[], &[]).await.unwrap();
6611
6612 let (pushed, restated) = tokio::time::timeout(Duration::from_secs(15), async {
6613 let mut pushed = None;
6614 loop {
6615 let event = futures::StreamExt::next(&mut stream).await.unwrap();
6616 match event.kind {
6617 AccountEventKind::BalanceStreamUpdate(update) => {
6618 pushed = Some((update.0.time_exchange, tokio::time::Instant::now()));
6619 }
6620 AccountEventKind::BalanceSnapshot(balance) => {
6621 break (
6622 pushed.expect("the pushed balance comes first"),
6623 (balance.0.time_exchange, tokio::time::Instant::now()),
6624 );
6625 }
6626 _ => {}
6627 }
6628 }
6629 })
6630 .await
6631 .expect("the balances are restated after the reconnect");
6632
6633 assert_eq!(restated.0, pushed.0);
6634 let allowance = Duration::from_secs(1);
6638 assert!(
6639 restated.1 + allowance >= pushed.1 + REST_BALANCE_LAG.to_std().unwrap(),
6640 "restated {:?} after the balance event, before the REST lag",
6641 restated.1 - pushed.1
6642 );
6643 }
6644
6645 #[tokio::test]
6648 async fn a_cancel_queued_behind_a_refused_handshake_reports_the_cooldown() {
6649 use tokio_tungstenite::tungstenite::http;
6650
6651 let handshakes = Arc::new(std::sync::atomic::AtomicU32::new(0));
6652 let url = ws_server({
6653 let handshakes = Arc::clone(&handshakes);
6654 move |stream| {
6655 let handshakes = Arc::clone(&handshakes);
6656 async move {
6657 handshakes.fetch_add(1, Ordering::SeqCst);
6658 tokio::time::sleep(Duration::from_millis(200)).await;
6660 #[allow(clippy::result_large_err)]
6663 let refuse = |_: &http::Request<()>, _| {
6664 Err(http::Response::builder()
6665 .status(http::StatusCode::TOO_MANY_REQUESTS)
6666 .body(None)
6667 .unwrap())
6668 };
6669 let _ = tokio_tungstenite::accept_hdr_async(stream, refuse).await;
6670 }
6671 }
6672 })
6673 .await;
6674 let client = client_with_ws_api(url);
6675 let instrument = InstrumentNameExchange::new("BTCUSDT");
6676
6677 let (first, second) = tokio::join!(
6678 client.cancel_order(cancel_request(&instrument)),
6679 client.cancel_order(cancel_request(&instrument)),
6680 );
6681
6682 assert_eq!(handshakes.load(Ordering::SeqCst), 1);
6683 let waits: Vec<_> = [&first.state, &second.state]
6684 .into_iter()
6685 .map(|state| match state {
6686 Err(UnindexedOrderError::Rejected(api @ ApiError::RateLimit { .. })) => {
6687 api.retry_after()
6688 }
6689 other => panic!("expected a rate-limit rejection, got {other:?}"),
6690 })
6691 .collect();
6692 let default = Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS);
6694 assert!(
6695 waits.contains(&None)
6696 && waits
6697 .iter()
6698 .flatten()
6699 .any(|wait| { *wait <= default && *wait > default - Duration::from_secs(5) }),
6700 "{waits:?}"
6701 );
6702 }
6703
6704 fn rest_against(server: &wiremock::MockServer) -> (Arc<RestApi>, Arc<RateLimitTracker>) {
6706 let rest = Arc::new(SpotRestApi::from_config(
6707 ConfigurationRestApi::builder()
6708 .api_key("key")
6709 .api_secret("secret")
6710 .base_path(server.uri())
6711 .build()
6712 .unwrap(),
6713 ));
6714 (rest, Arc::new(RateLimitTracker::new(WeightPool::Spot)))
6715 }
6716
6717 async fn mount_refusal(
6720 server: &wiremock::MockServer,
6721 status: u16,
6722 retry_after: Option<&str>,
6723 times: u64,
6724 ) {
6725 let mut refusal = wiremock::ResponseTemplate::new(status).set_body_json(
6726 serde_json::json!({ "code": -1003, "msg": "Too much request weight used." }),
6727 );
6728 if let Some(retry_after) = retry_after {
6729 refusal = refusal.insert_header("retry-after", retry_after);
6730 }
6731 wiremock::Mock::given(wiremock::matchers::method("GET"))
6732 .and(wiremock::matchers::path("/api/v3/openOrders"))
6733 .respond_with(refusal)
6734 .up_to_n_times(times)
6735 .mount(server)
6736 .await;
6737 }
6738
6739 fn query_retry_after<T: std::fmt::Debug>(
6741 result: &Result<T, UnindexedClientError>,
6742 ) -> Option<Duration> {
6743 match result {
6744 Err(UnindexedClientError::Api(api @ ApiError::RateLimit { .. })) => api.retry_after(),
6745 other => panic!("expected a rate-limit refusal, got {other:?}"),
6746 }
6747 }
6748
6749 #[tokio::test]
6752 async fn a_rest_ban_is_not_retried_and_holds_back_later_requests() {
6753 let server = wiremock::MockServer::start().await;
6754 mount_refusal(&server, 418, Some("120"), u64::MAX).await;
6755 let (rest, tracker) = rest_against(&server);
6756
6757 let result = fetch_all_open_orders(Arc::clone(&rest), Arc::clone(&tracker)).await;
6758 assert_eq!(query_retry_after(&result), Some(Duration::from_secs(120)));
6759 assert!(
6760 tracker
6761 .cooldown_remaining()
6762 .is_some_and(|left| left > Duration::from_secs(115)),
6763 "{:?}",
6764 tracker.cooldown_remaining()
6765 );
6766
6767 let result = fetch_all_open_orders(rest, tracker).await;
6768 assert!(
6769 query_retry_after(&result).is_some_and(
6770 |left| left > Duration::from_secs(115) && left <= Duration::from_secs(120)
6771 ),
6772 "{result:?}"
6773 );
6774 assert_eq!(server.received_requests().await.unwrap().len(), 1);
6775 }
6776
6777 #[tokio::test]
6780 async fn a_rest_ban_without_retry_after_is_not_retried() {
6781 let server = wiremock::MockServer::start().await;
6782 mount_refusal(&server, 418, None, u64::MAX).await;
6783 let (rest, tracker) = rest_against(&server);
6784
6785 let result = fetch_all_open_orders(rest, Arc::clone(&tracker)).await;
6786 assert_eq!(query_retry_after(&result), None);
6787 assert_eq!(server.received_requests().await.unwrap().len(), 1);
6788 assert!(
6789 tracker
6790 .cooldown_remaining()
6791 .is_some_and(|left| left > Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS - 5)),
6792 "{:?}",
6793 tracker.cooldown_remaining()
6794 );
6795 }
6796
6797 #[tokio::test]
6799 async fn a_rest_throttle_with_a_long_retry_after_is_not_retried() {
6800 let server = wiremock::MockServer::start().await;
6801 mount_refusal(&server, 429, Some("30"), u64::MAX).await;
6802 let (rest, tracker) = rest_against(&server);
6803
6804 let result = fetch_all_open_orders(rest, tracker).await;
6805 assert_eq!(query_retry_after(&result), Some(Duration::from_secs(30)));
6806 assert_eq!(server.received_requests().await.unwrap().len(), 1);
6807 }
6808
6809 #[tokio::test]
6812 async fn a_rest_throttle_with_a_short_retry_after_is_retried_after_it() {
6813 let server = wiremock::MockServer::start().await;
6814 mount_refusal(&server, 429, Some("2"), 1).await;
6815 wiremock::Mock::given(wiremock::matchers::method("GET"))
6816 .and(wiremock::matchers::path("/api/v3/openOrders"))
6817 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
6818 .mount(&server)
6819 .await;
6820 let (rest, tracker) = rest_against(&server);
6821
6822 let started = std::time::Instant::now();
6823 let orders = fetch_all_open_orders(rest, tracker).await.unwrap();
6824
6825 assert!(orders.is_empty());
6826 assert_eq!(server.received_requests().await.unwrap().len(), 2);
6827 assert!(
6828 started.elapsed() >= Duration::from_secs(2),
6829 "{:?}",
6830 started.elapsed()
6831 );
6832 }
6833
6834 #[tokio::test]
6837 async fn a_handshake_refused_with_429_starts_a_cooldown() {
6838 use tokio_tungstenite::tungstenite::http;
6839
6840 let url = ws_server(|stream| async move {
6841 #[allow(clippy::result_large_err)]
6843 let refuse = |_: &http::Request<()>, _| {
6844 Err(http::Response::builder()
6845 .status(http::StatusCode::TOO_MANY_REQUESTS)
6846 .body(None)
6847 .unwrap())
6848 };
6849 let _ = tokio_tungstenite::accept_hdr_async(stream, refuse).await;
6850 })
6851 .await;
6852 let client = client_with_ws_api(url);
6853 let instrument = InstrumentNameExchange::new("BTCUSDT");
6854
6855 let response = client.cancel_order(cancel_request(&instrument)).await;
6856
6857 assert!(
6858 matches!(
6859 response.state,
6860 Err(UnindexedOrderError::Rejected(ApiError::RateLimit { .. }))
6861 ),
6862 "{:?}",
6863 response.state
6864 );
6865 assert!(client.rate_limiter.is_blocked());
6866 }
6867
6868 #[tokio::test]
6871 async fn a_handshake_refused_with_418_starts_the_shortest_ban() {
6872 use tokio_tungstenite::tungstenite::http;
6873
6874 let url = ws_server(|stream| async move {
6875 #[allow(clippy::result_large_err)]
6877 let refuse = |_: &http::Request<()>, _| {
6878 Err(http::Response::builder()
6879 .status(http::StatusCode::IM_A_TEAPOT)
6880 .body(None)
6881 .unwrap())
6882 };
6883 let _ = tokio_tungstenite::accept_hdr_async(stream, refuse).await;
6884 })
6885 .await;
6886 let client = client_with_ws_api(url);
6887 let instrument = InstrumentNameExchange::new("BTCUSDT");
6888
6889 let response = client.cancel_order(cancel_request(&instrument)).await;
6890
6891 assert_rate_limited_for(&response.state, MIN_IP_BAN);
6892 assert_cooldown_about(&client, MIN_IP_BAN);
6893 }
6894
6895 #[tokio::test]
6898 async fn a_ws_api_response_near_the_weight_limit_pauses_queries() {
6899 use futures::{SinkExt as _, StreamExt as _};
6900 use tokio_tungstenite::tungstenite::Message;
6901
6902 let url = ws_server(|stream| async move {
6903 let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else {
6904 return;
6905 };
6906 while let Some(Ok(message)) = ws.next().await {
6907 let Message::Text(text) = message else {
6908 continue;
6909 };
6910 let request: serde_json::Value = serde_json::from_str(&text).unwrap();
6911 let response = serde_json::json!({
6912 "id": request["id"],
6913 "status": 200,
6914 "result": {
6915 "symbol": "BTCUSDT",
6916 "origClientOrderId": "cid",
6917 "orderId": 7,
6918 "transactTime": 1_700_000_000_000_i64,
6919 "executedQty": "0",
6920 },
6921 "rateLimits": [{
6922 "rateLimitType": "REQUEST_WEIGHT",
6923 "interval": "MINUTE",
6924 "intervalNum": 1,
6925 "limit": 1_000,
6926 "count": 950,
6927 }],
6928 });
6929 if ws.send(Message::text(response.to_string())).await.is_err() {
6930 return;
6931 }
6932 }
6933 })
6934 .await;
6935 let client = client_with_ws_api(url);
6936 let instrument = InstrumentNameExchange::new("BTCUSDT");
6937
6938 let response = client.cancel_order(cancel_request(&instrument)).await;
6939
6940 assert!(response.state.is_ok(), "{:?}", response.state);
6941 assert_eq!(client.rate_limiter.weight_limit(), 1_000);
6942 assert!(
6943 tokio::time::timeout(
6944 Duration::from_millis(50),
6945 client.rate_limiter.wait_if_blocked(RequestKind::Query),
6946 )
6947 .await
6948 .is_err(),
6949 "queries pause"
6950 );
6951 tokio::time::timeout(
6952 Duration::from_millis(50),
6953 client.rate_limiter.wait_if_blocked(RequestKind::Order),
6954 )
6955 .await
6956 .expect("orders never wait on the pause")
6957 .expect("no cooldown too long to sleep through");
6958 }
6959
6960 async fn ws_server_reporting_close(
6963 result: serde_json::Value,
6964 ) -> (String, mpsc::UnboundedReceiver<()>) {
6965 use futures::{SinkExt as _, StreamExt as _};
6966 use tokio_tungstenite::tungstenite::Message;
6967
6968 let (closed_tx, closed_rx) = mpsc::unbounded_channel();
6969 let url = ws_server(move |stream| {
6970 let result = result.clone();
6971 let closed_tx = closed_tx.clone();
6972 async move {
6973 let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else {
6974 return;
6975 };
6976 while let Some(Ok(message)) = ws.next().await {
6977 match message {
6978 Message::Text(text) => {
6979 let request: serde_json::Value = serde_json::from_str(&text).unwrap();
6980 let response = serde_json::json!({
6981 "id": request["id"],
6982 "status": 200,
6983 "result": result,
6984 });
6985 if ws.send(Message::text(response.to_string())).await.is_err() {
6986 return;
6987 }
6988 }
6989 Message::Close(_) => {
6990 let _ = closed_tx.send(());
6991 return;
6992 }
6993 _ => {}
6994 }
6995 }
6996 }
6997 })
6998 .await;
6999 (url, closed_rx)
7000 }
7001
7002 #[tokio::test]
7008 async fn dropping_the_account_stream_closes_its_connection() {
7009 for running in [Duration::ZERO, Duration::from_millis(100)] {
7010 let (url, mut closed) =
7011 ws_server_reporting_close(serde_json::json!({ "subscriptionId": 0 })).await;
7012 let client = client_with_ws_api(url);
7013 let holders = client.known_live.holders();
7014
7015 let stream = client.account_stream(&[], &[]).await.unwrap();
7016 tokio::time::sleep(running).await;
7017 drop(stream);
7018
7019 tokio::time::timeout(Duration::from_secs(5), closed.recv())
7020 .await
7021 .expect("the dropped stream closes its connection")
7022 .unwrap();
7023 for _ in 0..10 {
7025 tokio::task::yield_now().await;
7026 }
7027 assert_eq!(
7028 client.known_live.holders(),
7029 holders,
7030 "the stream's state is released"
7031 );
7032 }
7033 }
7034
7035 #[tokio::test]
7038 async fn dropping_the_last_client_closes_its_order_session() {
7039 let (url, mut closed) = ws_server_reporting_close(serde_json::json!({
7040 "symbol": "BTCUSDT",
7041 "origClientOrderId": "cid",
7042 "orderId": 7,
7043 "transactTime": 1_700_000_000_000_i64,
7044 "executedQty": "0",
7045 }))
7046 .await;
7047 let client = client_with_ws_api(url);
7048 let clone = client.clone();
7049 let instrument = InstrumentNameExchange::new("BTCUSDT");
7050 let response = client.cancel_order(cancel_request(&instrument)).await;
7051 assert!(response.state.is_ok(), "{:?}", response.state);
7052
7053 drop(client);
7054 assert!(
7055 tokio::time::timeout(Duration::from_millis(200), closed.recv())
7056 .await
7057 .is_err(),
7058 "a clone still holds the session"
7059 );
7060
7061 drop(clone);
7062 tokio::time::timeout(Duration::from_secs(5), closed.recv())
7063 .await
7064 .expect("the last client's drop closes the session")
7065 .unwrap();
7066 }
7067
7068 #[tokio::test]
7071 async fn fill_recovery_does_not_wait_for_the_weight_pause() {
7072 let server = wiremock::MockServer::start().await;
7073 wiremock::Mock::given(wiremock::matchers::method("GET"))
7074 .and(wiremock::matchers::path("/api/v3/myTrades"))
7075 .respond_with(
7077 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([{
7078 "symbol": "BTCUSDT",
7079 "id": 1,
7080 "orderId": 7,
7081 "price": "100",
7082 "qty": "1",
7083 "commission": "0",
7084 "commissionAsset": "USDT",
7085 "time": Utc::now().timestamp_millis(),
7086 "isBuyer": true,
7087 "isMaker": false,
7088 }])),
7089 )
7090 .expect(2)
7091 .mount(&server)
7092 .await;
7093 let rest = Arc::new(SpotRestApi::from_config(
7094 ConfigurationRestApi::builder()
7095 .api_key("key")
7096 .api_secret("secret")
7097 .base_path(server.uri())
7098 .build()
7099 .unwrap(),
7100 ));
7101 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Spot));
7102 tracker.throttle(Duration::from_secs(60));
7103 let (tx, _rx) = mpsc::unbounded_channel();
7104 let mut unrecovered = UnrecoveredFills::default();
7105 unrecovered.open(
7106 &[InstrumentNameExchange::new("BTCUSDT")],
7107 Utc::now() - chrono::Duration::minutes(1),
7108 Utc::now(),
7109 );
7110
7111 tokio::time::timeout(
7113 Duration::from_millis(800),
7114 recover_fills(
7115 &rest,
7116 &tracker,
7117 &mut unrecovered,
7118 &tx,
7119 &new_dedup_cache(),
7120 &KnownLiveOrders::shared(ExchangeId::BinanceSpot),
7121 ),
7122 )
7123 .await
7124 .expect("recovery does not wait for the pause");
7125 }
7126
7127 #[tokio::test]
7129 async fn a_gap_whose_every_read_fails_is_reported_when_given_up() {
7130 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7131 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7132 let server = wiremock::MockServer::start().await;
7133 wiremock::Mock::given(wiremock::matchers::method("GET"))
7134 .and(wiremock::matchers::path("/api/v3/myTrades"))
7135 .respond_with(
7136 wiremock::ResponseTemplate::new(400)
7137 .set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
7138 )
7139 .mount(&server)
7140 .await;
7141 let rest = Arc::new(SpotRestApi::from_config(
7142 ConfigurationRestApi::builder()
7143 .api_key("key")
7144 .api_secret("secret")
7145 .base_path(server.uri())
7146 .build()
7147 .expect("valid config"),
7148 ));
7149 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Spot));
7150 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
7151 let dedup = new_dedup_cache();
7152 let btc = InstrumentNameExchange::new("BTCUSDT");
7153 let mut unrecovered = UnrecoveredFills::default();
7154 unrecovered.open(std::slice::from_ref(&btc), disconnect, reconnect);
7155
7156 for _ in 0..=crate::client::order_recovery::MAX_GAP_RETRIES {
7157 recover_fills(
7158 &rest,
7159 &tracker,
7160 &mut unrecovered,
7161 &tx,
7162 &dedup,
7163 &KnownLiveOrders::shared(ExchangeId::BinanceSpot),
7164 )
7165 .await;
7166 unrecovered.make_due();
7167 }
7168
7169 assert!(unrecovered.is_empty(), "the gap is given up");
7170 let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7171 let [event] = forwarded.as_slice() else {
7172 panic!("only the give-up is sent: {forwarded:?}");
7173 };
7174 let AccountEventKind::FillRecoveryGaveUp(gave_up) = &event.kind else {
7175 panic!("expected FillRecoveryGaveUp, got {event:?}");
7176 };
7177 assert_eq!(event.exchange, ExchangeId::BinanceSpot);
7178 assert_eq!(
7179 gave_up.scope,
7180 crate::FillRecoveryScope::Instruments(vec![btc])
7181 );
7182 assert_eq!(
7183 gave_up.start.timestamp_millis(),
7184 disconnect.timestamp_millis()
7185 );
7186 assert_eq!(
7187 gave_up.attempts,
7188 crate::client::order_recovery::MAX_GAP_RETRIES + 1
7189 );
7190 assert!(
7191 matches!(&gave_up.reason, FillRecoveryFailure::Request(reason) if reason.contains("rejected")),
7192 "the last read's error: {:?}",
7193 gave_up.reason
7194 );
7195 }
7196
7197 #[tokio::test]
7200 async fn a_failed_recovery_keeps_its_closed_gap_for_the_retry() {
7201 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7202 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7203 let trade = |id: i64, time: DateTime<Utc>| {
7204 serde_json::json!({
7205 "symbol": "BTCUSDT", "id": id, "orderId": id, "price": "100", "qty": "1",
7206 "commission": "0", "commissionAsset": "USDT", "time": time.timestamp_millis(),
7207 "isBuyer": true, "isMaker": false, "isBestMatch": true,
7208 })
7209 };
7210 let server = wiremock::MockServer::start().await;
7211 wiremock::Mock::given(wiremock::matchers::method("GET"))
7212 .and(wiremock::matchers::path("/api/v3/myTrades"))
7213 .respond_with(
7214 wiremock::ResponseTemplate::new(400)
7215 .set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
7216 )
7217 .up_to_n_times(1)
7218 .with_priority(1)
7219 .mount(&server)
7220 .await;
7221 wiremock::Mock::given(wiremock::matchers::method("GET"))
7222 .and(wiremock::matchers::path("/api/v3/myTrades"))
7223 .respond_with(
7224 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([
7225 trade(1, disconnect + chrono::Duration::minutes(1)),
7226 trade(2, Utc::now() - chrono::Duration::minutes(1)),
7227 ])),
7228 )
7229 .mount(&server)
7230 .await;
7231 let rest = Arc::new(SpotRestApi::from_config(
7232 ConfigurationRestApi::builder()
7233 .api_key("key")
7234 .api_secret("secret")
7235 .base_path(server.uri())
7236 .build()
7237 .unwrap(),
7238 ));
7239 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Spot));
7240 let (tx, mut rx) = mpsc::unbounded_channel();
7241 let dedup = new_dedup_cache();
7242 let mut unrecovered = UnrecoveredFills::default();
7243 unrecovered.open(
7244 &[InstrumentNameExchange::new("BTCUSDT")],
7245 disconnect,
7246 reconnect,
7247 );
7248
7249 recover_fills(
7250 &rest,
7251 &tracker,
7252 &mut unrecovered,
7253 &tx,
7254 &dedup,
7255 &KnownLiveOrders::shared(ExchangeId::BinanceSpot),
7256 )
7257 .await;
7258 assert!(!unrecovered.is_empty(), "the failed gap is kept");
7259 assert!(rx.try_recv().is_err(), "nothing forwarded");
7260 assert!(
7261 unrecovered.due(tokio::time::Instant::now()).is_empty(),
7262 "and waits for its retry"
7263 );
7264
7265 unrecovered.make_due();
7266 recover_fills(
7267 &rest,
7268 &tracker,
7269 &mut unrecovered,
7270 &tx,
7271 &dedup,
7272 &KnownLiveOrders::shared(ExchangeId::BinanceSpot),
7273 )
7274 .await;
7275 assert!(unrecovered.is_empty(), "the retry recovered it");
7276 let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7277 assert_eq!(
7278 forwarded.len(),
7279 1,
7280 "only the fill inside the gap: {forwarded:?}"
7281 );
7282
7283 let start_times: Vec<String> = server
7284 .received_requests()
7285 .await
7286 .unwrap()
7287 .iter()
7288 .filter_map(|request| {
7289 request
7290 .url
7291 .query_pairs()
7292 .find(|(key, _)| key == "startTime")
7293 .map(|(_, value)| value.into_owned())
7294 })
7295 .collect();
7296 let start = disconnect.timestamp_millis().to_string();
7297 assert_eq!(start_times, [start.clone(), start]);
7298 }
7299
7300 #[tokio::test]
7303 async fn recovered_fills_carry_their_orders_client_order_ids() {
7304 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7305 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7306 let trade = |id: i64| {
7307 serde_json::json!({
7308 "symbol": "BTCUSDT", "id": id, "orderId": id, "price": "100", "qty": "1",
7309 "commission": "0", "commissionAsset": "USDT",
7310 "time": (disconnect + chrono::Duration::minutes(1)).timestamp_millis(),
7311 "isBuyer": true, "isMaker": false, "isBestMatch": true,
7312 })
7313 };
7314 let server = wiremock::MockServer::start().await;
7315 for id in [1, 2] {
7317 wiremock::Mock::given(wiremock::matchers::method("GET"))
7318 .and(wiremock::matchers::path("/api/v3/myTrades"))
7319 .and(wiremock::matchers::query_param("orderId", id.to_string()))
7320 .respond_with(
7321 wiremock::ResponseTemplate::new(200)
7322 .set_body_json(serde_json::json!([trade(id)])),
7323 )
7324 .with_priority(1)
7325 .mount(&server)
7326 .await;
7327 }
7328 wiremock::Mock::given(wiremock::matchers::method("GET"))
7329 .and(wiremock::matchers::path("/api/v3/myTrades"))
7330 .respond_with(
7331 wiremock::ResponseTemplate::new(200)
7332 .set_body_json(serde_json::json!([trade(1), trade(2)])),
7333 )
7334 .mount(&server)
7335 .await;
7336 wiremock::Mock::given(wiremock::matchers::method("GET"))
7337 .and(wiremock::matchers::path("/api/v3/order"))
7338 .and(wiremock::matchers::query_param("orderId", "1"))
7339 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
7340 serde_json::json!({"symbol": "BTCUSDT", "orderId": 1, "clientOrderId": "sweep-1"}),
7341 ))
7342 .mount(&server)
7343 .await;
7344 wiremock::Mock::given(wiremock::matchers::method("GET"))
7345 .and(wiremock::matchers::path("/api/v3/order"))
7346 .and(wiremock::matchers::query_param("orderId", "2"))
7347 .respond_with(
7348 wiremock::ResponseTemplate::new(400).set_body_json(
7349 serde_json::json!({"code": -2013, "msg": "Order does not exist."}),
7350 ),
7351 )
7352 .mount(&server)
7353 .await;
7354 let rest = Arc::new(SpotRestApi::from_config(
7355 ConfigurationRestApi::builder()
7356 .api_key("key")
7357 .api_secret("secret")
7358 .base_path(server.uri())
7359 .build()
7360 .unwrap(),
7361 ));
7362 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Spot));
7363 let (tx, mut rx) = mpsc::unbounded_channel();
7364 let mut unrecovered = UnrecoveredFills::default();
7365 unrecovered.open(
7366 &[InstrumentNameExchange::new("BTCUSDT")],
7367 disconnect,
7368 reconnect,
7369 );
7370
7371 recover_fills(
7372 &rest,
7373 &tracker,
7374 &mut unrecovered,
7375 &tx,
7376 &new_dedup_cache(),
7377 &KnownLiveOrders::shared(ExchangeId::BinanceSpot),
7378 )
7379 .await;
7380
7381 let mut cids: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
7382 .filter_map(|event| match event.kind {
7383 AccountEventKind::Trade(trade) => Some((
7384 trade.id.0.to_string(),
7385 trade.client_order_id,
7386 trade.order_filled_quantity,
7387 )),
7388 _ => None,
7389 })
7390 .collect();
7391 cids.sort();
7392 assert_eq!(
7394 cids,
7395 [
7396 (
7397 "1".to_string(),
7398 Some(ClientOrderId::new("sweep-1")),
7399 Some(Decimal::ONE)
7400 ),
7401 ("2".to_string(), None, Some(Decimal::ONE)),
7402 ]
7403 );
7404 let order_reads = server
7405 .received_requests()
7406 .await
7407 .unwrap()
7408 .iter()
7409 .filter(|request| request.url.path() == "/api/v3/order")
7410 .count();
7411 assert_eq!(order_reads, 2, "one read per order");
7412 }
7413
7414 #[tokio::test]
7417 async fn a_span_walk_pages_by_id_and_stops_past_the_end() {
7418 let start = Utc::now().timestamp_millis() - 600_000;
7419 let end = start + 10_000;
7420 let trade = |id: i64, time: i64| {
7421 serde_json::json!({
7422 "symbol": "BTCUSDT", "id": id, "orderId": id, "price": "100", "qty": "1",
7423 "commission": "0", "commissionAsset": "USDT", "time": time,
7424 "isBuyer": true, "isMaker": false, "isBestMatch": true,
7425 })
7426 };
7427 let server = wiremock::MockServer::start().await;
7428 wiremock::Mock::given(wiremock::matchers::method("GET"))
7429 .and(wiremock::matchers::path("/api/v3/myTrades"))
7430 .and(wiremock::matchers::query_param("fromId", "1001"))
7431 .respond_with(
7433 wiremock::ResponseTemplate::new(200).set_body_json(
7434 (1_001..=2_000)
7435 .map(|id| trade(id, if id == 1_001 { end } else { end + 1 }))
7436 .collect::<Vec<_>>(),
7437 ),
7438 )
7439 .with_priority(1)
7440 .mount(&server)
7441 .await;
7442 let full_page: Vec<_> = (1..=1_000).map(|id| trade(id, start + id)).collect();
7443 wiremock::Mock::given(wiremock::matchers::method("GET"))
7444 .and(wiremock::matchers::path("/api/v3/myTrades"))
7445 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(full_page))
7446 .mount(&server)
7447 .await;
7448 let rest = Arc::new(SpotRestApi::from_config(
7449 ConfigurationRestApi::builder()
7450 .api_key("key")
7451 .api_secret("secret")
7452 .base_path(server.uri())
7453 .build()
7454 .unwrap(),
7455 ));
7456
7457 let read = paginate_my_trades(
7458 &rest,
7459 &Arc::new(RateLimitTracker::new(WeightPool::Spot)),
7460 &InstrumentNameExchange::new("BTCUSDT"),
7461 MyTradesFrom::Span { start, end },
7462 RequestKind::Query,
7463 )
7464 .await
7465 .unwrap();
7466
7467 let ids: Vec<_> = read.iter().filter_map(|t| t.id).collect();
7468 assert_eq!(ids, (1..=1_001).collect::<Vec<_>>());
7469 assert_eq!(server.received_requests().await.unwrap().len(), 2);
7470 }
7471
7472 #[tokio::test]
7475 async fn fetch_trades_returns_only_the_span() {
7476 let start = DateTime::from_timestamp_millis(
7478 (Utc::now() - chrono::Duration::minutes(10)).timestamp_millis(),
7479 )
7480 .unwrap()
7481 + chrono::Duration::microseconds(500);
7482 let end = start + chrono::Duration::seconds(10);
7483 let (start_ms, end_ms) = (start.timestamp_millis(), end.timestamp_millis());
7484 let trade = |id: i64, time: i64| {
7485 serde_json::json!({
7486 "symbol": "BTCUSDT", "id": id, "orderId": id, "price": "100", "qty": "1",
7487 "commission": "0", "commissionAsset": "USDT", "time": time,
7488 "isBuyer": true, "isMaker": false, "isBestMatch": true,
7489 })
7490 };
7491 let server = wiremock::MockServer::start().await;
7492 wiremock::Mock::given(wiremock::matchers::method("GET"))
7493 .and(wiremock::matchers::path("/api/v3/myTrades"))
7494 .and(wiremock::matchers::query_param(
7495 "startTime",
7496 start_ms.to_string(),
7497 ))
7498 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(vec![
7499 trade(1, start_ms),
7500 trade(2, start_ms + 1),
7501 trade(3, end_ms),
7502 trade(4, end_ms + 1),
7503 ]))
7504 .mount(&server)
7505 .await;
7506 let mut client = <BinanceSpot as ExecutionClient>::new(
7507 BinanceSpotConfig::new("key".into(), "secret".into()).with_server_time_sync(false),
7508 );
7509 client.rest = Arc::new(SpotRestApi::from_config(
7510 ConfigurationRestApi::builder()
7511 .api_key("key")
7512 .api_secret("secret")
7513 .base_path(server.uri())
7514 .build()
7515 .unwrap(),
7516 ));
7517 let btc = [InstrumentNameExchange::new("BTCUSDT")];
7518
7519 let read = client.fetch_trades(start, end, &btc).await.unwrap();
7520 let ids: Vec<_> = read.trades.iter().map(|t| t.id.0.as_str()).collect();
7521 assert_eq!(ids, ["2", "3"]);
7522 assert_eq!(read.resume, None, "spot reads a span whole");
7523
7524 let empty = client.fetch_trades(end, start, &btc).await.unwrap();
7525 assert!(empty.trades.is_empty() && empty.resume.is_none());
7526 assert_eq!(server.received_requests().await.unwrap().len(), 1);
7527 }
7528
7529 #[tokio::test]
7531 async fn a_gap_whose_every_read_times_out_is_reported_when_given_up() {
7532 let server = wiremock::MockServer::start().await;
7533 wiremock::Mock::given(wiremock::matchers::method("GET"))
7534 .and(wiremock::matchers::path("/api/v3/myTrades"))
7535 .respond_with(
7536 wiremock::ResponseTemplate::new(200)
7537 .set_body_json(serde_json::json!([]))
7538 .set_delay(Duration::from_secs(3_600)),
7539 )
7540 .mount(&server)
7541 .await;
7542 let rest = Arc::new(SpotRestApi::from_config(
7543 ConfigurationRestApi::builder()
7544 .api_key("key")
7545 .api_secret("secret")
7546 .base_path(server.uri())
7547 .timeout(3_600_000_u64)
7548 .retries(0_u32)
7549 .build()
7550 .expect("valid config"),
7551 ));
7552 let btc = InstrumentNameExchange::new("BTCUSDT");
7553 let mut unrecovered = UnrecoveredFills::default();
7554 unrecovered.open(
7555 std::slice::from_ref(&btc),
7556 Utc::now() - chrono::Duration::minutes(10),
7557 Utc::now(),
7558 );
7559 let (tx, mut rx) = mpsc::unbounded_channel();
7560 tokio::time::pause();
7562
7563 for _ in 0..=crate::client::order_recovery::MAX_GAP_RETRIES {
7564 recover_fills(
7565 &rest,
7566 &Arc::new(RateLimitTracker::new(WeightPool::Spot)),
7567 &mut unrecovered,
7568 &tx,
7569 &new_dedup_cache(),
7570 &KnownLiveOrders::shared(ExchangeId::BinanceSpot),
7571 )
7572 .await;
7573 unrecovered.make_due();
7574 }
7575
7576 assert!(unrecovered.is_empty(), "the gap is given up");
7577 let Ok(event) = rx.try_recv() else {
7578 panic!("the given-up gap is reported");
7579 };
7580 let AccountEventKind::FillRecoveryGaveUp(gave_up) = &event.kind else {
7581 panic!("expected FillRecoveryGaveUp, got {event:?}");
7582 };
7583 assert_eq!(
7584 gave_up.scope,
7585 crate::FillRecoveryScope::Instruments(vec![btc])
7586 );
7587 assert_eq!(
7588 gave_up.reason,
7589 FillRecoveryFailure::TimedOut {
7590 timeout_secs: FILL_RECOVERY_TIMEOUT_SECS
7591 }
7592 );
7593 assert!(rx.try_recv().is_err(), "reported once");
7594 }
7595
7596 #[tokio::test]
7599 async fn a_timed_out_recovery_charges_only_the_gaps_it_started() {
7600 let server = wiremock::MockServer::start().await;
7601 wiremock::Mock::given(wiremock::matchers::method("GET"))
7602 .and(wiremock::matchers::path("/api/v3/myTrades"))
7603 .respond_with(
7604 wiremock::ResponseTemplate::new(200)
7605 .set_body_json(serde_json::json!([]))
7606 .set_delay(Duration::from_secs(3_600)),
7607 )
7608 .mount(&server)
7609 .await;
7610 let rest = Arc::new(SpotRestApi::from_config(
7611 ConfigurationRestApi::builder()
7612 .api_key("key")
7613 .api_secret("secret")
7614 .base_path(server.uri())
7615 .timeout(3_600_000_u64)
7616 .retries(0_u32)
7617 .build()
7618 .unwrap(),
7619 ));
7620 let instruments: Vec<_> = (0..9)
7621 .map(|i| InstrumentNameExchange::new(format!("SYM{i}USDT")))
7622 .collect();
7623 let mut unrecovered = UnrecoveredFills::default();
7624 unrecovered.open(
7625 &instruments,
7626 Utc::now() - chrono::Duration::minutes(10),
7627 Utc::now(),
7628 );
7629 let (tx, _rx) = mpsc::unbounded_channel();
7630 tokio::time::pause();
7632
7633 recover_fills(
7634 &rest,
7635 &Arc::new(RateLimitTracker::new(WeightPool::Spot)),
7636 &mut unrecovered,
7637 &tx,
7638 &new_dedup_cache(),
7639 &KnownLiveOrders::shared(ExchangeId::BinanceSpot),
7640 )
7641 .await;
7642
7643 let now = tokio::time::Instant::now();
7644 assert_eq!(
7645 unrecovered.due(now).len(),
7646 1,
7647 "the gap never started stays due"
7648 );
7649 assert_eq!(
7650 unrecovered
7651 .due(now + Duration::from_secs(crate::client::order_recovery::GAP_RETRY_BASE_SECS))
7652 .len(),
7653 9,
7654 "the eight started wait for their first retry"
7655 );
7656 }
7657
7658 fn order_row(cid: &str, status: &str) -> serde_json::Value {
7664 serde_json::json!({
7665 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid, "price": "100",
7666 "origQty": "2", "executedQty": "1", "cummulativeQuoteQty": "105", "status": status,
7667 "timeInForce": "GTC", "type": "LIMIT", "side": "BUY", "time": 1_700_000_000_000_i64,
7668 "updateTime": 1_700_000_060_000_i64,
7669 })
7670 }
7671
7672 fn get_order_row(row: serde_json::Value) -> binance_sdk::spot::rest_api::GetOrderResponse {
7673 serde_json::from_value(row).unwrap()
7674 }
7675
7676 fn spot_key(instrument: &str, cid: &str) -> UnindexedOrderKey {
7677 OrderKey::new(
7678 ExchangeId::BinanceSpot,
7679 InstrumentNameExchange::new(instrument),
7680 StrategyId::new("strategy"),
7681 ClientOrderId::new(cid),
7682 )
7683 }
7684
7685 fn ended_state(
7686 key: &UnindexedOrderKey,
7687 row: serde_json::Value,
7688 ) -> Option<crate::order::state::UnindexedInactiveOrderState> {
7689 let order = convert_ended_order(&get_order_row(row), ExchangeId::BinanceSpot, key, None)?;
7690 assert_eq!(&order.key, key, "reported under the key asked");
7691 assert_eq!(order.quantity, Decimal::TWO);
7692 Some(order.state)
7693 }
7694
7695 #[test]
7698 fn an_ended_order_row_without_a_fill_reports_it_unknown() {
7699 use crate::order::state::{Expired, Filled, InactiveOrderState};
7700 let key = spot_key("BTCUSDT", "a");
7701 let ended = Utc.timestamp_millis_opt(1_700_000_060_000).unwrap();
7702 let id = OrderId::new("7");
7703 let row = |status: &str, executed_qty: Option<&str>| {
7704 let mut row = order_row("a", status);
7705 row["executedQty"] = executed_qty.map_or(serde_json::Value::Null, Into::into);
7706 row
7707 };
7708
7709 for executed_qty in [None, Some("not a number")] {
7710 assert_eq!(
7711 ended_state(&key, row("CANCELED", executed_qty)),
7712 Some(InactiveOrderState::Cancelled(Cancelled::new(
7713 id.clone(),
7714 ended,
7715 None
7716 ))),
7717 "{executed_qty:?}"
7718 );
7719 assert_eq!(
7720 ended_state(&key, row("EXPIRED", executed_qty)),
7721 Some(InactiveOrderState::Expired(Expired::new(
7722 id.clone(),
7723 ended,
7724 None
7725 ))),
7726 "{executed_qty:?}"
7727 );
7728 assert_eq!(
7729 ended_state(&key, row("FILLED", executed_qty)),
7730 Some(InactiveOrderState::FullyFilled(Filled::new(
7731 id.clone(),
7732 ended,
7733 Decimal::TWO,
7734 Some(Decimal::new(525, 1)),
7735 ))),
7736 "the whole quantity, at the quote over it: {executed_qty:?}"
7737 );
7738 }
7739 }
7740
7741 #[test]
7742 fn an_ended_order_row_says_how_the_order_ended() {
7743 use crate::order::state::{Expired, Filled, InactiveOrderState};
7744 let key = spot_key("BTCUSDT", "a");
7745 let ended = Utc.timestamp_millis_opt(1_700_000_060_000).unwrap();
7746 let id = OrderId::new("7");
7747
7748 let mut filled = order_row("a", "FILLED");
7749 filled["executedQty"] = "2".into();
7750 filled["cummulativeQuoteQty"] = "210".into();
7751 assert_eq!(
7752 ended_state(&key, filled),
7753 Some(InactiveOrderState::FullyFilled(Filled::new(
7754 id.clone(),
7755 ended,
7756 Decimal::TWO,
7757 Some(Decimal::from(105)),
7758 )))
7759 );
7760 assert_eq!(
7761 ended_state(&key, order_row("a", "CANCELED")),
7762 Some(InactiveOrderState::Cancelled(Cancelled::new(
7763 id.clone(),
7764 ended,
7765 Some(Decimal::ONE)
7766 )))
7767 );
7768 for status in ["EXPIRED", "EXPIRED_IN_MATCH"] {
7769 assert_eq!(
7770 ended_state(&key, order_row("a", status)),
7771 Some(InactiveOrderState::Expired(Expired::new(
7772 id.clone(),
7773 ended,
7774 Some(Decimal::ONE)
7775 ))),
7776 "{status}"
7777 );
7778 }
7779 assert!(matches!(
7780 ended_state(&key, order_row("a", "REJECTED")),
7781 Some(InactiveOrderState::OpenFailed(OrderError::Rejected(
7782 ApiError::OrderRejected(_)
7783 )))
7784 ));
7785 }
7786
7787 #[test]
7788 fn a_live_or_unreadable_order_row_is_not_ended() {
7789 let key = spot_key("BTCUSDT", "a");
7790 for status in [
7791 "NEW",
7792 "PARTIALLY_FILLED",
7793 "PENDING_NEW",
7794 "PENDING_CANCEL",
7795 "SOMETHING_NEW",
7796 ] {
7797 assert_eq!(ended_state(&key, order_row("a", status)), None, "{status}");
7798 }
7799 let mut no_status = order_row("a", "FILLED");
7800 no_status["status"] = serde_json::Value::Null;
7801 assert_eq!(ended_state(&key, no_status), None);
7802 let mut no_side = order_row("a", "FILLED");
7803 no_side["side"] = serde_json::Value::Null;
7804 assert_eq!(ended_state(&key, no_side), None);
7805 }
7806
7807 #[test]
7808 fn an_unknown_order_or_symbol_is_recognised_by_its_code() {
7809 use binance_sdk::common::errors::ConnectorError;
7810 let bad = |code| {
7811 anyhow::Error::new(ConnectorError::BadRequestError {
7812 msg: "no".into(),
7813 code: Some(code),
7814 })
7815 };
7816 assert!(is_unknown_order(&bad(-2013)));
7817 assert!(is_unknown_order(&bad(-1121)));
7818 assert!(!is_unknown_order(&bad(-1100)));
7819 assert!(is_unknown_order(&anyhow::Error::new(
7820 ConnectorError::NotFoundError {
7821 msg: "no".into(),
7822 code: Some(-2013),
7823 }
7824 )));
7825 assert!(!is_unknown_order(&anyhow::anyhow!("-2013 in text only")));
7826 }
7827
7828 #[tokio::test(start_paused = true)]
7831 async fn the_host_clock_lead_is_the_clients_server_time_reads() {
7832 let config = BinanceSpotConfig::testnet("key".into(), "secret".into());
7833 assert!(config.server_time_sync);
7834 assert!(
7835 BinanceSpot::new(config.clone())
7836 .rate_limiter
7837 .signing_clock()
7838 .is_some()
7839 );
7840 let mut client = BinanceSpot::new(config.with_server_time_sync(false));
7841 assert!(client.rate_limiter.signing_clock().is_none());
7842 assert_eq!(client.host_clock_lead(), None);
7843
7844 let host_ms = Utc::now().timestamp_millis();
7845 let read: crate::client::binance::server_time::ReadServerTime =
7846 Arc::new(move || Box::pin(async move { Ok(host_ms - 2_000) }));
7847 client.rate_limiter = Arc::new(
7848 RateLimitTracker::new(WeightPool::Spot)
7849 .with_server_time(Some(ServerTime::new(SigningClock::default(), read))),
7850 );
7851 client
7852 .rate_limiter
7853 .wait_if_blocked(RequestKind::Query)
7854 .await
7855 .unwrap();
7856
7857 let lead = client.host_clock_lead().unwrap();
7858 assert!(
7859 lead >= TimeDelta::seconds(2) && lead < TimeDelta::seconds(3),
7860 "{lead}"
7861 );
7862 }
7863
7864 fn rest_at(server: &wiremock::MockServer) -> Arc<RestApi> {
7865 Arc::new(SpotRestApi::from_config(
7866 ConfigurationRestApi::builder()
7867 .api_key("key")
7868 .api_secret("secret")
7869 .base_path(server.uri())
7870 .build()
7871 .unwrap(),
7872 ))
7873 }
7874
7875 async fn mount_order(
7877 server: &wiremock::MockServer,
7878 cid: &str,
7879 response: wiremock::ResponseTemplate,
7880 ) {
7881 wiremock::Mock::given(wiremock::matchers::method("GET"))
7882 .and(wiremock::matchers::path("/api/v3/order"))
7883 .and(wiremock::matchers::query_param("origClientOrderId", cid))
7884 .respond_with(response)
7885 .mount(server)
7886 .await;
7887 }
7888
7889 fn venue_error(code: i64) -> wiremock::ResponseTemplate {
7890 wiremock::ResponseTemplate::new(400)
7891 .set_body_json(serde_json::json!({"code": code, "msg": "refused"}))
7892 }
7893
7894 fn row_response(cid: &str, status: &str) -> wiremock::ResponseTemplate {
7895 wiremock::ResponseTemplate::new(200).set_body_json(order_row(cid, status))
7896 }
7897
7898 async fn looked_up(server: &wiremock::MockServer) -> Vec<String> {
7900 server
7901 .received_requests()
7902 .await
7903 .unwrap()
7904 .iter()
7905 .filter(|request| request.url.path() == "/api/v3/order")
7906 .filter_map(|request| {
7907 request
7908 .url
7909 .query_pairs()
7910 .find(|(key, _)| key == "origClientOrderId")
7911 .map(|(_, value)| value.into_owned())
7912 })
7913 .collect()
7914 }
7915
7916 #[tokio::test]
7921 async fn a_balance_read_is_stamped_the_rest_lag_before_its_request_was_sent() {
7922 use crate::client::binance::shared::REST_BALANCE_LAG;
7923 const ANSWER_DELAY: Duration = Duration::from_millis(200);
7924 let server = wiremock::MockServer::start().await;
7925 wiremock::Mock::given(wiremock::matchers::method("GET"))
7926 .and(wiremock::matchers::path("/api/v3/account"))
7927 .respond_with(
7928 wiremock::ResponseTemplate::new(429)
7929 .insert_header("retry-after", "1")
7930 .set_body_json(serde_json::json!({ "code": -1003, "msg": "Too much weight." })),
7931 )
7932 .up_to_n_times(1)
7933 .mount(&server)
7934 .await;
7935 wiremock::Mock::given(wiremock::matchers::method("GET"))
7936 .and(wiremock::matchers::path("/api/v3/account"))
7937 .respond_with(
7938 wiremock::ResponseTemplate::new(200)
7939 .set_body_json(serde_json::json!({
7940 "balances": [{ "asset": "BTC", "free": "1", "locked": "0" }],
7941 }))
7942 .set_delay(ANSWER_DELAY),
7943 )
7944 .mount(&server)
7945 .await;
7946 let mut client = <BinanceSpot as ExecutionClient>::new(
7947 BinanceSpotConfig::new("key".into(), "secret".into()).with_server_time_sync(false),
7948 );
7949 client.rest = rest_at(&server);
7950
7951 let before = Utc::now();
7952 let fetched = client.fetch_balances(&[]).await.unwrap();
7953 let retried = Utc::now();
7954 let snapshot = client.account_snapshot(&[], &[]).await.unwrap();
7955 let after = Utc::now();
7956
7957 let within = |time, from, to| {
7958 assert!(
7959 from <= time + REST_BALANCE_LAG && time + REST_BALANCE_LAG + ANSWER_DELAY <= to,
7960 "{time} + the lag not within {from} ..= {to} - {ANSWER_DELAY:?}"
7961 );
7962 };
7963 assert_eq!((fetched.len(), snapshot.balances.len()), (1, 1));
7964 within(
7966 fetched[0].time_exchange,
7967 before + Duration::from_secs(1),
7968 retried,
7969 );
7970 within(snapshot.balances[0].time_exchange, retried, after);
7971 }
7972
7973 #[tokio::test]
7978 async fn a_balance_read_is_stamped_on_binances_clock_when_the_host_clock_leads() {
7979 use crate::client::binance::shared::REST_BALANCE_LAG;
7980 const LEAD: TimeDelta = TimeDelta::seconds(60);
7981 let server = wiremock::MockServer::start().await;
7982 wiremock::Mock::given(wiremock::matchers::method("GET"))
7983 .and(wiremock::matchers::path("/api/v3/account"))
7984 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
7985 serde_json::json!({ "balances": [{ "asset": "BTC", "free": "1", "locked": "0" }] }),
7986 ))
7987 .mount(&server)
7988 .await;
7989 let mut client = <BinanceSpot as ExecutionClient>::new(
7990 BinanceSpotConfig::new("key".into(), "secret".into()).with_server_time_sync(false),
7991 );
7992 client.rest = rest_at(&server);
7993 client.rate_limiter = Arc::new(
7994 RateLimitTracker::new(WeightPool::Spot)
7995 .with_server_time(Some(ServerTime::trailing_host(LEAD))),
7996 );
7997 let clock = StreamClock::shared(client.rate_limiter.signing_clock());
7998
7999 let before = Utc::now();
8000 let fetched = client.fetch_balances(&[]).await.unwrap();
8001 let snapshot = client.account_snapshot(&[], &[]).await.unwrap();
8002 let restated = read_spot_restatement(&client.rest, &client.rate_limiter, &[], &clock, None)
8003 .await
8004 .unwrap();
8005 let after = Utc::now();
8006
8007 let [restated] = restated.as_slice() else {
8008 panic!("one balance: {restated:?}");
8009 };
8010 let AccountEventKind::BalanceSnapshot(restated) = &restated.kind else {
8011 panic!("a balance snapshot: {restated:?}");
8012 };
8013 assert_eq!((fetched.len(), snapshot.balances.len()), (1, 1));
8014 for time in [
8015 fetched[0].time_exchange,
8016 snapshot.balances[0].time_exchange,
8017 restated.0.time_exchange,
8018 ] {
8019 let sent = time + REST_BALANCE_LAG + LEAD;
8022 assert!(
8023 before - ServerTime::TRAILING_HOST_BEHIND <= sent && sent <= after,
8024 "{time} + the lag and the lead not within {before} ..= {after}"
8025 );
8026 }
8027 }
8028
8029 #[tokio::test]
8032 async fn a_restatement_read_returns_the_requested_balances_as_snapshots() {
8033 use crate::client::binance::shared::REST_BALANCE_LAG;
8034 let server = wiremock::MockServer::start().await;
8035 wiremock::Mock::given(wiremock::matchers::method("GET"))
8036 .and(wiremock::matchers::path("/api/v3/account"))
8037 .respond_with(wiremock::ResponseTemplate::new(500))
8038 .up_to_n_times(1)
8039 .mount(&server)
8040 .await;
8041 wiremock::Mock::given(wiremock::matchers::method("GET"))
8042 .and(wiremock::matchers::path("/api/v3/account"))
8043 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
8044 serde_json::json!({ "balances": [
8045 { "asset": "BTC", "free": "1", "locked": "0.5" },
8046 { "asset": "ETH", "free": "2", "locked": "0" },
8047 ] }),
8048 ))
8049 .mount(&server)
8050 .await;
8051 let (rest, limiter) = (
8052 rest_at(&server),
8053 Arc::new(RateLimitTracker::new(WeightPool::Spot)),
8054 );
8055 let clock = SharedStreamClock::default();
8056 let btc = [AssetNameExchange::new("BTC")];
8057
8058 let failed = read_spot_restatement(&rest, &limiter, &btc, &clock, None).await;
8059 assert!(failed.is_err(), "{failed:?}");
8060
8061 let before = Utc::now();
8062 let sent = read_spot_restatement(&rest, &limiter, &btc, &clock, None)
8063 .await
8064 .unwrap();
8065 let after = Utc::now();
8066
8067 let [event] = sent.as_slice() else {
8068 panic!("one balance: {sent:?}");
8069 };
8070 assert_eq!(event.exchange, ExchangeId::BinanceSpot);
8071 let AccountEventKind::BalanceSnapshot(balance) = &event.kind else {
8072 panic!("a balance snapshot: {event:?}");
8073 };
8074 assert_eq!(balance.0.asset, btc[0]);
8075 assert_eq!(
8076 balance.0.balance,
8077 Balance::new(Decimal::new(15, 1), Decimal::ONE)
8078 );
8079 let read = balance.0.time_exchange + REST_BALANCE_LAG;
8080 assert!(
8081 before <= read && read <= after,
8082 "{read} not within {before} ..= {after}"
8083 );
8084 }
8085
8086 #[tokio::test]
8087 async fn fetch_ended_orders_reports_each_ended_order_under_its_key_and_omits_the_rest() {
8088 let server = wiremock::MockServer::start().await;
8089 mount_order(&server, "filled", row_response("filled", "FILLED")).await;
8090 mount_order(&server, "cancelled", row_response("cancelled", "CANCELED")).await;
8091 mount_order(&server, "live", row_response("live", "PARTIALLY_FILLED")).await;
8092 mount_order(&server, "unknown", venue_error(-2013)).await;
8093 mount_order(&server, "no-symbol", venue_error(-1121)).await;
8094 let mut client = <BinanceSpot as ExecutionClient>::new(
8095 BinanceSpotConfig::new("key".into(), "secret".into()).with_server_time_sync(false),
8096 );
8097 client.rest = rest_at(&server);
8098 let keys = ["filled", "cancelled", "live", "unknown"].map(|cid| spot_key("BTCUSDT", cid));
8099
8100 let mut ended = client
8101 .fetch_ended_orders(&[keys.to_vec(), vec![spot_key("NOSUCH", "no-symbol")]].concat())
8102 .await
8103 .unwrap();
8104
8105 ended.sort_by(|a, b| a.key.cid.cmp(&b.key.cid));
8106 let reported: Vec<_> = ended.iter().map(|order| &order.key).collect();
8107 assert_eq!(
8108 reported,
8109 [&keys[1], &keys[0]],
8110 "the strategy asked survives"
8111 );
8112 assert!(matches!(
8113 ended[0].state,
8114 crate::order::state::InactiveOrderState::Cancelled(_)
8115 ));
8116 assert!(matches!(
8117 ended[1].state,
8118 crate::order::state::InactiveOrderState::FullyFilled(_)
8119 ));
8120 assert!(client.fetch_ended_orders(&[]).await.unwrap().is_empty());
8121 }
8122
8123 #[tokio::test]
8124 async fn fetch_ended_orders_fails_when_any_lookup_fails() {
8125 let server = wiremock::MockServer::start().await;
8126 mount_order(&server, "cancelled", row_response("cancelled", "CANCELED")).await;
8127 mount_order(&server, "refused", venue_error(-1100)).await;
8128 let mut client = <BinanceSpot as ExecutionClient>::new(
8129 BinanceSpotConfig::new("key".into(), "secret".into()).with_server_time_sync(false),
8130 );
8131 client.rest = rest_at(&server);
8132
8133 let result = client
8134 .fetch_ended_orders(&[
8135 spot_key("BTCUSDT", "cancelled"),
8136 spot_key("BTCUSDT", "refused"),
8137 ])
8138 .await;
8139
8140 assert!(
8141 matches!(
8142 result,
8143 Err(UnindexedClientError::Api(ApiError::RequestRejected(_)))
8144 ),
8145 "{result:?}"
8146 );
8147 }
8148
8149 async fn venue_after_a_disconnect(delay: Duration) -> wiremock::MockServer {
8152 let server = wiremock::MockServer::start().await;
8153 wiremock::Mock::given(wiremock::matchers::method("GET"))
8154 .and(wiremock::matchers::path("/api/v3/openOrders"))
8155 .and(wiremock::matchers::query_param("symbol", "BTCUSDT"))
8156 .respond_with(
8157 wiremock::ResponseTemplate::new(200)
8158 .set_body_json(serde_json::json!([order_row("live", "NEW")])),
8159 )
8160 .mount(&server)
8161 .await;
8162 mount_order(
8163 &server,
8164 "gone",
8165 row_response("gone", "CANCELED").set_delay(delay),
8166 )
8167 .await;
8168 mount_order(&server, "unknown", venue_error(-2013)).await;
8169 server
8170 }
8171
8172 fn held_orders() -> SharedKnownLiveOrders {
8174 let known = KnownLiveOrders::shared(ExchangeId::BinanceSpot);
8175 for (n, cid) in ["live", "gone", "unknown"].into_iter().enumerate() {
8176 known.lock().live(
8177 &spot_key("BTCUSDT", cid),
8178 Decimal::TWO,
8179 OrderKind::Limit,
8180 &Open::new(
8181 VenueOrderId::Assigned(OrderId::new(n.to_string())),
8182 Utc::now(),
8183 Decimal::ZERO,
8184 ),
8185 );
8186 }
8187 known
8188 }
8189
8190 fn btcusdt_unchecked() -> UncheckedOrders {
8191 let mut unchecked = UncheckedOrders::default();
8192 unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
8193 unchecked
8194 }
8195
8196 #[tokio::test]
8197 async fn a_reconnect_reports_how_an_unlisted_order_ended_and_forgets_an_unknown_one() {
8198 let server = venue_after_a_disconnect(Duration::ZERO).await;
8199 let known = held_orders();
8200 let mut unchecked = btcusdt_unchecked();
8201 let (tx, mut rx) = mpsc::unbounded_channel();
8202
8203 recover_spot_ended_orders(
8204 &rest_at(&server),
8205 &Arc::new(RateLimitTracker::new(WeightPool::Spot)),
8206 &known,
8207 &mut unchecked,
8208 &UnrecoveredFills::default(),
8209 &tx,
8210 )
8211 .await;
8212
8213 let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
8214 let [event] = sent.as_slice() else {
8215 panic!("one order ended: {sent:?}");
8216 };
8217 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
8218 panic!("an order snapshot: {event:?}");
8219 };
8220 assert_eq!(snapshot.0.key.cid, ClientOrderId::new("gone"));
8221 assert_eq!(snapshot.0.key.strategy, StrategyId::unknown());
8222 assert!(matches!(
8223 snapshot.0.state,
8224 OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(_))
8225 ));
8226 let mut asked = looked_up(&server).await;
8227 asked.sort();
8228 assert_eq!(
8229 asked,
8230 ["gone", "unknown"],
8231 "a listed order is not looked up"
8232 );
8233 let known = known.lock();
8234 assert!(known.contains(&ClientOrderId::new("live")));
8235 assert!(!known.contains(&ClientOrderId::new("gone")));
8236 assert!(
8237 !known.contains(&ClientOrderId::new("unknown")),
8238 "asking again cannot help"
8239 );
8240 assert!(unchecked.is_empty());
8241 }
8242
8243 #[tokio::test]
8244 async fn an_order_the_stream_reports_ending_during_the_check_is_not_reported_again() {
8245 let server = venue_after_a_disconnect(Duration::from_millis(300)).await;
8246 let known = held_orders();
8247 let mut unchecked = btcusdt_unchecked();
8248 let (tx, mut rx) = mpsc::unbounded_channel();
8249 let (rest, tracker) = (
8250 rest_at(&server),
8251 Arc::new(RateLimitTracker::new(WeightPool::Spot)),
8252 );
8253 let unrecovered = UnrecoveredFills::default();
8254
8255 tokio::join!(
8256 recover_spot_ended_orders(&rest, &tracker, &known, &mut unchecked, &unrecovered, &tx,),
8257 async {
8258 tokio::time::sleep(Duration::from_millis(100)).await;
8259 known.lock().ended(&ClientOrderId::new("gone"));
8260 }
8261 );
8262
8263 assert!(rx.try_recv().is_err(), "the stream already reported it");
8264 assert!(unchecked.is_empty());
8265 }
8266
8267 #[tokio::test]
8268 async fn an_instrument_with_unrecovered_fills_is_not_checked_yet() {
8269 let server = venue_after_a_disconnect(Duration::ZERO).await;
8270 let known = held_orders();
8271 let mut unchecked = btcusdt_unchecked();
8272 let mut unrecovered = UnrecoveredFills::default();
8273 unrecovered.open(
8274 &[InstrumentNameExchange::new("BTCUSDT")],
8275 Utc::now() - chrono::Duration::minutes(1),
8276 Utc::now(),
8277 );
8278 let (tx, mut rx) = mpsc::unbounded_channel();
8279
8280 recover_spot_ended_orders(
8281 &rest_at(&server),
8282 &Arc::new(RateLimitTracker::new(WeightPool::Spot)),
8283 &known,
8284 &mut unchecked,
8285 &unrecovered,
8286 &tx,
8287 )
8288 .await;
8289
8290 assert!(rx.try_recv().is_err());
8291 assert!(server.received_requests().await.unwrap().is_empty());
8292 assert!(!unchecked.is_empty(), "it waits for its fills");
8293 assert_eq!(unchecked.next_due(&unrecovered), None, "woken by the gap");
8294 }
8295
8296 #[tokio::test]
8297 async fn a_failed_check_sends_nothing_and_waits_for_its_retry() {
8298 let server = wiremock::MockServer::start().await;
8299 wiremock::Mock::given(wiremock::matchers::method("GET"))
8300 .and(wiremock::matchers::path("/api/v3/openOrders"))
8301 .respond_with(venue_error(-1100))
8302 .mount(&server)
8303 .await;
8304 let server_rest = rest_at(&server);
8305 let known = held_orders();
8306 let mut unchecked = btcusdt_unchecked();
8307 let (tx, mut rx) = mpsc::unbounded_channel();
8308
8309 recover_spot_ended_orders(
8310 &server_rest,
8311 &Arc::new(RateLimitTracker::new(WeightPool::Spot)),
8312 &known,
8313 &mut unchecked,
8314 &UnrecoveredFills::default(),
8315 &tx,
8316 )
8317 .await;
8318
8319 assert!(rx.try_recv().is_err());
8320 assert!(!unchecked.is_empty());
8321 assert!(
8322 unchecked
8323 .next_due(&UnrecoveredFills::default())
8324 .is_some_and(|due| due > tokio::time::Instant::now()),
8325 "retried after a backoff"
8326 );
8327 assert!(
8328 known.lock().contains(&ClientOrderId::new("gone")),
8329 "still held"
8330 );
8331 }
8332
8333 #[tokio::test]
8334 async fn an_unlisted_order_the_lookup_finds_live_stays_held() {
8335 let server = wiremock::MockServer::start().await;
8336 wiremock::Mock::given(wiremock::matchers::method("GET"))
8337 .and(wiremock::matchers::path("/api/v3/openOrders"))
8338 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
8339 .mount(&server)
8340 .await;
8341 for (cid, status) in [
8342 ("live", "NEW"),
8343 ("gone", "PENDING_CANCEL"),
8344 ("unknown", "WHAT"),
8345 ] {
8346 mount_order(&server, cid, row_response(cid, status)).await;
8347 }
8348 let known = held_orders();
8349 let mut unchecked = btcusdt_unchecked();
8350 let (tx, mut rx) = mpsc::unbounded_channel();
8351
8352 recover_spot_ended_orders(
8353 &rest_at(&server),
8354 &Arc::new(RateLimitTracker::new(WeightPool::Spot)),
8355 &known,
8356 &mut unchecked,
8357 &UnrecoveredFills::default(),
8358 &tx,
8359 )
8360 .await;
8361
8362 assert!(rx.try_recv().is_err(), "none has ended");
8363 let known = known.lock();
8364 for cid in ["live", "gone", "unknown"] {
8365 assert!(
8366 known.contains(&ClientOrderId::new(cid)),
8367 "{cid}: only an order Binance does not know is dropped"
8368 );
8369 }
8370 assert!(unchecked.is_empty());
8371 }
8372
8373 #[tokio::test]
8374 async fn one_instrument_failing_does_not_hold_up_another() {
8375 let server = venue_after_a_disconnect(Duration::ZERO).await;
8376 wiremock::Mock::given(wiremock::matchers::method("GET"))
8377 .and(wiremock::matchers::path("/api/v3/openOrders"))
8378 .and(wiremock::matchers::query_param("symbol", "ETHUSDT"))
8379 .respond_with(venue_error(-1100))
8380 .mount(&server)
8381 .await;
8382 let known = held_orders();
8383 let eth = spot_key("ETHUSDT", "eth");
8384 known.lock().live(
8385 ð,
8386 Decimal::TWO,
8387 OrderKind::Limit,
8388 &Open::new(
8389 VenueOrderId::Assigned(OrderId::new("9")),
8390 Utc::now(),
8391 Decimal::ZERO,
8392 ),
8393 );
8394 let mut unchecked = btcusdt_unchecked();
8395 unchecked.open([eth.instrument.clone()]);
8396 let (tx, mut rx) = mpsc::unbounded_channel();
8397
8398 recover_spot_ended_orders(
8399 &rest_at(&server),
8400 &Arc::new(RateLimitTracker::new(WeightPool::Spot)),
8401 &known,
8402 &mut unchecked,
8403 &UnrecoveredFills::default(),
8404 &tx,
8405 )
8406 .await;
8407
8408 let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
8409 assert_eq!(sent.len(), 1, "BTCUSDT's ended order: {sent:?}");
8410 assert!(!unchecked.contains(&InstrumentNameExchange::new("BTCUSDT")));
8411 assert!(
8412 unchecked.contains(ð.instrument),
8413 "ETHUSDT waits for its retry"
8414 );
8415 assert!(known.lock().contains(ð.cid));
8416 }
8417
8418 #[tokio::test]
8419 async fn a_check_dropped_part_way_changes_nothing() {
8420 let server = venue_after_a_disconnect(Duration::from_millis(500)).await;
8421 let known = held_orders();
8422 let mut unchecked = btcusdt_unchecked();
8423 let (tx, mut rx) = mpsc::unbounded_channel();
8424
8425 let dropped = tokio::time::timeout(
8426 Duration::from_millis(100),
8427 recover_spot_ended_orders(
8428 &rest_at(&server),
8429 &Arc::new(RateLimitTracker::new(WeightPool::Spot)),
8430 &known,
8431 &mut unchecked,
8432 &UnrecoveredFills::default(),
8433 &tx,
8434 ),
8435 )
8436 .await;
8437
8438 assert!(dropped.is_err(), "the lookup of `gone` was still waiting");
8439 assert!(rx.try_recv().is_err());
8440 assert!(known.lock().contains(&ClientOrderId::new("gone")));
8441 assert!(unchecked.contains(&InstrumentNameExchange::new("BTCUSDT")));
8442 assert!(
8443 unchecked
8444 .next_due(&UnrecoveredFills::default())
8445 .is_some_and(|due| due <= tokio::time::Instant::now()),
8446 "still due, not charged a failure"
8447 );
8448 }
8449
8450 fn commission_answer(symbol: &str) -> serde_json::Value {
8452 let part = |maker: &str, taker: &str, buyer: &str, seller: &str| serde_json::json!({"maker": maker, "taker": taker, "buyer": buyer, "seller": seller});
8453 serde_json::json!({
8454 "symbol": symbol,
8455 "standardCommission": part("0.00000010", "0.00000020", "0.00000030", "0.00000040"),
8456 "specialCommission": part("0.01000000", "0.02000000", "0.03000000", "0.04000000"),
8457 "taxCommission": part("0.00000112", "0.00000114", "0.00000118", "0.00000116"),
8458 "discount": {
8459 "enabledForAccount": true,
8460 "enabledForSymbol": true,
8461 "discountAsset": "BNB",
8462 "discount": "0.75000000"
8463 }
8464 })
8465 }
8466
8467 #[tokio::test]
8468 async fn spot_commission_rates_are_read_signed_for_the_symbol() {
8469 use crate::client::binance::{CommissionDiscount, CommissionPart};
8470 use rust_decimal_macros::dec;
8471 use rustrade_instrument::asset::name::AssetNameExchange;
8472 let server = wiremock::MockServer::start().await;
8473 wiremock::Mock::given(wiremock::matchers::method("GET"))
8474 .and(wiremock::matchers::path("/api/v3/account/commission"))
8475 .and(wiremock::matchers::query_param("symbol", "BTCUSDT"))
8476 .and(wiremock::matchers::header("X-MBX-APIKEY", "key"))
8477 .respond_with(
8478 wiremock::ResponseTemplate::new(200).set_body_json(commission_answer("BTCUSDT")),
8479 )
8480 .expect(1)
8481 .mount(&server)
8482 .await;
8483 let mut client = <BinanceSpot as ExecutionClient>::new(
8484 BinanceSpotConfig::new("key".into(), "secret".into()).with_server_time_sync(false),
8485 );
8486 client.rest = rest_at(&server);
8487 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
8488
8489 let rates = client.fetch_commission_rates(&btcusdt).await.unwrap();
8490 let requests = server.received_requests().await.unwrap();
8491 assert!(
8492 requests[0]
8493 .url
8494 .query_pairs()
8495 .any(|(key, _)| key == "signature"),
8496 "signed: {}",
8497 requests[0].url
8498 );
8499 assert_eq!(rates.instrument, btcusdt);
8500 assert_eq!(
8501 rates.special,
8502 CommissionPart::new(dec!(0.01), dec!(0.02), dec!(0.03), dec!(0.04))
8503 );
8504 assert_eq!(
8505 rates.discount,
8506 Some(CommissionDiscount::new(
8507 true,
8508 true,
8509 Some(AssetNameExchange::new("BNB")),
8510 dec!(0.75)
8511 ))
8512 );
8513 }
8514
8515 #[tokio::test]
8516 async fn spot_commission_rates_of_an_unlisted_symbol_name_it() {
8517 let server = wiremock::MockServer::start().await;
8518 wiremock::Mock::given(wiremock::matchers::path("/api/v3/account/commission"))
8519 .respond_with(
8520 wiremock::ResponseTemplate::new(400)
8521 .set_body_json(serde_json::json!({"code": -1121, "msg": "Invalid symbol."})),
8522 )
8523 .mount(&server)
8524 .await;
8525 let mut client = <BinanceSpot as ExecutionClient>::new(
8526 BinanceSpotConfig::new("key".into(), "secret".into()).with_server_time_sync(false),
8527 );
8528 client.rest = rest_at(&server);
8529 let nope = InstrumentNameExchange::new("NOPEUSDT");
8530 match client.fetch_commission_rates(&nope).await {
8531 Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(instrument, _))) => {
8532 assert_eq!(instrument, nope);
8533 }
8534 other => panic!("{other:?}"),
8535 }
8536 }
8537
8538 mod cancel_replace {
8540 use super::*;
8541 use crate::{
8542 client::ExecutionClient,
8543 order::{
8544 request::{OnCancelFailure, RequestCancel, RequestOpen},
8545 state::InactiveOrderState,
8546 },
8547 };
8548 use rust_decimal_macros::dec;
8549 use wiremock::matchers::{header, method, path, query_param};
8550
8551 fn client(server: &wiremock::MockServer) -> BinanceSpot {
8554 let mut client = client_with_ws_api("ws://127.0.0.1:9".to_string());
8555 client.signed_rest = Arc::new(SignedRest::new(
8556 server.uri(),
8557 "key".to_string(),
8558 "secret".to_string(),
8559 None,
8560 Duration::from_secs(5),
8561 ));
8562 client
8563 }
8564
8565 fn btcusdt() -> InstrumentNameExchange {
8566 InstrumentNameExchange::new("BTCUSDT")
8567 }
8568
8569 fn cancel<'a>(
8570 instrument: &'a InstrumentNameExchange,
8571 id: Option<&str>,
8572 ) -> OrderRequestCancel<ExchangeId, &'a InstrumentNameExchange> {
8573 OrderRequestCancel {
8574 key: OrderKey::new(
8575 ExchangeId::BinanceSpot,
8576 instrument,
8577 StrategyId::new("test"),
8578 ClientOrderId::new("stop"),
8579 ),
8580 state: RequestCancel {
8581 id: id.map(|id| VenueOrderId::Assigned(OrderId::new(id))),
8582 },
8583 }
8584 }
8585
8586 fn open(
8587 instrument: &InstrumentNameExchange,
8588 request: RequestOpen,
8589 ) -> OrderRequestOpen<ExchangeId, &InstrumentNameExchange> {
8590 OrderRequestOpen {
8591 key: OrderKey::new(
8592 ExchangeId::BinanceSpot,
8593 instrument,
8594 StrategyId::new("test"),
8595 ClientOrderId::new("exit"),
8596 ),
8597 state: request,
8598 }
8599 }
8600
8601 fn limit() -> RequestOpen {
8602 RequestOpen::new(
8603 Side::Sell,
8604 Some(dec!(101.5)),
8605 dec!(0.5),
8606 OrderKind::Limit,
8607 TimeInForce::GoodUntilCancelled { post_only: false },
8608 )
8609 }
8610
8611 #[tokio::test]
8614 async fn one_signed_request_answers_each_leg() {
8615 let server = wiremock::MockServer::start().await;
8616 wiremock::Mock::given(method("POST"))
8617 .and(path("/api/v3/order/cancelReplace"))
8618 .and(header("X-MBX-APIKEY", "key"))
8619 .and(query_param("symbol", "BTCUSDT"))
8620 .and(query_param("cancelReplaceMode", "STOP_ON_FAILURE"))
8621 .and(query_param("cancelOrderId", "11"))
8622 .and(query_param("newClientOrderId", "exit"))
8623 .and(query_param("side", "SELL"))
8624 .and(query_param("type", "LIMIT"))
8625 .and(query_param("price", "101.5"))
8626 .and(query_param("quantity", "0.5"))
8627 .respond_with(wiremock::ResponseTemplate::new(409).set_body_json(
8628 serde_json::json!({
8629 "code": -2021, "msg": "Order cancel-replace partially failed.",
8630 "data": {
8631 "cancelResult": "SUCCESS", "newOrderResult": "FAILURE",
8632 "cancelResponse": {
8633 "symbol": "BTCUSDT", "orderId": 11, "status": "CANCELED",
8634 "transactTime": 1_700_000_000_000_i64, "executedQty": "0",
8635 },
8636 "newOrderResponse": {
8637 "code": -2010,
8638 "msg": "Account has insufficient balance for requested action.",
8639 },
8640 },
8641 }),
8642 ))
8643 .expect(1)
8644 .mount(&server)
8645 .await;
8646 let client = client(&server);
8647 let instrument = btcusdt();
8648
8649 let answer = client
8650 .cancel_replace(
8651 cancel(&instrument, Some("11")),
8652 open(&instrument, limit()),
8653 OnCancelFailure::DropOpens,
8654 )
8655 .expect("a same-symbol limit order goes as one request")
8656 .await;
8657
8658 let requests = server.received_requests().await.unwrap();
8659 let query = requests[0].url.query().unwrap_or_default();
8660 assert!(query.contains("timestamp="), "{query}");
8661 assert!(query.contains("signature="), "{query}");
8662
8663 assert_eq!(answer.cancel.key.cid, ClientOrderId::new("stop"));
8664 assert_eq!(
8665 answer.cancel.state.map(|cancelled| cancelled.id),
8666 Ok(OrderId::new("11"))
8667 );
8668 let opened = answer.open.expect("the new order was attempted");
8669 assert_eq!(opened.key.cid, ClientOrderId::new("exit"));
8670 assert_eq!(opened.price, Some(dec!(101.5)));
8671 assert!(
8672 matches!(
8673 opened.state,
8674 OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::Rejected(_)))
8675 ),
8676 "{:?}",
8677 opened.state
8678 );
8679 }
8680
8681 #[tokio::test]
8684 async fn a_throttled_request_fails_the_cancel_and_attempts_nothing() {
8685 let server = wiremock::MockServer::start().await;
8686 wiremock::Mock::given(method("POST"))
8687 .and(path("/api/v3/order/cancelReplace"))
8688 .respond_with(
8689 wiremock::ResponseTemplate::new(418)
8690 .insert_header("retry-after", "600")
8691 .set_body_json(
8692 serde_json::json!({ "code": -1003, "msg": "Way too much request weight used." }),
8693 ),
8694 )
8695 .expect(1)
8696 .mount(&server)
8697 .await;
8698 let client = client(&server);
8699 let instrument = btcusdt();
8700
8701 let answer = client
8702 .cancel_replace(
8703 cancel(&instrument, None),
8704 open(&instrument, limit()),
8705 OnCancelFailure::DropOpens,
8706 )
8707 .expect("offered")
8708 .await;
8709
8710 assert!(
8711 matches!(
8712 answer.cancel.state,
8713 Err(OrderError::Rejected(ApiError::RateLimit { .. }))
8714 ),
8715 "{:?}",
8716 answer.cancel.state
8717 );
8718 assert_eq!(answer.open, None);
8719 }
8720
8721 #[tokio::test]
8724 async fn a_pair_the_request_cannot_carry_is_declined() {
8725 let server = wiremock::MockServer::start().await;
8726 let client = client(&server);
8727 let instrument = btcusdt();
8728 let other = InstrumentNameExchange::new("ETHUSDT");
8729 let unsupported = RequestOpen::new(
8730 Side::Sell,
8731 Some(dec!(101.5)),
8732 dec!(0.5),
8733 OrderKind::Limit,
8734 TimeInForce::GoodUntilEndOfDay,
8735 );
8736 let borrowing = limit().with_margin_effect(MarginEffect::Borrow);
8737
8738 let declined = [
8739 (cancel(&instrument, Some("11")), open(&other, limit())),
8740 (
8741 cancel(&instrument, Some("11")),
8742 open(&instrument, unsupported),
8743 ),
8744 (
8745 cancel(&instrument, Some("11")),
8746 open(&instrument, borrowing),
8747 ),
8748 (cancel(&instrument, Some("x11")), open(&instrument, limit())),
8749 ];
8750 for (cancel, open) in declined {
8751 assert!(
8752 client
8753 .cancel_replace(cancel, open, OnCancelFailure::DropOpens)
8754 .is_none()
8755 );
8756 }
8757 assert!(server.received_requests().await.unwrap().is_empty());
8758 }
8759
8760 #[tokio::test]
8763 async fn an_open_the_rules_refuse_sends_the_cancel_alone() {
8764 let server = wiremock::MockServer::start().await;
8765 wiremock::Mock::given(method("GET"))
8766 .and(path("/api/v3/exchangeInfo"))
8767 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
8768 serde_json::json!({"symbols": [{
8769 "symbol": "BTCUSDT", "status": "TRADING",
8770 "filters": [
8771 {"filterType": "LOT_SIZE", "minQty": "1", "maxQty": "100", "stepSize": "1"},
8772 ],
8773 }]}),
8774 ))
8775 .mount(&server)
8776 .await;
8777 let mut client = client(&server);
8778 client.rules = Arc::new(RulesLoader::new(
8779 PreSendRules::Enforce,
8780 Arc::new(SpotRestApi::from_config(
8781 ConfigurationRestApi::builder()
8782 .api_key("key")
8783 .api_secret("secret")
8784 .base_path(server.uri())
8785 .build()
8786 .unwrap(),
8787 )),
8788 Arc::new(RateLimitTracker::new(WeightPool::Spot)),
8789 ));
8790 let instrument = btcusdt();
8791
8792 for (policy, open_answered) in [
8793 (OnCancelFailure::DropOpens, false),
8794 (OnCancelFailure::SendOpens, true),
8795 ] {
8796 let answer = client
8797 .cancel_replace(
8798 cancel(&instrument, Some("11")),
8799 open(&instrument, limit()),
8800 policy,
8801 )
8802 .expect("offered")
8803 .await;
8804
8805 assert!(answer.cancel.state.is_err());
8807 assert_eq!(answer.open.is_some(), open_answered, "{policy:?}");
8808 if let Some(opened) = answer.open {
8809 assert!(
8810 matches!(
8811 opened.state,
8812 OrderState::Inactive(InactiveOrderState::OpenFailed(
8813 OrderError::InvalidPrecision(_)
8814 ))
8815 ),
8816 "{:?}",
8817 opened.state
8818 );
8819 }
8820 }
8821 let requests = server.received_requests().await.unwrap();
8822 assert!(
8823 requests
8824 .iter()
8825 .all(|request| request.url.path() != "/api/v3/order/cancelReplace"),
8826 "{requests:?}"
8827 );
8828 }
8829 }
8830}