1mod records;
59
60pub use records::MarginForcedLiquidation;
61
62use super::margin_risk::{
63 RISK_STREAM_URL, RiskStreamTask, SapiRiskListenKeys, log_liability_change,
64 log_margin_level_change, margin_level_notice,
65};
66use super::shared::{
67 AbortOnDropStream, BINANCE_MAX_TRADES, BinanceOrderType, BinanceTimeInForce,
68 CONNECT_TIMEOUT_SECS, DisconnectOnDrop, ExecutionReport, ExponentialBackoff,
69 FILL_RECOVERY_TIMEOUT_SECS, HEARTBEAT_TIMEOUT_SECS, MyTradesFrom, ORDER_EXECUTIONS_BUDGET,
70 OpenOrderListing, PlacementResponse, RateLimitTracker, RequestKind,
71 SIGNAL_RECOVERY_LOOKBACK_MS, SharedDedupCache, TrailingOrderError, UnhandledEvents,
72 UnrecoveredFills, UnsubscribeOnDrop, UserDataFrame, WeightPool, binance_filled_qty,
73 classify_order_kind_tif, classify_rest_order_error, classify_rest_query_error,
74 convert_ended_order, convert_execution_report, convert_open_order_listing,
75 convert_open_order_owned_symbol, dedup_key_from_event, disconnect_error, drop_after,
76 frame_excerpt, gap_failed, gap_time, is_duplicate, is_unknown_order, log_unhandled_event,
77 log_unrecognised_frame, new_dedup_cache, parse_user_data_frame, placed_order_state,
78 recovered_order_cids, recovered_order_totals, response_decode_error, rest_call_with_retry,
79 trailing_params,
80};
81use crate::{
82 AccountEventKind, AccountSnapshot, FillRecoveryFailure, InstrumentAccountSnapshot,
83 InstrumentBalanceUpdate, IsolatedInstrumentState, IsolatedMarginRisk, UnindexedAccountEvent,
84 UnindexedAccountSnapshot,
85 balance::{AssetBalance, AssetBalanceUpdate, Balance, BalanceUpdate},
86 borrow_capacity::BorrowCapacity,
87 client::{
88 BorrowCapacityClient, ExecutionClient, OrderStatusClient, ShortabilityClient,
89 order_recovery::{
90 KnownLiveOrders, OpenListing, OrderLookup, SharedKnownLiveOrders, UncheckedOrders,
91 fetch_ended_by_key, recover_ended_orders,
92 },
93 },
94 emit_stream_terminated,
95 error::{ApiError, ConnectivityError, OrderError, UnindexedClientError},
96 order::{
97 Order, OrderKey, OrderKind, TimeInForce, UnindexedInactiveOrder, UnindexedOrderKey,
98 id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
99 request::{OrderRequestCancel, OrderRequestOpen, UnindexedOrderResponseCancel},
100 state::{Cancelled, Open, OrderState, UnindexedOrderState},
101 },
102 position::PositionReport,
103 shortability::Shortability,
104 trade::{AssetFees, Trade, TradeId, TradesRead},
105};
106use binance_sdk::{
107 common::{
108 config::{ConfigurationRestApi, ConfigurationWebsocketApi},
109 constants::{MARGIN_TRADING_REST_API_PROD_URL, SPOT_WS_API_PROD_URL},
110 errors::WebsocketError,
111 models::WebsocketEvent,
112 websocket::{
113 SendWebsocketMessageResult, WebsocketApi as WsApiBase, WebsocketMessageSendOptions,
114 },
115 },
116 margin_trading::{
117 MarginTradingRestApi,
118 rest_api::{
119 GetAllCrossMarginPairsParams, GetAllIsolatedMarginSymbolParams,
120 GetAllMarginAssetsParams, GetDelistScheduleParams,
121 GetFutureHourlyInterestRateIsIsolatedEnum, GetFutureHourlyInterestRateParams,
122 MarginAccountCancelOrderIsIsolatedEnum, MarginAccountCancelOrderParams,
123 MarginAccountNewOrderIsIsolatedEnum, MarginAccountNewOrderNewOrderRespTypeEnum,
124 MarginAccountNewOrderParams, MarginAccountNewOrderSideEffectTypeEnum,
125 MarginAccountNewOrderSideEnum, MarginAccountNewOrderTimeInForceEnum,
126 MarginAccountNewOrderTypeEnum, QueryCrossMarginAccountDetailsParams,
127 QueryCrossMarginAccountDetailsResponseUserAssetsInner,
128 QueryIsolatedMarginAccountInfoParams,
129 QueryIsolatedMarginAccountInfoResponseAssetsInner,
130 QueryMarginAccountsOpenOrdersIsIsolatedEnum, QueryMarginAccountsOpenOrdersParams,
131 QueryMarginAccountsOrderIsIsolatedEnum, QueryMarginAccountsOrderParams,
132 QueryMarginAccountsTradeListIsIsolatedEnum, QueryMarginAccountsTradeListParams,
133 QueryMarginAccountsTradeListResponseInner, QueryMaxBorrowParams, RestApi,
134 },
135 websocket_streams::{
136 MarginLevelStatusChange, OutboundAccountPosition, UserLiabilityChange,
137 },
138 },
139};
140use chrono::{DateTime, TimeZone, Utc};
141use fnv::FnvHashSet;
142use futures::stream::BoxStream;
143use rust_decimal::Decimal;
144use rustrade_instrument::{
145 Side,
146 asset::name::AssetNameExchange,
147 exchange::ExchangeId,
148 instrument::{kind::InstrumentKindDiscriminant, name::InstrumentNameExchange},
149};
150use serde::{Deserialize, Serialize};
151use smol_str::{SmolStr, format_smolstr};
152use std::{
153 collections::{BTreeMap, HashMap},
154 str::FromStr,
155 sync::{
156 Arc, Mutex,
157 atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
158 },
159 time::Duration,
160};
161use tokio::sync::{mpsc, oneshot};
162use tracing::{debug, error, info, trace, warn};
163
164#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
182#[serde(rename_all = "snake_case")]
183pub enum MarginSideEffect {
184 #[default]
192 AutoBorrowRepay,
193 NoBorrow,
197}
198
199impl MarginSideEffect {
200 pub fn as_binance_str(self) -> &'static str {
202 match self {
203 MarginSideEffect::AutoBorrowRepay => "AUTO_BORROW_REPAY",
204 MarginSideEffect::NoBorrow => "NO_SIDE_EFFECT",
205 }
206 }
207
208 fn as_sdk_enum(self) -> MarginAccountNewOrderSideEffectTypeEnum {
214 match self {
215 MarginSideEffect::AutoBorrowRepay => {
216 MarginAccountNewOrderSideEffectTypeEnum::AutoBorrowRepay
217 }
218 MarginSideEffect::NoBorrow => MarginAccountNewOrderSideEffectTypeEnum::NoSideEffect,
219 }
220 }
221}
222
223#[derive(Clone, Deserialize)]
233pub struct BinanceMarginConfig {
234 api_key: String,
237 secret_key: String,
238 pub testnet: bool,
244 #[serde(default)]
251 pub is_isolated: bool,
252 #[serde(default)]
263 pub isolated_symbols: Vec<InstrumentNameExchange>,
264 #[serde(default)]
268 pub side_effect: MarginSideEffect,
269}
270
271impl std::fmt::Debug for BinanceMarginConfig {
273 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
274 f.debug_struct("BinanceMarginConfig")
275 .field("api_key", &"***")
276 .field("secret_key", &"***")
277 .field("testnet", &self.testnet)
278 .field("is_isolated", &self.is_isolated)
279 .field("isolated_symbols", &self.isolated_symbols)
280 .field("side_effect", &self.side_effect)
281 .finish()
282 }
283}
284
285impl BinanceMarginConfig {
286 pub fn new(
296 api_key: String,
297 secret_key: String,
298 testnet: bool,
299 is_isolated: bool,
300 isolated_symbols: Vec<InstrumentNameExchange>,
301 side_effect: MarginSideEffect,
302 ) -> Self {
303 Self {
304 api_key,
305 secret_key,
306 testnet,
307 is_isolated,
308 isolated_symbols,
309 side_effect,
310 }
311 }
312
313 pub fn cross_margin(api_key: String, secret_key: String) -> Self {
320 Self::new(
321 api_key,
322 secret_key,
323 false,
324 false,
325 Vec::new(),
326 MarginSideEffect::default(),
327 )
328 }
329
330 pub fn isolated(
338 api_key: String,
339 secret_key: String,
340 symbols: Vec<InstrumentNameExchange>,
341 ) -> Self {
342 Self::new(
343 api_key,
344 secret_key,
345 false,
346 true,
347 symbols,
348 MarginSideEffect::default(),
349 )
350 }
351
352 pub fn api_key(&self) -> &str {
354 &self.api_key
355 }
356}
357
358#[derive(Clone)]
479pub struct BinanceMargin {
480 config: Arc<BinanceMarginConfig>,
481 rest: Arc<RestApi>,
483 rest_config: Arc<ConfigurationRestApi>,
487 ws_config: ConfigurationWebsocketApi,
491 rate_limiter: Arc<RateLimitTracker>,
493 known_live: SharedKnownLiveOrders,
496}
497
498impl std::fmt::Debug for BinanceMargin {
499 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
500 f.debug_struct("BinanceMargin")
501 .field("testnet", &self.config.testnet)
502 .field("is_isolated", &self.config.is_isolated)
503 .field("side_effect", &self.config.side_effect)
504 .finish_non_exhaustive()
505 }
506}
507
508impl BinanceMargin {
509 fn remember_live<'a>(
511 &self,
512 orders: impl IntoIterator<Item = &'a Order<ExchangeId, InstrumentNameExchange, Open>>,
513 ) {
514 let mut known = self.known_live.lock();
515 for order in orders {
516 known.live(&order.key, order.quantity, order.kind, &order.state);
517 }
518 }
519
520 #[allow(clippy::expect_used)] fn build_rest_config(config: &BinanceMarginConfig) -> ConfigurationRestApi {
530 ConfigurationRestApi::builder()
531 .api_key(config.api_key.clone())
532 .api_secret(config.secret_key.clone())
533 .base_path(MARGIN_TRADING_REST_API_PROD_URL)
534 .build()
535 .expect("failed to build Binance margin REST configuration")
536 }
537
538 #[allow(clippy::expect_used)] fn build_ws_config(config: &BinanceMarginConfig) -> ConfigurationWebsocketApi {
549 ConfigurationWebsocketApi::builder()
550 .api_key(config.api_key.clone())
551 .api_secret(config.secret_key.clone())
552 .ws_url(SPOT_WS_API_PROD_URL)
556 .build()
557 .expect("failed to build Binance margin WebSocket configuration")
558 }
559
560 fn effective_isolated_set(
573 &self,
574 instruments: &[InstrumentNameExchange],
575 ) -> Vec<InstrumentNameExchange> {
576 if instruments.is_empty() {
577 return self.config.isolated_symbols.clone();
578 }
579 instruments
580 .iter()
581 .filter(|inst| {
582 let in_set = self.config.isolated_symbols.contains(*inst);
583 if !in_set {
584 warn!(
585 instrument = %inst.name(),
586 "BinanceMargin isolated: requested instrument not in configured isolated_symbols — skipping"
587 );
588 }
589 in_set
590 })
591 .cloned()
592 .collect()
593 }
594
595 async fn isolated_account_snapshot(
599 &self,
600 instruments: &[InstrumentNameExchange],
601 ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
602 use futures::{StreamExt as _, TryStreamExt as _};
603
604 let effective = self.effective_isolated_set(instruments);
605 if effective.is_empty() {
606 return Ok(AccountSnapshot::new(
609 ExchangeId::BinanceMargin,
610 Vec::new(),
611 Vec::new(),
612 ));
613 }
614
615 let mut balances_by_symbol = convert_isolated_margin_assets(
617 fetch_isolated_margin_account_info(
618 self.rest.clone(),
619 self.rate_limiter.clone(),
620 effective.clone(),
621 )
622 .await?,
623 );
624
625 let instrument_snapshots: Vec<_> =
628 futures::stream::iter(effective.into_iter().map(|instrument| {
629 fetch_margin_open_orders_for_instrument(
630 self.rest.clone(),
631 self.rate_limiter.clone(),
632 instrument,
633 true,
634 RequestKind::Query,
635 )
636 }))
637 .buffer_unordered(8)
638 .map(|result| {
639 let (inst, listing) = result?;
640 self.remember_live(&listing.orders);
641 let wrapped = listing
642 .orders
643 .into_iter()
644 .map(active_order_snapshot)
645 .collect();
646 let isolated = balances_by_symbol.remove(&inst);
647 if isolated.is_none() {
648 warn!(
652 instrument = %inst.name(),
653 "BinanceMargin isolated: no isolated balance entry returned for instrument — snapshot will carry isolated: None"
654 );
655 }
656 Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
657 inst,
658 wrapped,
659 listing.complete,
660 PositionReport::Unreported,
661 isolated,
662 ))
663 })
664 .try_collect()
665 .await?;
666
667 Ok(AccountSnapshot::new(
670 ExchangeId::BinanceMargin,
671 Vec::new(),
672 instrument_snapshots,
673 ))
674 }
675}
676
677impl ExecutionClient for BinanceMargin {
678 const EXCHANGE: ExchangeId = ExchangeId::BinanceMargin;
679
680 const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] =
683 &[InstrumentKindDiscriminant::Spot];
684 type Config = BinanceMarginConfig;
685 type AccountStream = BoxStream<'static, UnindexedAccountEvent>;
686
687 fn new(config: Self::Config) -> Self {
700 if config.testnet {
701 warn!(
702 "BinanceMarginConfig.testnet = true is ignored: Binance margin has no testnet; \
703 using production endpoints"
704 );
705 }
706 assert!(
707 !(config.is_isolated && config.isolated_symbols.is_empty()),
708 "BinanceMarginConfig: is_isolated = true requires a non-empty isolated_symbols"
709 );
710 let rest_config = Self::build_rest_config(&config);
715 let rest = Arc::new(MarginTradingRestApi::production(rest_config.clone()));
716 let rest_config = Arc::new(rest_config);
717 let ws_config = Self::build_ws_config(&config);
718 Self {
719 config: Arc::new(config),
720 rest,
721 rest_config,
722 ws_config,
723 rate_limiter: Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
724 known_live: KnownLiveOrders::shared(ExchangeId::BinanceMargin),
725 }
726 }
727
728 async fn open_order(
747 &self,
748 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
749 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
750 let instrument = request.key.instrument.clone();
751 let side = request.state.side;
752 let price = request.state.price;
753 let quantity = request.state.quantity;
754 let kind = request.state.kind;
755 let time_in_force = request.state.time_in_force;
756 let cid = request.key.cid.clone();
757
758 let order_key = OrderKey::new(
759 ExchangeId::BinanceMargin,
760 instrument.clone(),
761 request.key.strategy.clone(),
762 cid.clone(),
763 );
764
765 let inactive = |state: UnindexedOrderState| Order {
768 key: order_key.clone(),
769 side,
770 price,
771 quantity,
772 kind,
773 time_in_force,
774 state,
775 };
776
777 let params = match build_new_order_params(
778 instrument.name().to_string(),
779 side,
780 price,
781 quantity,
782 kind,
783 time_in_force,
784 cid.0.to_string(),
785 self.config.side_effect,
786 self.config.is_isolated,
787 ) {
788 Ok(params) => params,
789 Err(BuildOrderError::Unsupported) => {
790 return inactive(OrderState::inactive(OrderError::UnsupportedOrderType(
791 format!(
792 "BinanceMargin does not support OrderKind::{kind:?} with {time_in_force:?}"
793 ),
794 )));
795 }
796 Err(BuildOrderError::Trailing(refused)) => {
797 return inactive(OrderState::inactive(refused.into_order_error()));
798 }
799 Err(BuildOrderError::Build(msg)) => {
800 error!(%msg, "BinanceMargin failed to build new order params");
801 return inactive(OrderState::inactive(OrderError::Rejected(
802 ApiError::OrderRejected(msg),
803 )));
804 }
805 };
806
807 let response =
808 match rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Order, |rest| {
809 let params = params.clone();
810 Box::pin(async move { rest.margin_account_new_order(params).await })
811 })
812 .await
813 {
814 Ok(response) => response,
815 Err(e) => {
816 return inactive(OrderState::inactive(classify_rest_order_error(
817 &e,
818 &instrument,
819 )));
820 }
821 };
822
823 let data = match response.data().await {
824 Ok(data) => data,
825 Err(e) => {
826 return inactive(OrderState::inactive(OrderError::Rejected(
829 ApiError::OrderRejected(e.to_string()),
830 )));
831 }
832 };
833
834 let time_exchange = data
835 .transact_time
836 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
837 .unwrap_or_else(Utc::now);
838
839 let exchange_order_id = match data.order_id {
840 Some(id) => OrderId(format_smolstr!("{id}")),
841 None => {
842 error!("BinanceMargin open_order response missing orderId");
843 return inactive(OrderState::inactive(OrderError::Rejected(
844 ApiError::OrderRejected("open_order response missing orderId".into()),
845 )));
846 }
847 };
848
849 let state = placed_order_state(
852 ExchangeId::BinanceMargin,
853 &instrument,
854 quantity,
855 PlacementResponse {
856 status: data.status.as_deref(),
857 order_id: exchange_order_id,
858 time_exchange,
859 executed_qty: data.executed_qty.as_deref(),
860 cumulative_quote_qty: data.cummulative_quote_qty.as_deref(),
861 },
862 );
863 self.known_live
864 .lock()
865 .placed(&order_key, quantity, kind, &state);
866
867 Order {
868 key: order_key,
869 side,
870 price,
871 quantity,
872 kind,
873 time_in_force,
874 state,
875 }
876 }
877
878 async fn cancel_order(
888 &self,
889 request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
890 ) -> UnindexedOrderResponseCancel {
891 let instrument = request.key.instrument.clone();
892 let key = OrderKey {
893 exchange: request.key.exchange,
894 instrument: instrument.clone(),
895 strategy: request.key.strategy.clone(),
896 cid: request.key.cid.clone(),
897 };
898
899 let params = match build_cancel_order_params(
900 instrument.name().to_string(),
901 request.state.id.as_ref().and_then(VenueOrderId::assigned),
902 &request.key.cid,
903 self.config.is_isolated,
904 ) {
905 Ok(p) => p,
906 Err(e) => {
907 error!(%e, "BinanceMargin failed to build cancel order params");
908 return UnindexedOrderResponseCancel {
909 key,
910 state: Err(OrderError::Rejected(ApiError::OrderRejected(e))),
911 };
912 }
913 };
914
915 let response =
916 match rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Order, |rest| {
917 let params = params.clone();
918 Box::pin(async move { rest.margin_account_cancel_order(params).await })
919 })
920 .await
921 {
922 Ok(response) => response,
923 Err(e) => {
924 return UnindexedOrderResponseCancel {
925 key,
926 state: Err(classify_rest_order_error(&e, &instrument)),
927 };
928 }
929 };
930
931 let data = match response.data().await {
932 Ok(data) => data,
933 Err(e) => {
934 return UnindexedOrderResponseCancel {
937 key,
938 state: Err(OrderError::Rejected(ApiError::OrderRejected(e.to_string()))),
939 };
940 }
941 };
942
943 let time_exchange = Utc::now();
945
946 let exchange_order_id = match data.order_id {
947 Some(id) => OrderId(SmolStr::new(id)),
950 None => {
951 error!("BinanceMargin cancel response missing orderId");
952 return UnindexedOrderResponseCancel {
953 key,
954 state: Err(OrderError::Rejected(ApiError::OrderRejected(
955 "cancel response missing orderId".into(),
956 ))),
957 };
958 }
959 };
960
961 let filled_qty = binance_filled_qty(
962 ExchangeId::BinanceMargin,
963 &exchange_order_id,
964 data.executed_qty.as_deref(),
965 );
966
967 self.known_live.lock().ended(&key.cid);
968 UnindexedOrderResponseCancel {
969 key,
970 state: Ok(Cancelled::new(exchange_order_id, time_exchange, filled_qty)),
971 }
972 }
973
974 async fn account_snapshot(
989 &self,
990 assets: &[AssetNameExchange],
991 instruments: &[InstrumentNameExchange],
992 ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
993 if self.config.is_isolated {
994 return self.isolated_account_snapshot(instruments).await;
995 }
996 let response =
997 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
998 Box::pin(async move {
999 let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
1000 rest.query_cross_margin_account_details(params).await
1001 })
1002 })
1003 .await
1004 .map_err(|e| classify_rest_query_error(&e, None))?;
1005
1006 let account = response.data().await.map_err(response_decode_error)?;
1007
1008 let balances =
1009 filter_and_convert_margin_balances(account.user_assets.unwrap_or_default(), assets);
1010
1011 use futures::{StreamExt as _, TryStreamExt as _};
1016 let instrument_snapshots: Vec<_> =
1017 futures::stream::iter(instruments.iter().cloned().map(|instrument| {
1018 fetch_margin_open_orders_for_instrument(
1019 self.rest.clone(),
1020 self.rate_limiter.clone(),
1021 instrument,
1022 false,
1024 RequestKind::Query,
1025 )
1026 }))
1027 .buffer_unordered(8)
1028 .map(|result| {
1029 let (inst, listing) = result?;
1030 self.remember_live(&listing.orders);
1031 let wrapped = listing
1032 .orders
1033 .into_iter()
1034 .map(active_order_snapshot)
1035 .collect();
1036 Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
1037 inst,
1038 wrapped,
1039 listing.complete,
1040 PositionReport::Unreported,
1041 None,
1042 ))
1043 })
1044 .try_collect()
1045 .await?;
1046
1047 Ok(AccountSnapshot::new(
1048 ExchangeId::BinanceMargin,
1049 balances,
1050 instrument_snapshots,
1051 ))
1052 }
1053
1054 async fn fetch_balances(
1064 &self,
1065 assets: &[AssetNameExchange],
1066 ) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
1067 if self.config.is_isolated {
1068 return Ok(Vec::new());
1069 }
1070 let response =
1071 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1072 Box::pin(async move {
1073 let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
1074 rest.query_cross_margin_account_details(params).await
1075 })
1076 })
1077 .await
1078 .map_err(|e| classify_rest_query_error(&e, None))?;
1079
1080 let account = response.data().await.map_err(response_decode_error)?;
1081
1082 Ok(filter_and_convert_margin_balances(
1083 account.user_assets.unwrap_or_default(),
1084 assets,
1085 ))
1086 }
1087
1088 async fn fetch_open_orders(
1107 &self,
1108 instruments: &[InstrumentNameExchange],
1109 ) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
1110 use futures::{StreamExt as _, TryStreamExt as _};
1111
1112 if self.config.is_isolated {
1116 let effective = self.effective_isolated_set(instruments);
1117 let capacity = effective.len();
1118 return futures::stream::iter(effective.into_iter().map(|instrument| {
1119 fetch_margin_open_orders_for_instrument(
1120 self.rest.clone(),
1121 self.rate_limiter.clone(),
1122 instrument,
1123 true,
1124 RequestKind::Query,
1125 )
1126 }))
1127 .buffer_unordered(8)
1128 .try_fold(
1129 Vec::with_capacity(capacity),
1130 |mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>,
1131 (_, listing)| async move {
1132 acc.extend(listing.orders);
1133 Ok(acc)
1134 },
1135 )
1136 .await
1137 .inspect(|orders| self.remember_live(orders));
1138 }
1139
1140 if instruments.is_empty() {
1143 let orders =
1144 fetch_margin_all_open_orders(self.rest.clone(), self.rate_limiter.clone(), false)
1145 .await?;
1146 self.remember_live(&orders);
1147 return Ok(orders);
1148 }
1149 futures::stream::iter(instruments.iter().cloned().map(|instrument| {
1150 fetch_margin_open_orders_for_instrument(
1151 self.rest.clone(),
1152 self.rate_limiter.clone(),
1153 instrument,
1154 false,
1155 RequestKind::Query,
1156 )
1157 }))
1158 .buffer_unordered(8)
1159 .try_fold(
1160 Vec::with_capacity(instruments.len()),
1161 |mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, (_, listing)| async move {
1162 acc.extend(listing.orders);
1163 Ok(acc)
1164 },
1165 )
1166 .await
1167 .inspect(|orders| self.remember_live(orders))
1168 }
1169
1170 async fn fetch_trades(
1199 &self,
1200 start: DateTime<Utc>,
1201 end: DateTime<Utc>,
1202 instruments: &[InstrumentNameExchange],
1203 ) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
1204 use futures::StreamExt as _;
1205
1206 if start > end {
1207 return Ok(TradesRead::complete(Vec::new()));
1208 }
1209
1210 let effective: Vec<InstrumentNameExchange> = if self.config.is_isolated {
1213 self.effective_isolated_set(instruments)
1214 } else {
1215 instruments.to_vec()
1216 };
1217 if effective.is_empty() {
1218 debug!(
1221 is_isolated = self.config.is_isolated,
1222 "BinanceMargin fetch_trades: empty effective instrument set — returning empty result"
1223 );
1224 return Ok(TradesRead::complete(Vec::new()));
1225 }
1226 let span = MyTradesFrom::Span {
1228 start: start.timestamp_millis(),
1229 end: end.timestamp_millis(),
1230 };
1231 let is_isolated = self.config.is_isolated;
1232 let mut all_trades = Vec::new();
1233
1234 let mut stream = futures::stream::iter(effective.into_iter().map(|inst| {
1237 let rest = self.rest.clone();
1238 let rate_limiter = self.rate_limiter.clone();
1239 async move {
1240 let pages = paginate_margin_my_trades(
1241 &rest,
1242 &rate_limiter,
1243 &inst,
1244 span,
1245 is_isolated,
1246 RequestKind::Query,
1247 )
1248 .await?;
1249 Ok::<_, UnindexedClientError>((inst, pages))
1250 }
1251 }))
1252 .buffer_unordered(8);
1253 while let Some(result) = stream.next().await {
1254 let (instrument, trades_data) = result?;
1255 for t in trades_data {
1256 if let Some(trade) = convert_margin_trade(&t, &instrument)
1257 && (start..=end).contains(&trade.time_exchange)
1258 {
1259 all_trades.push(trade);
1260 }
1261 }
1262 }
1263
1264 Ok(TradesRead::complete(all_trades))
1265 }
1266
1267 async fn account_stream(
1403 &self,
1404 _assets: &[AssetNameExchange],
1407 instruments: &[InstrumentNameExchange],
1408 ) -> Result<Self::AccountStream, UnindexedClientError> {
1409 let (tx, rx) = mpsc::unbounded_channel::<UnindexedAccountEvent>();
1410 let dedup = new_dedup_cache();
1411 let rest = self.rest.clone();
1412 let rest_config = self.rest_config.clone();
1413 let ws_config = self.ws_config.clone();
1414 let rate_limiter = self.rate_limiter.clone();
1415 let known_live = self.known_live.clone();
1416
1417 let cm_handle = if self.config.is_isolated {
1418 let symbols = self.config.isolated_symbols.clone();
1422 debug_assert!(
1425 symbols.iter().all(|i| i.name().len() <= 23),
1426 "instrument name exceeds SmolStr inline capacity: {:?}",
1427 symbols.iter().find(|i| i.name().len() > 23)
1428 );
1429
1430 let assets = fetch_isolated_margin_account_info(
1436 rest.clone(),
1437 rate_limiter.clone(),
1438 symbols.clone(),
1439 )
1440 .await?;
1441 let base_quote = Arc::new(build_base_quote_map(&assets));
1442 for sym in &symbols {
1443 if !base_quote.contains_key(sym) {
1444 warn!(
1445 symbol = %sym.name(),
1446 "BinanceMargin isolated: account info returned no base/quote for configured \
1447 symbol — its live per-pair balance frames will be dropped (fills/orders \
1448 unaffected; balances available via account_snapshot)"
1449 );
1450 }
1451 }
1452
1453 let tokens = acquire_all_isolated_tokens(&rest_config, &rate_limiter, &symbols).await?;
1457 rate_limiter.before_stream_open().await?;
1460 let live = isolated_connect_and_subscribe(
1461 &ws_config,
1462 &tx,
1463 &dedup,
1464 &known_live,
1465 &base_quote,
1466 &tokens,
1467 )
1468 .await
1469 .map_err(|e| {
1470 rate_limiter.stream_open_error(&e, |e| {
1471 UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
1472 })
1473 })?;
1474
1475 tokio::spawn(isolated_connection_manager(
1476 tx,
1477 dedup,
1478 ws_config,
1479 rest_config,
1480 rest,
1481 rate_limiter,
1482 known_live,
1483 symbols,
1484 base_quote,
1485 Some(live),
1486 ))
1487 } else {
1488 let instruments = instruments.to_vec();
1490 debug_assert!(
1494 instruments.iter().all(|i| i.name().len() <= 23),
1495 "instrument name exceeds SmolStr inline capacity: {:?}",
1496 instruments.iter().find(|i| i.name().len() > 23)
1497 );
1498
1499 let initial_token =
1505 acquire_user_listen_token(&rest_config, &rate_limiter, None).await?;
1506 rate_limiter.before_stream_open().await?;
1508 let initial_ws = connect_margin_ws(&ws_config).await.map_err(|e| {
1509 rate_limiter.stream_open_error(&e, |e| {
1510 UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
1511 })
1512 })?;
1513
1514 let risk = RiskStreamTask::spawn(
1518 SapiRiskListenKeys {
1519 rest: rest.clone(),
1520 rate_limiter: rate_limiter.clone(),
1521 },
1522 RISK_STREAM_URL.to_owned(),
1523 tx.clone(),
1524 dedup.clone(),
1525 );
1526 let manager = margin_connection_manager(
1527 tx,
1528 dedup,
1529 ws_config,
1530 rest_config,
1531 rest,
1532 rate_limiter,
1533 known_live,
1534 instruments,
1535 Some((initial_ws, initial_token)),
1536 );
1537 tokio::spawn(async move {
1538 let _risk = risk;
1539 manager.await;
1540 })
1541 };
1542
1543 let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
1544 let guarded_stream = AbortOnDropStream::new(rx_stream, cm_handle);
1545 Ok(futures::StreamExt::boxed(guarded_stream))
1546 }
1547}
1548
1549impl OrderStatusClient for BinanceMargin {
1566 async fn fetch_ended_orders(
1567 &self,
1568 orders: &[UnindexedOrderKey],
1569 ) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError> {
1570 fetch_ended_by_key(orders, |key| {
1571 let recorded_kind = self.known_live.lock().kind_of(&key.cid);
1572 fetch_margin_order_lookup(
1573 self.rest.clone(),
1574 self.rate_limiter.clone(),
1575 key,
1576 self.config.is_isolated,
1577 RequestKind::Query,
1578 recorded_kind,
1579 )
1580 })
1581 .await
1582 }
1583}
1584
1585const HOURS_PER_YEAR: u32 = 24 * 365;
1591
1592impl ShortabilityClient for BinanceMargin {
1593 async fn fetch_shortability(
1620 &self,
1621 instrument: &InstrumentNameExchange,
1622 ) -> Result<Shortability, UnindexedClientError> {
1623 let pair = self.fetch_margin_pair(instrument).await?;
1624 let (borrowable, hourly_rate, delisting_at) = tokio::try_join!(
1625 self.fetch_asset_borrowable(&pair.base),
1626 self.fetch_next_hourly_interest_rate(&pair.base),
1627 self.fetch_delisting_at(instrument, &pair),
1628 )?;
1629
1630 let mut shortability = Shortability::new();
1631 if let Some(delisting_at) = delisting_at {
1632 shortability = shortability.with_delisting_at(delisting_at);
1633 }
1634 let not_delisted = Some(delisting_at.is_none_or(|delisting_at| delisting_at > Utc::now()));
1638 let flags = [
1639 pair.is_margin_trade,
1640 pair.is_sell_allowed,
1641 borrowable,
1642 not_delisted,
1643 ];
1644 if flags.contains(&Some(false)) {
1645 shortability = shortability.with_shortable(false);
1646 } else if flags.iter().all(|flag| *flag == Some(true)) {
1647 shortability = shortability.with_shortable(true);
1648 }
1649 if let Some(hourly_rate) = hourly_rate {
1650 shortability = shortability.with_fee_rate(hourly_rate * Decimal::from(HOURS_PER_YEAR));
1651 }
1652 Ok(shortability)
1653 }
1654}
1655
1656impl BorrowCapacityClient for BinanceMargin {
1657 async fn fetch_borrow_capacity(
1674 &self,
1675 instrument: &InstrumentNameExchange,
1676 side: Side,
1677 ) -> Result<BorrowCapacity, UnindexedClientError> {
1678 let pair = self.fetch_margin_pair(instrument).await?;
1679 let asset = match side {
1680 Side::Buy => pair.quote,
1681 Side::Sell => pair.base,
1682 };
1683 let isolated_symbol = self
1684 .config
1685 .is_isolated
1686 .then(|| instrument.name().to_string());
1687 let response =
1688 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1689 let asset = asset.name().to_string();
1690 let isolated_symbol = isolated_symbol.clone();
1691 Box::pin(async move {
1692 let params = QueryMaxBorrowParams::builder(asset)
1693 .isolated_symbol(isolated_symbol)
1694 .build()?;
1695 rest.query_max_borrow(params).await
1696 })
1697 })
1698 .await
1699 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
1700 let max_borrow = response.data().await.map_err(response_decode_error)?;
1701
1702 let mut capacity = BorrowCapacity::new(asset);
1703 if let Some(amount) = optional_decimal("amount", max_borrow.amount.as_deref())? {
1704 capacity = capacity.with_borrowable_now(amount);
1705 }
1706 if let Some(limit) = optional_decimal("borrowLimit", max_borrow.borrow_limit.as_deref())? {
1707 capacity = capacity.with_account_limit(limit);
1708 }
1709 Ok(capacity)
1710 }
1711}
1712
1713#[derive(Debug)]
1715struct MarginPair {
1716 base: AssetNameExchange,
1717 quote: AssetNameExchange,
1718 is_margin_trade: Option<bool>,
1719 is_sell_allowed: Option<bool>,
1720}
1721
1722impl BinanceMargin {
1723 async fn fetch_margin_pair(
1727 &self,
1728 instrument: &InstrumentNameExchange,
1729 ) -> Result<MarginPair, UnindexedClientError> {
1730 let symbol = instrument.name().to_string();
1731 let listed = if self.config.is_isolated {
1732 let response =
1733 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1734 let symbol = symbol.clone();
1735 Box::pin(async move {
1736 let params = GetAllIsolatedMarginSymbolParams::builder()
1737 .symbol(symbol)
1738 .build()?;
1739 rest.get_all_isolated_margin_symbol(params).await
1740 })
1741 })
1742 .await
1743 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
1744 response
1745 .data()
1746 .await
1747 .map_err(response_decode_error)?
1748 .into_iter()
1749 .find(|pair| pair.symbol.as_deref() == Some(symbol.as_str()))
1750 .map(|pair| {
1751 (
1752 pair.base,
1753 pair.quote,
1754 pair.is_margin_trade,
1755 pair.is_sell_allowed,
1756 )
1757 })
1758 } else {
1759 let response =
1760 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1761 let symbol = symbol.clone();
1762 Box::pin(async move {
1763 let params = GetAllCrossMarginPairsParams::builder()
1764 .symbol(symbol)
1765 .build()?;
1766 rest.get_all_cross_margin_pairs(params).await
1767 })
1768 })
1769 .await
1770 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
1771 response
1772 .data()
1773 .await
1774 .map_err(response_decode_error)?
1775 .into_iter()
1776 .find(|pair| pair.symbol.as_deref() == Some(symbol.as_str()))
1777 .map(|pair| {
1778 (
1779 pair.base,
1780 pair.quote,
1781 pair.is_margin_trade,
1782 pair.is_sell_allowed,
1783 )
1784 })
1785 };
1786 let margin = if self.config.is_isolated {
1787 "isolated"
1788 } else {
1789 "cross"
1790 };
1791 let Some((base, quote, is_margin_trade, is_sell_allowed)) = listed else {
1792 return Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(
1793 instrument.clone(),
1794 format!("Binance lists no {margin} margin pair {symbol}"),
1795 )));
1796 };
1797 let (Some(base), Some(quote)) = (base, quote) else {
1798 return Err(UnindexedClientError::Internal(format!(
1799 "Binance lists {margin} margin pair {symbol} without its base or quote asset"
1800 )));
1801 };
1802 Ok(MarginPair {
1803 base: AssetNameExchange::new(base),
1804 quote: AssetNameExchange::new(quote),
1805 is_margin_trade,
1806 is_sell_allowed,
1807 })
1808 }
1809
1810 async fn fetch_asset_borrowable(
1813 &self,
1814 asset: &AssetNameExchange,
1815 ) -> Result<Option<bool>, UnindexedClientError> {
1816 let name = asset.name().to_string();
1817 let response =
1818 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1819 let name = name.clone();
1820 Box::pin(async move {
1821 let params = GetAllMarginAssetsParams::builder().asset(name).build()?;
1822 rest.get_all_margin_assets(params).await
1823 })
1824 })
1825 .await
1826 .map_err(|e| classify_rest_query_error(&e, None))?;
1827 Ok(response
1828 .data()
1829 .await
1830 .map_err(response_decode_error)?
1831 .into_iter()
1832 .find(|listed| listed.asset_name.as_deref() == Some(name.as_str()))
1833 .and_then(|listed| listed.is_borrowable))
1834 }
1835
1836 async fn fetch_delisting_at(
1841 &self,
1842 instrument: &InstrumentNameExchange,
1843 pair: &MarginPair,
1844 ) -> Result<Option<DateTime<Utc>>, UnindexedClientError> {
1845 let response =
1846 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1847 Box::pin(async move {
1848 let params = GetDelistScheduleParams::builder().build()?;
1849 rest.get_delist_schedule(params).await
1850 })
1851 })
1852 .await
1853 .map_err(|e| classify_rest_query_error(&e, None))?;
1854 let schedule = response.data().await.map_err(response_decode_error)?;
1855
1856 let symbol = instrument.name().as_str();
1857 let (base, quote) = (pair.base.name().as_str(), pair.quote.name().as_str());
1858 let mut earliest: Option<DateTime<Utc>> = None;
1859 for entry in schedule {
1860 let scheduled = if self.config.is_isolated {
1861 entry
1862 .isolated_margin_symbols
1863 .iter()
1864 .flatten()
1865 .any(|listed| listed == symbol)
1866 } else {
1867 entry
1868 .cross_margin_assets
1869 .iter()
1870 .flatten()
1871 .any(|listed| listed == base || listed == quote)
1872 };
1873 if !scheduled {
1874 continue;
1875 }
1876 let delisting_at = entry
1878 .delist_time
1879 .and_then(DateTime::from_timestamp_millis)
1880 .ok_or_else(|| {
1881 UnindexedClientError::Internal(format!(
1882 "decoding Binance response: delist-schedule lists {symbol} with delistTime \
1883 {:?}",
1884 entry.delist_time
1885 ))
1886 })?;
1887 earliest = Some(earliest.map_or(delisting_at, |earliest| earliest.min(delisting_at)));
1888 }
1889 Ok(earliest)
1890 }
1891
1892 async fn fetch_next_hourly_interest_rate(
1896 &self,
1897 asset: &AssetNameExchange,
1898 ) -> Result<Option<Decimal>, UnindexedClientError> {
1899 let name = asset.name().to_string();
1900 let is_isolated = self.config.is_isolated;
1901 let response =
1902 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1903 let name = name.clone();
1904 Box::pin(async move {
1905 let params = GetFutureHourlyInterestRateParams::builder(
1906 name,
1907 IsolatedFlag::from_flag(is_isolated),
1908 )
1909 .build()?;
1910 rest.get_future_hourly_interest_rate(params).await
1911 })
1912 })
1913 .await
1914 .map_err(|e| classify_rest_query_error(&e, None))?;
1915 let rate = response
1916 .data()
1917 .await
1918 .map_err(response_decode_error)?
1919 .into_iter()
1920 .find(|listed| listed.asset.as_deref() == Some(name.as_str()))
1921 .and_then(|listed| listed.next_hourly_interest_rate);
1922 optional_decimal("nextHourlyInterestRate", rate.as_deref())
1923 }
1924}
1925
1926fn optional_decimal(
1929 field: &str,
1930 value: Option<&str>,
1931) -> Result<Option<Decimal>, UnindexedClientError> {
1932 value
1933 .map(|value| {
1934 Decimal::from_str(value).map_err(|error| {
1935 UnindexedClientError::Internal(format!(
1936 "decoding Binance response: {field} {value:?}: {error}"
1937 ))
1938 })
1939 })
1940 .transpose()
1941}
1942
1943const TOKEN_RENEW_MARGIN_SECS: i64 = 300; const TOKEN_MIN_LIFETIME_SECS: u64 = 60;
1951
1952struct UserListenToken {
1954 token: String,
1955 expiration_time_ms: i64,
1956}
1957
1958fn build_listen_token_query(
1962 symbol: Option<&InstrumentNameExchange>,
1963) -> BTreeMap<String, serde_json::Value> {
1964 let mut query = BTreeMap::new();
1965 if let Some(sym) = symbol {
1966 query.insert(
1967 "isIsolated".to_string(),
1968 serde_json::Value::String("TRUE".to_string()),
1969 );
1970 query.insert(
1971 "symbol".to_string(),
1972 serde_json::Value::String(sym.name().to_string()),
1973 );
1974 }
1975 query
1976}
1977
1978#[derive(Deserialize)]
1980struct UserListenTokenResponse {
1981 #[serde(alias = "listenToken")]
1983 token: String,
1984 #[serde(rename = "expirationTime")]
1985 expiration_time: i64,
1986}
1987
1988async fn acquire_user_listen_token(
2003 rest_config: &Arc<ConfigurationRestApi>,
2004 rate_limiter: &RateLimitTracker,
2005 symbol: Option<&InstrumentNameExchange>,
2006) -> Result<UserListenToken, UnindexedClientError> {
2007 let query = build_listen_token_query(symbol);
2010 let response = rest_call_with_retry(rest_config, rate_limiter, RequestKind::Essential, |cfg| {
2011 let query = query.clone();
2012 Box::pin(async move {
2013 binance_sdk::common::utils::send_request::<UserListenTokenResponse>(
2014 &cfg,
2015 "/sapi/v1/userListenToken",
2016 reqwest::Method::POST,
2017 query,
2020 BTreeMap::new(),
2021 None,
2022 true, )
2024 .await
2025 })
2026 })
2027 .await
2028 .map_err(|e| classify_rest_query_error(&e, symbol))?;
2029
2030 let data = response.data().await.map_err(response_decode_error)?;
2031 Ok(UserListenToken {
2032 token: data.token,
2033 expiration_time_ms: data.expiration_time,
2034 })
2035}
2036
2037fn token_renew_after(expiration_time_ms: i64) -> Duration {
2040 let now_ms = Utc::now().timestamp_millis();
2041 let remaining_secs = expiration_time_ms
2045 .saturating_sub(now_ms)
2046 .saturating_div(1_000)
2047 .saturating_sub(TOKEN_RENEW_MARGIN_SECS);
2048 let secs = u64::try_from(remaining_secs).unwrap_or(0);
2049 if secs < TOKEN_MIN_LIFETIME_SECS {
2052 warn!(
2053 computed_secs = secs,
2054 floor_secs = TOKEN_MIN_LIFETIME_SECS,
2055 "BinanceMargin token renewal wait clamped to floor (near-expiry or odd expirationTime)"
2056 );
2057 }
2058 Duration::from_secs(secs.max(TOKEN_MIN_LIFETIME_SECS))
2059}
2060
2061async fn connect_margin_ws(
2068 ws_config: &ConfigurationWebsocketApi,
2069) -> anyhow::Result<DisconnectOnDrop<Arc<WsApiBase>>> {
2070 let ws = WsApiBase::new(ws_config.clone(), vec![]);
2071 tokio::time::timeout(
2074 Duration::from_secs(CONNECT_TIMEOUT_SECS),
2075 ws.clone().connect(),
2076 )
2077 .await
2078 .map_err(|_| {
2079 anyhow::anyhow!("BinanceMargin WS connect timed out after {CONNECT_TIMEOUT_SECS}s")
2080 })??;
2081 Ok(DisconnectOnDrop::new(ws, "BinanceMargin account stream"))
2083}
2084
2085async fn subscribe_listen_token(ws: &Arc<WsApiBase>, token: &str) -> anyhow::Result<()> {
2092 let mut payload = BTreeMap::new();
2095 payload.insert(
2096 "listenToken".to_string(),
2097 serde_json::Value::String(token.to_string()),
2098 );
2099 ws.send_message::<serde_json::Value>(
2100 "userDataStream.subscribe.listenToken",
2101 payload,
2102 WebsocketMessageSendOptions::new(),
2103 )
2104 .await
2105 .map_err(subscribe_error)?;
2106 Ok(())
2107}
2108
2109fn subscribe_error(e: WebsocketError) -> anyhow::Error {
2112 let message = format!("userDataStream.subscribe.listenToken failed: {e}");
2113 anyhow::Error::new(e).context(message)
2114}
2115
2116async fn subscribe_listen_token_capture(
2129 ws: &Arc<WsApiBase>,
2130 token: &str,
2131) -> anyhow::Result<Option<i64>> {
2132 #[derive(Deserialize)]
2135 struct SubscribeAck {
2136 #[serde(rename = "subscriptionId", default)]
2137 subscription_id: Option<i64>,
2138 }
2139
2140 let mut payload = BTreeMap::new();
2141 payload.insert(
2142 "listenToken".to_string(),
2143 serde_json::Value::String(token.to_string()),
2144 );
2145 let result = ws
2146 .send_message::<SubscribeAck>(
2147 "userDataStream.subscribe.listenToken",
2148 payload,
2149 WebsocketMessageSendOptions::new(),
2150 )
2151 .await
2152 .map_err(subscribe_error)?;
2153 let ack = match result {
2154 SendWebsocketMessageResult::Single(resp) => resp.data()?,
2155 SendWebsocketMessageResult::Multiple(responses) => responses
2157 .into_iter()
2158 .next()
2159 .ok_or_else(|| anyhow::anyhow!("empty subscribe ack"))?
2160 .data()?,
2161 };
2162 Ok(ack.subscription_id)
2163}
2164
2165fn cross_account_position_handler(
2169 position: OutboundAccountPosition,
2170 _subscription_id: Option<i64>,
2171 buf: &mut Vec<UnindexedAccountEvent>,
2172) {
2173 convert_margin_account_position(position, buf);
2174}
2175
2176fn cross_notice_instrument(_subscription_id: Option<i64>) -> Option<InstrumentNameExchange> {
2180 None
2181}
2182
2183fn isolated_notice_instrument(
2189 subscription_id: Option<i64>,
2190 sub_map: &Mutex<HashMap<i64, InstrumentNameExchange>>,
2191) -> Option<InstrumentNameExchange> {
2192 let Some(sub_id) = subscription_id else {
2193 warn!(
2194 "BinanceMargin isolated: MARGIN_LEVEL_STATUS_CHANGE without subscriptionId — \
2195 delivering the notice without an instrument"
2196 );
2197 return None;
2198 };
2199 let instrument = sub_map
2200 .lock()
2201 .unwrap_or_else(|p| p.into_inner())
2202 .get(&sub_id)
2203 .cloned();
2204 if instrument.is_none() {
2205 warn!(
2206 subscription_id = sub_id,
2207 "BinanceMargin isolated: MARGIN_LEVEL_STATUS_CHANGE for unmapped subscriptionId — \
2208 delivering the notice without an instrument"
2209 );
2210 }
2211 instrument
2212}
2213
2214#[cfg(test)]
2220fn convert_margin_user_data_events(frame: &str, buf: &mut Vec<UnindexedAccountEvent>) -> bool {
2221 convert_margin_user_data_events_with(
2222 frame,
2223 buf,
2224 &mut cross_account_position_handler,
2225 &mut cross_notice_instrument,
2226 )
2227}
2228
2229fn convert_margin_user_data_events_with(
2246 frame: &str,
2247 buf: &mut Vec<UnindexedAccountEvent>,
2248 handle_position: &mut impl FnMut(
2249 OutboundAccountPosition,
2250 Option<i64>,
2251 &mut Vec<UnindexedAccountEvent>,
2252 ),
2253 notice_instrument: &mut impl FnMut(Option<i64>) -> Option<InstrumentNameExchange>,
2254) -> bool {
2255 let (subscription_id, event_type, event_raw) = match parse_user_data_frame(frame) {
2256 UserDataFrame::Response => return false,
2258 UserDataFrame::Event {
2259 subscription_id,
2260 event_type,
2261 event,
2262 } => (subscription_id, event_type, event),
2263 UserDataFrame::Unrecognised => {
2264 static SEEN: AtomicU64 = AtomicU64::new(0);
2265 log_unrecognised_frame("BinanceMargin", &SEEN, frame);
2266 return false;
2267 }
2268 };
2269 match event_type {
2270 "executionReport" => {
2271 match serde_json::from_str::<ExecutionReport>(event_raw) {
2274 Ok(report) => {
2275 debug!(
2279 symbol = report.symbol.as_deref().unwrap_or("?"),
2280 order_id = ?report.order_id,
2281 client_order_id = report.client_order_id.as_deref().unwrap_or("?"),
2282 order_type = report.order_type.as_deref().unwrap_or("?"),
2283 execution_type = report.execution_type.as_deref().unwrap_or("?"),
2284 order_status = report.order_status.as_deref().unwrap_or("?"),
2285 "BinanceMargin executionReport"
2286 );
2287 convert_execution_report(&report, ExchangeId::BinanceMargin, buf);
2288 }
2289 Err(e) => {
2290 warn!(error = %e, frame = frame_excerpt(event_raw), "BinanceMargin: undeserializable executionReport, dropping")
2291 }
2292 }
2293 false
2294 }
2295 "outboundAccountPosition" => {
2296 match serde_json::from_str::<OutboundAccountPosition>(event_raw) {
2297 Ok(position) => handle_position(position, subscription_id, buf),
2299 Err(e) => {
2300 warn!(error = %e, frame = frame_excerpt(event_raw), "BinanceMargin: undeserializable outboundAccountPosition, dropping")
2301 }
2302 }
2303 false
2304 }
2305 "balanceUpdate" => {
2306 false
2310 }
2311 "USER_LIABILITY_CHANGE" | "userLiabilityChange" => {
2315 match serde_json::from_str::<UserLiabilityChange>(event_raw) {
2322 Ok(c) => log_liability_change(&c, "user data"),
2323 Err(e) => warn!(
2324 error = %e,
2325 frame = frame_excerpt(event_raw),
2326 "BinanceMargin: undeserializable USER_LIABILITY_CHANGE, dropping"
2327 ),
2328 }
2329 false
2330 }
2331 "MARGIN_LEVEL_STATUS_CHANGE" | "marginLevelStatusChange" => {
2332 match serde_json::from_str::<MarginLevelStatusChange>(event_raw) {
2335 Ok(change) => {
2336 log_margin_level_change(&change, "user data");
2337 buf.extend(margin_level_notice(
2338 change,
2339 notice_instrument(subscription_id),
2340 ));
2341 }
2342 Err(e) => warn!(
2343 error = %e,
2344 frame = frame_excerpt(event_raw),
2345 "BinanceMargin: undeserializable MARGIN_LEVEL_STATUS_CHANGE, dropping"
2346 ),
2347 }
2348 false
2349 }
2350 "eventStreamTerminated" => {
2351 warn!("BinanceMargin user data stream terminated by exchange, signalling reconnect");
2352 true
2353 }
2354 "listStatus" => {
2355 trace!("BinanceMargin ignoring listStatus user data event");
2358 false
2359 }
2360 other => {
2361 static SEEN: UnhandledEvents = UnhandledEvents::new();
2362 log_unhandled_event("BinanceMargin", &SEEN, other, event_raw);
2363 false
2364 }
2365 }
2366}
2367
2368fn convert_margin_account_position(
2373 position: OutboundAccountPosition,
2374 buf: &mut Vec<UnindexedAccountEvent>,
2375) {
2376 let time_exchange = position
2377 .u
2378 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
2379 .unwrap_or_else(Utc::now);
2380
2381 for b in position.b_uppercase.unwrap_or_default() {
2382 let asset = match b.a {
2383 Some(a) => AssetNameExchange::new(a),
2384 None => {
2385 warn!("BinanceMargin account position entry missing asset name");
2386 continue;
2387 }
2388 };
2389 let free = match b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2390 Some(v) => v,
2391 None => {
2392 warn!(%asset, "BinanceMargin account position missing/unparseable 'free' field");
2393 continue;
2394 }
2395 };
2396 let locked = match b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2397 Some(v) => v,
2398 None => {
2399 warn!(%asset, "BinanceMargin account position missing/unparseable 'locked' field");
2400 continue;
2401 }
2402 };
2403 let update =
2404 AssetBalanceUpdate::new(asset, BalanceUpdate::new(free, locked), time_exchange);
2405 buf.push(UnindexedAccountEvent::new(
2406 ExchangeId::BinanceMargin,
2407 AccountEventKind::BalanceStreamUpdate(
2408 rustrade_integration::collection::snapshot::Snapshot::new(update),
2409 ),
2410 ));
2411 }
2412}
2413
2414fn route_isolated_account_position(
2430 position: OutboundAccountPosition,
2431 subscription_id: Option<i64>,
2432 sub_map: &Mutex<HashMap<i64, InstrumentNameExchange>>,
2433 base_quote: &HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>,
2434 buf: &mut Vec<UnindexedAccountEvent>,
2435) {
2436 let Some(sub_id) = subscription_id else {
2437 warn!(
2438 "BinanceMargin isolated: outboundAccountPosition without subscriptionId — dropping \
2439 (per-instrument balance routing requires it)"
2440 );
2441 return;
2442 };
2443 let instrument = {
2444 let map = sub_map.lock().unwrap_or_else(|p| p.into_inner());
2447 match map.get(&sub_id) {
2448 Some(inst) => inst.clone(),
2449 None => {
2450 warn!(
2451 subscription_id = sub_id,
2452 "BinanceMargin isolated: outboundAccountPosition for unmapped subscriptionId — dropping"
2453 );
2454 return;
2455 }
2456 }
2457 };
2458 let Some((base_asset, quote_asset)) = base_quote.get(&instrument) else {
2459 warn!(
2460 instrument = %instrument.name(),
2461 "BinanceMargin isolated: no base/quote known for instrument — dropping balance frame"
2462 );
2463 return;
2464 };
2465
2466 let time_exchange = position
2467 .u
2468 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
2469 .unwrap_or_else(Utc::now);
2470
2471 for b in position.b_uppercase.unwrap_or_default() {
2472 let Some(asset) = b.a.as_deref() else {
2473 warn!(instrument = %instrument.name(), "BinanceMargin isolated account position entry missing asset name");
2474 continue;
2475 };
2476 if asset != base_asset.name().as_str() && asset != quote_asset.name().as_str() {
2478 debug!(%asset, instrument = %instrument.name(), "BinanceMargin isolated account position: skipping asset outside the pair");
2479 continue;
2480 }
2481 let (Some(free), Some(locked)) = (
2482 b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()),
2483 b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()),
2484 ) else {
2485 warn!(%asset, instrument = %instrument.name(), "BinanceMargin isolated account position missing/unparseable free/locked");
2486 continue;
2487 };
2488 buf.push(UnindexedAccountEvent::new(
2489 ExchangeId::BinanceMargin,
2490 AccountEventKind::InstrumentBalanceUpdate(InstrumentBalanceUpdate::new(
2491 instrument.clone(),
2492 AssetBalanceUpdate::new(
2493 AssetNameExchange::new(asset),
2494 BalanceUpdate::new(free, locked),
2495 time_exchange,
2496 ),
2497 )),
2498 ));
2499 }
2500}
2501
2502fn register_user_data_listener(
2515 ws: &Arc<WsApiBase>,
2516 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
2517 dedup: SharedDedupCache,
2518 known: SharedKnownLiveOrders,
2519 heartbeat_flag: Arc<AtomicBool>,
2520 signal_tx: oneshot::Sender<()>,
2521 mut handle_position: impl FnMut(
2522 OutboundAccountPosition,
2523 Option<i64>,
2524 &mut Vec<UnindexedAccountEvent>,
2525 ) + Send
2526 + 'static,
2527 mut notice_instrument: impl FnMut(Option<i64>) -> Option<InstrumentNameExchange> + Send + 'static,
2528) -> UnsubscribeOnDrop {
2529 let mut signal_tx_opt = Some(signal_tx);
2530 let mut event_tx = Some(tx);
2531 let mut event_buf = Vec::with_capacity(32);
2535
2536 UnsubscribeOnDrop::new(ws.common.events.subscribe(move |event| {
2540 let Some(ref sender) = event_tx else { return };
2541 match event {
2542 WebsocketEvent::Message(json_str) => {
2543 heartbeat_flag.store(true, Ordering::Release);
2544 let terminated = convert_margin_user_data_events_with(
2547 &json_str,
2548 &mut event_buf,
2549 &mut handle_position,
2550 &mut notice_instrument,
2551 );
2552 for ev in event_buf.drain(..) {
2553 if let Some(key) = dedup_key_from_event(&ev)
2554 && is_duplicate(&dedup, key)
2555 {
2556 trace!("BinanceMargin dedup: skipping duplicate event");
2557 continue;
2558 }
2559 let held = KnownLiveOrders::observes(&ev.kind).then(|| {
2563 let mut held = known.lock();
2564 held.observe(&ev);
2565 held
2566 });
2567 if sender.send(ev).is_err() {
2568 drop(held);
2569 warn!("BinanceMargin account_stream receiver dropped, suppressing sends");
2570 event_tx.take();
2571 if let Some(s) = signal_tx_opt.take() {
2572 let _ = s.send(());
2573 }
2574 return;
2575 }
2576 }
2577 if terminated {
2578 event_tx.take();
2579 if let Some(s) = signal_tx_opt.take() {
2580 let _ = s.send(());
2581 }
2582 }
2583 }
2584 WebsocketEvent::Ping | WebsocketEvent::Pong => {
2585 heartbeat_flag.store(true, Ordering::Release);
2586 }
2587 WebsocketEvent::Error(e) => {
2588 warn!(%e, "BinanceMargin WebSocket error, will attempt reconnect");
2589 event_tx.take();
2590 if let Some(s) = signal_tx_opt.take() {
2591 let _ = s.send(());
2592 }
2593 }
2594 WebsocketEvent::Close(code, reason) => {
2595 warn!(code, %reason, "BinanceMargin WebSocket closed");
2596 event_tx.take();
2597 if let Some(s) = signal_tx_opt.take() {
2598 let _ = s.send(());
2599 }
2600 }
2601 _ => {
2602 trace!("BinanceMargin ignoring unhandled WebsocketEvent variant");
2603 }
2604 }
2605 }))
2606}
2607
2608async fn recover_margin_fills(
2630 rest: &Arc<RestApi>,
2631 rate_limiter: &Arc<RateLimitTracker>,
2632 unrecovered: &mut UnrecoveredFills,
2633 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
2634 dedup: &SharedDedupCache,
2635 known: &SharedKnownLiveOrders,
2636 is_isolated: bool,
2637) {
2638 use futures::StreamExt as _;
2639
2640 let due = unrecovered.due(tokio::time::Instant::now());
2641 let Some(oldest) = due.iter().map(|(_, gap)| gap.start_ms).min() else {
2642 return;
2643 };
2644 info!(
2645 gaps = due.len(),
2646 oldest = %gap_time(oldest),
2647 is_isolated,
2648 "BinanceMargin recovering fills missed while disconnected"
2649 );
2650
2651 let mut settled = vec![false; due.len()];
2654 let started = AtomicUsize::new(0);
2655 let recovery = async {
2656 let order_executions_deadline = tokio::time::Instant::now() + ORDER_EXECUTIONS_BUDGET;
2657 let mut recovered = 0u32;
2658 let mut duplicates = 0u32;
2659
2660 let mut stream =
2661 futures::stream::iter(due.iter().cloned().enumerate().map(|(index, (inst, gap))| {
2662 started.store(index + 1, Ordering::Relaxed);
2663 let rest = rest.clone();
2664 let rl = rate_limiter.clone();
2665 async move {
2666 let span = MyTradesFrom::Span {
2667 start: gap.start_ms,
2668 end: gap.end_ms,
2669 };
2670 let raw = match paginate_margin_my_trades(
2671 &rest,
2672 &rl,
2673 &inst,
2674 span,
2675 is_isolated,
2676 RequestKind::Essential,
2677 )
2678 .await
2679 {
2680 Ok(pages) => pages,
2681 Err(e) => return (index, Err(e)),
2682 };
2683 let totals = recovered_order_totals(
2687 ExchangeId::BinanceMargin,
2688 &inst,
2689 &raw,
2690 order_executions_deadline,
2691 |order_id| {
2692 paginate_margin_my_trades(
2693 &rest,
2694 &rl,
2695 &inst,
2696 MyTradesFrom::Order(order_id),
2697 is_isolated,
2698 RequestKind::Essential,
2699 )
2700 },
2701 )
2702 .await;
2703 let cids = recovered_order_cids(
2706 ExchangeId::BinanceMargin,
2707 &inst,
2708 &raw,
2709 order_executions_deadline,
2710 |order_id| fetch_margin_order_cid(&rest, &rl, &inst, order_id, is_isolated),
2711 )
2712 .await;
2713 let trades = raw
2714 .iter()
2715 .filter_map(|t| {
2716 let mut trade = convert_margin_trade(t, &inst)?;
2717 trade.order_filled_quantity =
2718 t.id.and_then(|id| totals.get(&id).copied());
2719 trade.client_order_id =
2720 t.order_id.and_then(|id| cids.get(&id).cloned());
2721 Some(trade)
2722 })
2723 .collect::<Vec<_>>();
2724 (index, Ok(trades))
2725 }
2726 }))
2727 .buffer_unordered(8);
2728 while let Some((index, result)) = stream.next().await {
2729 let (inst, gap) = &due[index];
2730 settled[index] = true;
2731 let trades = match result {
2732 Ok(trades) => trades,
2733 Err(e) => {
2734 gap_failed(
2735 unrecovered,
2736 tx,
2737 ExchangeId::BinanceMargin,
2738 inst,
2739 gap,
2740 FillRecoveryFailure::Request(e.to_string()),
2741 );
2742 continue;
2743 }
2744 };
2745 for trade in trades {
2746 let event = UnindexedAccountEvent::new(
2747 ExchangeId::BinanceMargin,
2748 AccountEventKind::Trade(trade),
2749 );
2750 if let Some(key) = dedup_key_from_event(&event)
2751 && is_duplicate(dedup, key)
2752 {
2753 duplicates += 1;
2754 continue;
2755 }
2756 let sent = {
2759 let mut known = known.lock();
2760 known.observe(&event);
2761 tx.send(event)
2762 };
2763 if sent.is_err() {
2764 debug!("BinanceMargin fill recovery: consumer dropped during recovery");
2765 return;
2766 }
2767 recovered += 1;
2768 }
2769 unrecovered.recovered(inst, gap);
2770 }
2771 info!(
2772 recovered,
2773 duplicates, is_isolated, "BinanceMargin fill recovery complete"
2774 );
2775 };
2776 if tokio::time::timeout(Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS), recovery)
2780 .await
2781 .is_err()
2782 {
2783 let started = started.load(Ordering::Relaxed);
2784 for ((inst, gap), _) in due[..started]
2785 .iter()
2786 .zip(&settled)
2787 .filter(|(_, settled)| !**settled)
2788 {
2789 gap_failed(
2790 unrecovered,
2791 tx,
2792 ExchangeId::BinanceMargin,
2793 inst,
2794 gap,
2795 FillRecoveryFailure::TimedOut {
2796 timeout_secs: FILL_RECOVERY_TIMEOUT_SECS,
2797 },
2798 );
2799 }
2800 }
2801}
2802
2803fn open_reconnect_recovery(
2808 unrecovered: &mut UnrecoveredFills,
2809 unchecked: &mut UncheckedOrders,
2810 known: &SharedKnownLiveOrders,
2811 instruments: &[InstrumentNameExchange],
2812 disconnected: DateTime<Utc>,
2813) {
2814 unrecovered.open(instruments, disconnected, Utc::now());
2815 let held = known.lock().instruments_among(instruments);
2816 unchecked.open(held);
2817}
2818
2819#[allow(
2827 clippy::cognitive_complexity,
2828 reason = "inherent reconnect-loop complexity (token + connect + subscribe + callback + recovery \
2829 + monitor + cleanup + backoff); mirrors spot's `connection_manager`, not worth splitting further"
2830)]
2831async fn margin_connection_manager(
2832 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
2833 dedup: SharedDedupCache,
2834 ws_config: ConfigurationWebsocketApi,
2835 rest_config: Arc<ConfigurationRestApi>,
2836 rest: Arc<RestApi>,
2837 rate_limiter: Arc<RateLimitTracker>,
2838 known: SharedKnownLiveOrders,
2839 instruments: Vec<InstrumentNameExchange>,
2840 initial: Option<(DisconnectOnDrop<Arc<WsApiBase>>, UserListenToken)>,
2841) {
2842 enum DisconnectReason {
2843 Signal,
2844 HeartbeatTimeout,
2845 TokenRefresh,
2846 ConsumerDropped,
2847 }
2848
2849 let mut backoff = ExponentialBackoff::new();
2850 let mut disconnect_time: Option<DateTime<Utc>> = None;
2851 let mut unrecovered = UnrecoveredFills::default();
2853 let mut unchecked = UncheckedOrders::default();
2855 let (mut current_ws, mut current_token) = match initial {
2856 Some((ws, token)) => (Some(ws), Some(token)),
2857 None => (None, None),
2858 };
2859
2860 loop {
2861 let token = match current_token.take() {
2863 Some(t) => t,
2864 None => {
2865 if let Err(reason) = rate_limiter.before_stream_connect().await {
2868 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
2869 break;
2870 }
2871 match acquire_user_listen_token(&rest_config, &rate_limiter, None).await {
2873 Ok(t) => t,
2874 Err(e) => {
2875 error!(%e, "BinanceMargin userListenToken acquisition failed");
2876 if let Err(reason) =
2877 backoff.retry_or_end(&rate_limiter, || e.to_string()).await
2878 {
2879 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
2880 break;
2881 }
2882 continue;
2883 }
2884 }
2885 }
2886 };
2887
2888 let ws = match current_ws.take() {
2890 Some(ws) => ws,
2891 None => match connect_margin_ws(&ws_config).await {
2892 Ok(ws) => ws,
2893 Err(e) => {
2894 error!(%e, "BinanceMargin WS connect failed");
2895 rate_limiter.on_stream_refused(&e);
2896 if let Err(reason) = backoff.retry_or_end(&rate_limiter, || e.to_string()).await
2897 {
2898 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
2899 break;
2900 }
2901 continue;
2902 }
2903 },
2904 };
2905
2906 let (signal_tx, signal_rx) = oneshot::channel::<()>();
2908 let heartbeat_flag = Arc::new(AtomicBool::new(true));
2910 let subscription = register_user_data_listener(
2913 &ws,
2914 tx.clone(),
2915 dedup.clone(),
2916 known.clone(),
2917 heartbeat_flag.clone(),
2918 signal_tx,
2919 cross_account_position_handler,
2920 cross_notice_instrument,
2921 );
2922
2923 if let Err(e) = subscribe_listen_token(&ws, &token.token).await {
2925 warn!(%e, "BinanceMargin user-data subscribe failed, cleaning up and retrying");
2926 subscription.unsubscribe();
2927 if let Err(de) = ws.disconnect().await {
2928 warn!(%de, "BinanceMargin failed to disconnect after subscribe failure");
2929 }
2930 rate_limiter.on_stream_refused(&e);
2931 if let Err(reason) = backoff.retry_or_end(&rate_limiter, || e.to_string()).await {
2933 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
2934 break;
2935 }
2936 continue;
2937 }
2938 info!("BinanceMargin account_stream connected and subscribed");
2939 backoff.reset();
2943
2944 let token_deadline =
2947 tokio::time::Instant::now() + token_renew_after(token.expiration_time_ms);
2948
2949 if let Some(dt) = disconnect_time.take() {
2954 open_reconnect_recovery(&mut unrecovered, &mut unchecked, &known, &instruments, dt);
2955 }
2956 recover_margin_fills(
2959 &rest,
2960 &rate_limiter,
2961 &mut unrecovered,
2962 &tx,
2963 &dedup,
2964 &known,
2965 false,
2966 )
2967 .await;
2968 recover_margin_ended_orders(
2969 &rest,
2970 &rate_limiter,
2971 &known,
2972 &mut unchecked,
2973 &unrecovered,
2974 &tx,
2975 false,
2976 )
2977 .await;
2978
2979 let reason = {
2981 let mut signal_rx = signal_rx;
2982 let retry_gaps = async {
2988 loop {
2989 let next_check = unchecked.next_due(&unrecovered);
2990 match unrecovered.next_due().into_iter().chain(next_check).min() {
2991 Some(due) => tokio::time::sleep_until(due).await,
2992 None => std::future::pending::<()>().await,
2993 }
2994 recover_margin_fills(
2995 &rest,
2996 &rate_limiter,
2997 &mut unrecovered,
2998 &tx,
2999 &dedup,
3000 &known,
3001 false,
3002 )
3003 .await;
3004 recover_margin_ended_orders(
3005 &rest,
3006 &rate_limiter,
3007 &known,
3008 &mut unchecked,
3009 &unrecovered,
3010 &tx,
3011 false,
3012 )
3013 .await;
3014 }
3015 };
3016 tokio::pin!(retry_gaps);
3017 loop {
3018 tokio::select! {
3019 biased;
3022 _ = tx.closed() => {
3023 debug!("BinanceMargin account_stream consumer dropped, terminating");
3024 break DisconnectReason::ConsumerDropped;
3025 }
3026 _ = &mut signal_rx => {
3027 warn!("BinanceMargin WS disconnected, will attempt reconnect");
3028 break DisconnectReason::Signal;
3029 }
3030 () = tokio::time::sleep_until(token_deadline) => {
3031 info!("BinanceMargin userListenToken nearing expiry, renewing");
3032 break DisconnectReason::TokenRefresh;
3033 }
3034 () = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
3035 if heartbeat_flag.swap(false, Ordering::AcqRel) {
3036 continue;
3040 }
3041 warn!("BinanceMargin heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
3042 break DisconnectReason::HeartbeatTimeout;
3043 }
3044 () = &mut retry_gaps => {}
3046 }
3047 }
3048 };
3049 let should_reconnect = !matches!(reason, DisconnectReason::ConsumerDropped);
3050
3051 if should_reconnect {
3052 disconnect_time = Some(match reason {
3053 DisconnectReason::HeartbeatTimeout => {
3054 Utc::now()
3055 - chrono::Duration::seconds(HEARTBEAT_TIMEOUT_SECS as i64)
3056 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3057 }
3058 DisconnectReason::TokenRefresh => {
3059 Utc::now()
3064 - chrono::Duration::seconds(CONNECT_TIMEOUT_SECS as i64)
3065 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3066 }
3067 _ => Utc::now() - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS),
3068 });
3069 }
3070
3071 subscription.unsubscribe();
3073 if let Err(e) = ws.disconnect().await {
3074 warn!(%e, "BinanceMargin failed to disconnect WebSocket");
3075 }
3076
3077 if !should_reconnect || tx.is_closed() {
3078 debug!("BinanceMargin connection manager exiting");
3080 break;
3081 }
3082
3083 if !matches!(reason, DisconnectReason::TokenRefresh) {
3086 let heartbeat_timed_out = matches!(reason, DisconnectReason::HeartbeatTimeout);
3089 let last_error = || disconnect_error(heartbeat_timed_out);
3090 if let Err(reason) = backoff.retry_or_end(&rate_limiter, last_error).await {
3091 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3092 break;
3093 }
3094 }
3095 }
3096}
3097
3098fn build_base_quote_map(
3110 assets: &[QueryIsolatedMarginAccountInfoResponseAssetsInner],
3111) -> HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)> {
3112 let mut map = HashMap::with_capacity(assets.len());
3113 for entry in assets {
3114 let (Some(symbol), Some(base), Some(quote)) = (
3115 entry.symbol.as_deref(),
3116 entry.base_asset.as_deref().and_then(|b| b.asset.as_deref()),
3117 entry
3118 .quote_asset
3119 .as_deref()
3120 .and_then(|q| q.asset.as_deref()),
3121 ) else {
3122 warn!(
3123 "BinanceMargin isolated: account-info entry missing symbol/base/quote asset — \
3124 skipping base/quote map entry"
3125 );
3126 continue;
3127 };
3128 map.insert(
3129 InstrumentNameExchange::new(symbol),
3130 (AssetNameExchange::new(base), AssetNameExchange::new(quote)),
3131 );
3132 }
3133 map
3134}
3135
3136async fn acquire_all_isolated_tokens(
3149 rest_config: &Arc<ConfigurationRestApi>,
3150 rate_limiter: &Arc<RateLimitTracker>,
3151 symbols: &[InstrumentNameExchange],
3152) -> Result<Vec<(InstrumentNameExchange, UserListenToken)>, UnindexedClientError> {
3153 use futures::{StreamExt as _, TryStreamExt as _};
3154
3155 futures::stream::iter(symbols.iter().cloned().map(|sym| {
3156 let rest_config = rest_config.clone();
3157 let rate_limiter = rate_limiter.clone();
3158 async move {
3159 let token = acquire_user_listen_token(&rest_config, &rate_limiter, Some(&sym)).await?;
3160 Ok::<_, UnindexedClientError>((sym, token))
3161 }
3162 }))
3163 .buffer_unordered(8)
3164 .try_collect()
3165 .await
3166}
3167
3168fn earliest_token_expiry_ms(tokens: &[(InstrumentNameExchange, UserListenToken)]) -> i64 {
3173 tokens
3174 .iter()
3175 .map(|(_, t)| t.expiration_time_ms)
3176 .min()
3177 .unwrap_or(i64::MAX)
3178}
3179
3180struct IsolatedLiveConn {
3183 ws: DisconnectOnDrop<Arc<WsApiBase>>,
3184 subscription: UnsubscribeOnDrop,
3185 signal_rx: oneshot::Receiver<()>,
3186 heartbeat_flag: Arc<AtomicBool>,
3187 earliest_expiry_ms: i64,
3190}
3191
3192async fn isolated_connect_and_subscribe(
3202 ws_config: &ConfigurationWebsocketApi,
3203 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3204 dedup: &SharedDedupCache,
3205 known: &SharedKnownLiveOrders,
3206 base_quote: &Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
3207 tokens: &[(InstrumentNameExchange, UserListenToken)],
3208) -> anyhow::Result<IsolatedLiveConn> {
3209 let ws = connect_margin_ws(ws_config).await?;
3210 let sub_map = Arc::new(Mutex::new(
3216 HashMap::<i64, InstrumentNameExchange>::with_capacity(tokens.len()),
3217 ));
3218 let (signal_tx, signal_rx) = oneshot::channel::<()>();
3219 let heartbeat_flag = Arc::new(AtomicBool::new(true));
3221
3222 let handle_position = {
3223 let sub_map = sub_map.clone();
3224 let base_quote = base_quote.clone();
3225 move |position, subscription_id, buf: &mut Vec<UnindexedAccountEvent>| {
3226 route_isolated_account_position(position, subscription_id, &sub_map, &base_quote, buf);
3227 }
3228 };
3229 let notice_instrument = {
3230 let sub_map = sub_map.clone();
3231 move |subscription_id| isolated_notice_instrument(subscription_id, &sub_map)
3232 };
3233 let subscription = register_user_data_listener(
3235 &ws,
3236 tx.clone(),
3237 dedup.clone(),
3238 known.clone(),
3239 heartbeat_flag.clone(),
3240 signal_tx,
3241 handle_position,
3242 notice_instrument,
3243 );
3244
3245 let earliest_expiry_ms = earliest_token_expiry_ms(tokens);
3246 for (sym, token) in tokens {
3247 match subscribe_listen_token_capture(&ws, &token.token).await {
3248 Ok(Some(id)) => {
3249 sub_map
3250 .lock()
3251 .unwrap_or_else(|p| p.into_inner())
3252 .insert(id, sym.clone());
3253 }
3254 Ok(None) => warn!(
3255 symbol = %sym.name(),
3256 "BinanceMargin isolated: subscribe ack carried no subscriptionId — live per-pair \
3257 balances unavailable for this symbol (fills/orders unaffected; balances via snapshot)"
3258 ),
3259 Err(e) => {
3260 warn!(%e, symbol = %sym.name(), "BinanceMargin isolated subscribe failed");
3263 subscription.unsubscribe();
3264 if let Err(de) = ws.disconnect().await {
3265 warn!(%de, "BinanceMargin isolated: disconnect after subscribe failure failed");
3266 }
3267 return Err(e);
3268 }
3269 }
3270 }
3271
3272 Ok(IsolatedLiveConn {
3273 ws,
3274 subscription,
3275 signal_rx,
3276 heartbeat_flag,
3277 earliest_expiry_ms,
3278 })
3279}
3280
3281#[allow(
3290 clippy::cognitive_complexity,
3291 reason = "inherent reconnect-loop complexity (tokens + connect + subscribe + monitor + cleanup \
3292 + backoff); mirrors the cross manager, not worth splitting further"
3293)]
3294async fn isolated_connection_manager(
3295 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
3296 dedup: SharedDedupCache,
3297 ws_config: ConfigurationWebsocketApi,
3298 rest_config: Arc<ConfigurationRestApi>,
3299 rest: Arc<RestApi>,
3300 rate_limiter: Arc<RateLimitTracker>,
3301 known: SharedKnownLiveOrders,
3302 symbols: Vec<InstrumentNameExchange>,
3303 base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
3304 initial: Option<IsolatedLiveConn>,
3305) {
3306 enum DisconnectReason {
3307 Signal,
3308 HeartbeatTimeout,
3309 TokenRefresh,
3310 ConsumerDropped,
3311 }
3312
3313 let mut backoff = ExponentialBackoff::new();
3314 let mut disconnect_time: Option<DateTime<Utc>> = None;
3315 let mut unrecovered = UnrecoveredFills::default();
3317 let mut unchecked = UncheckedOrders::default();
3319 let mut current = initial;
3320
3321 loop {
3322 let IsolatedLiveConn {
3324 ws,
3325 subscription,
3326 signal_rx,
3327 heartbeat_flag,
3328 earliest_expiry_ms,
3329 } = match current.take() {
3330 Some(live) => live,
3332 None => {
3333 if let Err(reason) = rate_limiter.before_stream_connect().await {
3335 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3336 break;
3337 }
3338 let tokens = match acquire_all_isolated_tokens(
3341 &rest_config,
3342 &rate_limiter,
3343 &symbols,
3344 )
3345 .await
3346 {
3347 Ok(t) => t,
3348 Err(e) => {
3349 error!(%e, "BinanceMargin isolated userListenToken acquisition failed");
3350 if let Err(reason) =
3351 backoff.retry_or_end(&rate_limiter, || e.to_string()).await
3352 {
3353 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3354 break;
3355 }
3356 continue;
3357 }
3358 };
3359 match isolated_connect_and_subscribe(
3360 &ws_config,
3361 &tx,
3362 &dedup,
3363 &known,
3364 &base_quote,
3365 &tokens,
3366 )
3367 .await
3368 {
3369 Ok(live) => live,
3370 Err(e) => {
3371 error!(%e, "BinanceMargin isolated connect/subscribe failed");
3372 rate_limiter.on_stream_refused(&e);
3373 if let Err(reason) =
3374 backoff.retry_or_end(&rate_limiter, || e.to_string()).await
3375 {
3376 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3377 break;
3378 }
3379 continue;
3380 }
3381 }
3382 }
3383 };
3384
3385 info!(
3386 symbols = symbols.len(),
3387 "BinanceMargin isolated account_stream connected and subscribed"
3388 );
3389 backoff.reset();
3392
3393 let token_deadline = tokio::time::Instant::now() + token_renew_after(earliest_expiry_ms);
3396
3397 if let Some(dt) = disconnect_time.take() {
3400 open_reconnect_recovery(&mut unrecovered, &mut unchecked, &known, &symbols, dt);
3401 }
3402 recover_margin_fills(
3404 &rest,
3405 &rate_limiter,
3406 &mut unrecovered,
3407 &tx,
3408 &dedup,
3409 &known,
3410 true,
3411 )
3412 .await;
3413 recover_margin_ended_orders(
3414 &rest,
3415 &rate_limiter,
3416 &known,
3417 &mut unchecked,
3418 &unrecovered,
3419 &tx,
3420 true,
3421 )
3422 .await;
3423
3424 let reason = {
3427 let mut signal_rx = signal_rx;
3428 let retry_gaps = async {
3431 loop {
3432 let next_check = unchecked.next_due(&unrecovered);
3433 match unrecovered.next_due().into_iter().chain(next_check).min() {
3434 Some(due) => tokio::time::sleep_until(due).await,
3435 None => std::future::pending::<()>().await,
3436 }
3437 recover_margin_fills(
3438 &rest,
3439 &rate_limiter,
3440 &mut unrecovered,
3441 &tx,
3442 &dedup,
3443 &known,
3444 true,
3445 )
3446 .await;
3447 recover_margin_ended_orders(
3448 &rest,
3449 &rate_limiter,
3450 &known,
3451 &mut unchecked,
3452 &unrecovered,
3453 &tx,
3454 true,
3455 )
3456 .await;
3457 }
3458 };
3459 tokio::pin!(retry_gaps);
3460 loop {
3461 tokio::select! {
3462 biased;
3463 _ = tx.closed() => {
3464 debug!("BinanceMargin isolated consumer dropped, terminating");
3465 break DisconnectReason::ConsumerDropped;
3466 }
3467 _ = &mut signal_rx => {
3468 warn!("BinanceMargin isolated WS disconnected, will attempt reconnect");
3469 break DisconnectReason::Signal;
3470 }
3471 () = tokio::time::sleep_until(token_deadline) => {
3472 info!("BinanceMargin isolated userListenToken nearing expiry, renewing all");
3473 break DisconnectReason::TokenRefresh;
3474 }
3475 () = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
3476 if heartbeat_flag.swap(false, Ordering::AcqRel) {
3477 continue;
3478 }
3479 warn!("BinanceMargin isolated heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
3480 break DisconnectReason::HeartbeatTimeout;
3481 }
3482 () = &mut retry_gaps => {}
3484 }
3485 }
3486 };
3487 let should_reconnect = !matches!(reason, DisconnectReason::ConsumerDropped);
3488
3489 if should_reconnect {
3490 disconnect_time = Some(match reason {
3491 DisconnectReason::HeartbeatTimeout => {
3492 Utc::now()
3493 - chrono::Duration::seconds(HEARTBEAT_TIMEOUT_SECS as i64)
3494 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3495 }
3496 DisconnectReason::TokenRefresh => {
3497 Utc::now()
3501 - chrono::Duration::seconds(CONNECT_TIMEOUT_SECS as i64)
3502 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3503 }
3504 _ => Utc::now() - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS),
3505 });
3506 }
3507
3508 subscription.unsubscribe();
3510 if let Err(e) = ws.disconnect().await {
3511 warn!(%e, "BinanceMargin isolated failed to disconnect WebSocket");
3512 }
3513
3514 if !should_reconnect || tx.is_closed() {
3515 debug!("BinanceMargin isolated connection manager exiting");
3517 break;
3518 }
3519
3520 if !matches!(reason, DisconnectReason::TokenRefresh) {
3523 let heartbeat_timed_out = matches!(reason, DisconnectReason::HeartbeatTimeout);
3526 let last_error = || disconnect_error(heartbeat_timed_out);
3527 if let Err(reason) = backoff.retry_or_end(&rate_limiter, last_error).await {
3528 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3529 break;
3530 }
3531 }
3532 }
3533}
3534
3535async fn fetch_margin_open_orders_for_instrument(
3542 rest: Arc<RestApi>,
3543 rate_limiter: Arc<RateLimitTracker>,
3544 instrument: InstrumentNameExchange,
3545 is_isolated: bool,
3546 kind: RequestKind,
3547) -> Result<(InstrumentNameExchange, OpenOrderListing), UnindexedClientError> {
3548 let symbol_str = instrument.name().to_string();
3550 let isolated = QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(is_isolated);
3551 let response = rest_call_with_retry(&rest, &rate_limiter, kind, |rest| {
3552 let sym = symbol_str.clone();
3553 let isolated = isolated.clone();
3554 Box::pin(async move {
3555 let params = QueryMarginAccountsOpenOrdersParams::builder()
3556 .symbol(sym)
3557 .is_isolated(isolated)
3558 .build()?;
3559 rest.query_margin_accounts_open_orders(params).await
3560 })
3561 })
3562 .await
3563 .map_err(|e| classify_rest_query_error(&e, Some(&instrument)))?;
3564
3565 let orders_data = response.data().await.map_err(response_decode_error)?;
3566
3567 let listing = convert_open_order_listing(&orders_data, ExchangeId::BinanceMargin, &instrument);
3568
3569 Ok((instrument, listing))
3570}
3571
3572async fn fetch_margin_all_open_orders(
3579 rest: Arc<RestApi>,
3580 rate_limiter: Arc<RateLimitTracker>,
3581 is_isolated: bool,
3582) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
3583 let isolated = QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(is_isolated);
3584 let response = rest_call_with_retry(&rest, &rate_limiter, RequestKind::Query, |rest| {
3585 let isolated = isolated.clone();
3586 Box::pin(async move {
3587 let params = QueryMarginAccountsOpenOrdersParams::builder()
3588 .is_isolated(isolated)
3589 .build()?;
3590 rest.query_margin_accounts_open_orders(params).await
3591 })
3592 })
3593 .await
3594 .map_err(|e| classify_rest_query_error(&e, None))?;
3595
3596 let orders_data = response.data().await.map_err(response_decode_error)?;
3597
3598 let orders = orders_data
3599 .into_iter()
3600 .filter_map(|o| convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin))
3601 .collect();
3602
3603 Ok(orders)
3604}
3605
3606async fn fetch_margin_order_lookup(
3613 rest: Arc<RestApi>,
3614 rate_limiter: Arc<RateLimitTracker>,
3615 key: UnindexedOrderKey,
3616 is_isolated: bool,
3617 kind: RequestKind,
3618 recorded_kind: Option<OrderKind>,
3619) -> Result<OrderLookup, UnindexedClientError> {
3620 let symbol = key.instrument.name().to_string();
3622 let cid = key.cid.0.to_string();
3623 let isolated = QueryMarginAccountsOrderIsIsolatedEnum::from_flag(is_isolated);
3624 let response = match rest_call_with_retry(&rest, &rate_limiter, kind, |rest| {
3625 let (symbol, cid, isolated) = (symbol.clone(), cid.clone(), isolated.clone());
3626 Box::pin(async move {
3627 let params = QueryMarginAccountsOrderParams::builder(symbol)
3628 .orig_client_order_id(cid)
3629 .is_isolated(isolated)
3630 .build()?;
3631 rest.query_margin_accounts_order(params).await
3632 })
3633 })
3634 .await
3635 {
3636 Ok(response) => response,
3637 Err(e) if is_unknown_order(&e) => {
3638 debug!(instrument = %key.instrument, cid = %key.cid, error = %e, "BinanceMargin does not know this order");
3639 return Ok(OrderLookup::Unknown);
3640 }
3641 Err(e) => return Err(classify_rest_query_error(&e, Some(&key.instrument))),
3642 };
3643
3644 let row = response.data().await.map_err(response_decode_error)?;
3645
3646 Ok(
3647 convert_ended_order(&row, ExchangeId::BinanceMargin, &key, recorded_kind)
3648 .map_or(OrderLookup::NotEnded, |order| {
3649 OrderLookup::Ended(Box::new(order))
3650 }),
3651 )
3652}
3653
3654async fn fetch_margin_order_cid(
3659 rest: &Arc<RestApi>,
3660 rate_limiter: &RateLimitTracker,
3661 instrument: &InstrumentNameExchange,
3662 order_id: i64,
3663 is_isolated: bool,
3664) -> Result<Option<ClientOrderId>, UnindexedClientError> {
3665 let symbol = instrument.name().to_string();
3666 let isolated = QueryMarginAccountsOrderIsIsolatedEnum::from_flag(is_isolated);
3667 let response = match rest_call_with_retry(rest, rate_limiter, RequestKind::Essential, |rest| {
3668 let (symbol, isolated) = (symbol.clone(), isolated.clone());
3669 Box::pin(async move {
3670 let params = QueryMarginAccountsOrderParams::builder(symbol)
3671 .order_id(order_id)
3672 .is_isolated(isolated)
3673 .build()?;
3674 rest.query_margin_accounts_order(params).await
3675 })
3676 })
3677 .await
3678 {
3679 Ok(response) => response,
3680 Err(e) if is_unknown_order(&e) => {
3681 debug!(%instrument, order_id, error = %e, "BinanceMargin does not know this order");
3682 return Ok(None);
3683 }
3684 Err(e) => return Err(classify_rest_query_error(&e, Some(instrument))),
3685 };
3686 let row = response.data().await.map_err(response_decode_error)?;
3687 Ok(row
3688 .client_order_id
3689 .filter(|cid| !cid.is_empty())
3690 .map(ClientOrderId::new))
3691}
3692
3693async fn listed_margin_open_cids(
3696 rest: Arc<RestApi>,
3697 rate_limiter: Arc<RateLimitTracker>,
3698 instruments: Vec<InstrumentNameExchange>,
3699 is_isolated: bool,
3700) -> Result<FnvHashSet<ClientOrderId>, UnindexedClientError> {
3701 let mut listed = FnvHashSet::default();
3702 for instrument in instruments {
3703 let (_, listing) = fetch_margin_open_orders_for_instrument(
3704 rest.clone(),
3705 rate_limiter.clone(),
3706 instrument,
3707 is_isolated,
3708 RequestKind::Essential,
3709 )
3710 .await?;
3711 listed.extend(listing.orders.into_iter().map(|order| order.key.cid));
3712 }
3713 Ok(listed)
3714}
3715
3716async fn recover_margin_ended_orders(
3719 rest: &Arc<RestApi>,
3720 rate_limiter: &Arc<RateLimitTracker>,
3721 known: &SharedKnownLiveOrders,
3722 unchecked: &mut UncheckedOrders,
3723 unrecovered: &UnrecoveredFills,
3724 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3725 is_isolated: bool,
3726) {
3727 recover_ended_orders(
3728 ExchangeId::BinanceMargin,
3729 known,
3730 unchecked,
3731 unrecovered,
3732 tx,
3733 OpenListing::PerInstrument,
3734 |instruments| {
3735 listed_margin_open_cids(rest.clone(), rate_limiter.clone(), instruments, is_isolated)
3736 },
3737 |key| {
3738 let recorded_kind = known.lock().kind_of(&key.cid);
3740 fetch_margin_order_lookup(
3741 rest.clone(),
3742 rate_limiter.clone(),
3743 key,
3744 is_isolated,
3745 RequestKind::Essential,
3746 recorded_kind,
3747 )
3748 },
3749 )
3750 .await;
3751}
3752
3753const MARGIN_MY_TRADES_WINDOW_MS: i64 = 23 * 60 * 60 * 1000;
3759
3760#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3763enum MarginTradesQuery {
3764 Window { start: i64, end: i64 },
3766 FromId(i64),
3768 Order(i64),
3770 OrderFromId { order_id: i64, from_id: i64 },
3772}
3773
3774impl MarginTradesQuery {
3775 fn first(from: MyTradesFrom, now_ms: i64) -> Option<Self> {
3779 match from {
3780 MyTradesFrom::Span { start, .. } => {
3781 (start <= now_ms).then(|| Self::window(start, now_ms))
3782 }
3783 MyTradesFrom::Order(order_id) => Some(Self::Order(order_id)),
3784 }
3785 }
3786
3787 fn window(start: i64, now_ms: i64) -> Self {
3789 Self::Window {
3790 start,
3791 end: start.saturating_add(MARGIN_MY_TRADES_WINDOW_MS).min(now_ms),
3792 }
3793 }
3794
3795 fn next(self, page_len: usize, last_id: Option<i64>, now_ms: i64) -> Option<Self> {
3798 let full = page_len >= BINANCE_MAX_TRADES;
3799 let after_last = || last_id?.checked_add(1);
3800 match self {
3801 Self::Window { end, .. } if page_len == 0 => {
3806 (end < now_ms).then(|| Self::window(end, now_ms))
3807 }
3808 Self::Window { end, .. } if end >= now_ms && !full => None,
3810 Self::Window { .. } => after_last().map(Self::FromId),
3813 Self::FromId(_) if full => after_last().map(Self::FromId),
3814 Self::Order(order_id) | Self::OrderFromId { order_id, .. } if full => {
3815 after_last().map(|from_id| Self::OrderFromId { order_id, from_id })
3816 }
3817 Self::FromId(_) | Self::Order(_) | Self::OrderFromId { .. } => None,
3818 }
3819 }
3820}
3821
3822async fn paginate_margin_my_trades(
3844 rest: &Arc<RestApi>,
3845 rate_limiter: &Arc<RateLimitTracker>,
3846 instrument: &InstrumentNameExchange,
3847 from: MyTradesFrom,
3848 is_isolated: bool,
3849 kind: RequestKind,
3850) -> Result<Vec<QueryMarginAccountsTradeListResponseInner>, UnindexedClientError> {
3851 let symbol_str = instrument.name().to_string();
3853 let isolated = QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(is_isolated);
3854 #[allow(clippy::cast_possible_wrap)]
3859 let limit = BINANCE_MAX_TRADES as i64;
3860 let now_ms = match from {
3863 MyTradesFrom::Span { end, .. } => end.min(Utc::now().timestamp_millis()),
3864 MyTradesFrom::Order(_) => Utc::now().timestamp_millis(),
3865 };
3866 let mut all_pages = Vec::new();
3867 let mut requests = 0u32;
3868 let mut next = MarginTradesQuery::first(from, now_ms);
3869 while let Some(query) = next {
3870 requests += 1;
3871 let response = rest_call_with_retry(rest, rate_limiter, kind, |rest| {
3872 let sym = symbol_str.clone();
3873 let isolated = isolated.clone();
3874 Box::pin(async move {
3875 let builder = QueryMarginAccountsTradeListParams::builder(sym)
3876 .is_isolated(isolated)
3877 .limit(limit);
3878 let params = match query {
3879 MarginTradesQuery::Window { start, end } => {
3880 builder.start_time(start).end_time(end)
3881 }
3882 MarginTradesQuery::FromId(id) => builder.from_id(id),
3883 MarginTradesQuery::Order(order_id) => builder.order_id(order_id),
3884 MarginTradesQuery::OrderFromId { order_id, from_id } => {
3885 builder.order_id(order_id).from_id(from_id)
3886 }
3887 }
3888 .build()?;
3889 rest.query_margin_accounts_trade_list(params).await
3890 })
3891 })
3892 .await
3893 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
3894
3895 let mut page = response.data().await.map_err(response_decode_error)?;
3896
3897 let page_len = page.len();
3898 let last_id = page.last().and_then(|t| t.id);
3899 let past_end = match from {
3901 MyTradesFrom::Span { end, .. } => drop_after(&mut page, end),
3902 MyTradesFrom::Order(_) => false,
3903 };
3904 all_pages.extend(page);
3905 if past_end {
3906 break;
3907 }
3908
3909 if page_len > 0 && last_id.is_none() {
3910 warn!(%instrument, "BinanceMargin paginate_my_trades: trade missing ID, cannot page past it");
3911 break;
3912 }
3913 next = query.next(page_len, last_id, now_ms);
3914 }
3915 debug!(
3916 %instrument,
3917 requests,
3918 trades = all_pages.len(),
3919 "BinanceMargin paginate_my_trades: walk complete"
3920 );
3921 Ok(all_pages)
3922}
3923
3924fn convert_margin_balance_entry(
3934 b: QueryCrossMarginAccountDetailsResponseUserAssetsInner,
3935 now: DateTime<Utc>,
3936) -> Option<AssetBalance<AssetNameExchange>> {
3937 let asset_name = AssetNameExchange::new(b.asset.as_deref()?);
3938 let free = match b.free.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
3939 Some(v) => v,
3940 None => {
3941 warn!(%asset_name, "BinanceMargin balance missing/unparseable 'free' field");
3942 return None;
3943 }
3944 };
3945 let locked = match b.locked.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
3946 Some(v) => v,
3947 None => {
3948 warn!(%asset_name, "BinanceMargin balance missing/unparseable 'locked' field");
3949 return None;
3950 }
3951 };
3952 let borrowed = b
3957 .borrowed
3958 .as_deref()
3959 .and_then(|s| Decimal::from_str(s).ok())
3960 .unwrap_or(Decimal::ZERO);
3961 let interest = b
3962 .interest
3963 .as_deref()
3964 .and_then(|s| Decimal::from_str(s).ok())
3965 .unwrap_or(Decimal::ZERO);
3966 Some(AssetBalance::new(
3967 asset_name,
3968 Balance::new_margin(free + locked, free, borrowed, interest),
3969 now,
3970 ))
3971}
3972
3973fn filter_and_convert_margin_balances(
3976 user_assets: Vec<QueryCrossMarginAccountDetailsResponseUserAssetsInner>,
3977 assets: &[AssetNameExchange],
3978) -> Vec<AssetBalance<AssetNameExchange>> {
3979 let now = Utc::now();
3980 if assets.is_empty() {
3982 return user_assets
3983 .into_iter()
3984 .filter_map(|b| convert_margin_balance_entry(b, now))
3985 .collect();
3986 }
3987 if assets.len() <= 16 {
3989 return user_assets
3990 .into_iter()
3991 .filter_map(|b| {
3992 let asset_name_str = b.asset.as_deref()?;
3993 if !assets.iter().any(|a| a.name().as_str() == asset_name_str) {
3994 return None;
3995 }
3996 convert_margin_balance_entry(b, now)
3997 })
3998 .collect();
3999 }
4000 use std::collections::HashSet;
4001 let asset_set: HashSet<&str> = assets.iter().map(|a| a.name().as_str()).collect();
4002 user_assets
4003 .into_iter()
4004 .filter_map(|b| {
4005 let asset_name_str = b.asset.as_deref()?;
4006 if !asset_set.contains(asset_name_str) {
4007 return None;
4008 }
4009 convert_margin_balance_entry(b, now)
4010 })
4011 .collect()
4012}
4013
4014fn active_order_snapshot(
4021 o: Order<ExchangeId, InstrumentNameExchange, Open>,
4022) -> Order<ExchangeId, InstrumentNameExchange, OrderState<AssetNameExchange, InstrumentNameExchange>>
4023{
4024 Order {
4025 key: o.key,
4026 side: o.side,
4027 price: o.price,
4028 quantity: o.quantity,
4029 kind: o.kind,
4030 time_in_force: o.time_in_force,
4031 state: OrderState::active(o.state),
4032 }
4033}
4034
4035fn convert_isolated_asset_balance(
4045 asset: Option<&str>,
4046 free: Option<&str>,
4047 locked: Option<&str>,
4048 borrowed: Option<&str>,
4049 interest: Option<&str>,
4050 now: DateTime<Utc>,
4051) -> Option<AssetBalance<AssetNameExchange>> {
4052 let asset_name = AssetNameExchange::new(asset?);
4053 let free = match free.and_then(|s| Decimal::from_str(s).ok()) {
4054 Some(v) => v,
4055 None => {
4056 warn!(%asset_name, "BinanceMargin isolated balance missing/unparseable 'free' field");
4057 return None;
4058 }
4059 };
4060 let locked = match locked.and_then(|s| Decimal::from_str(s).ok()) {
4061 Some(v) => v,
4062 None => {
4063 warn!(%asset_name, "BinanceMargin isolated balance missing/unparseable 'locked' field");
4064 return None;
4065 }
4066 };
4067 let borrowed = borrowed
4068 .and_then(|s| Decimal::from_str(s).ok())
4069 .unwrap_or(Decimal::ZERO);
4070 let interest = interest
4071 .and_then(|s| Decimal::from_str(s).ok())
4072 .unwrap_or(Decimal::ZERO);
4073 Some(AssetBalance::new(
4074 asset_name,
4075 Balance::new_margin(free + locked, free, borrowed, interest),
4076 now,
4077 ))
4078}
4079
4080fn convert_isolated_margin_assets(
4088 assets: Vec<QueryIsolatedMarginAccountInfoResponseAssetsInner>,
4089) -> HashMap<InstrumentNameExchange, IsolatedInstrumentState<AssetNameExchange>> {
4090 let now = Utc::now();
4093 let mut map = HashMap::with_capacity(assets.len());
4094 for entry in assets {
4095 let Some(symbol) = entry.symbol.as_deref() else {
4096 warn!("BinanceMargin isolated asset entry missing 'symbol' — skipping");
4097 continue;
4098 };
4099 let instrument = InstrumentNameExchange::new(symbol);
4100
4101 let Some(base_raw) = entry.base_asset.as_deref() else {
4102 warn!(%instrument, "BinanceMargin isolated entry missing baseAsset — skipping");
4103 continue;
4104 };
4105 let Some(quote_raw) = entry.quote_asset.as_deref() else {
4106 warn!(%instrument, "BinanceMargin isolated entry missing quoteAsset — skipping");
4107 continue;
4108 };
4109
4110 let Some(base) = convert_isolated_asset_balance(
4111 base_raw.asset.as_deref(),
4112 base_raw.free.as_deref(),
4113 base_raw.locked.as_deref(),
4114 base_raw.borrowed.as_deref(),
4115 base_raw.interest.as_deref(),
4116 now,
4117 ) else {
4118 warn!(%instrument, "BinanceMargin isolated baseAsset missing required field — skipping");
4119 continue;
4120 };
4121 let Some(quote) = convert_isolated_asset_balance(
4122 quote_raw.asset.as_deref(),
4123 quote_raw.free.as_deref(),
4124 quote_raw.locked.as_deref(),
4125 quote_raw.borrowed.as_deref(),
4126 quote_raw.interest.as_deref(),
4127 now,
4128 ) else {
4129 warn!(%instrument, "BinanceMargin isolated quoteAsset missing required field — skipping");
4130 continue;
4131 };
4132
4133 let risk = IsolatedMarginRisk {
4134 margin_level: entry
4135 .margin_level
4136 .as_deref()
4137 .and_then(|s| Decimal::from_str(s).ok()),
4138 margin_ratio: entry
4139 .margin_ratio
4140 .as_deref()
4141 .and_then(|s| Decimal::from_str(s).ok()),
4142 liquidation_price: entry
4143 .liquidate_price
4144 .as_deref()
4145 .and_then(|s| Decimal::from_str(s).ok()),
4146 };
4147
4148 map.insert(instrument, IsolatedInstrumentState { base, quote, risk });
4149 }
4150 map
4151}
4152
4153fn chunk_symbols(symbols: &[InstrumentNameExchange]) -> Vec<String> {
4156 symbols
4157 .chunks(5)
4158 .map(|chunk| {
4159 chunk
4160 .iter()
4161 .map(|s| s.name().as_str())
4162 .collect::<Vec<_>>()
4163 .join(",")
4164 })
4165 .collect()
4166}
4167
4168async fn fetch_isolated_margin_account_info(
4174 rest: Arc<RestApi>,
4175 rate_limiter: Arc<RateLimitTracker>,
4176 symbols: Vec<InstrumentNameExchange>,
4177) -> Result<Vec<QueryIsolatedMarginAccountInfoResponseAssetsInner>, UnindexedClientError> {
4178 use futures::{StreamExt as _, TryStreamExt as _};
4179
4180 let chunks = chunk_symbols(&symbols);
4181
4182 futures::stream::iter(chunks.into_iter().map(|symbols_param| {
4186 let rest = rest.clone();
4187 let rate_limiter = rate_limiter.clone();
4188 async move {
4189 let response = rest_call_with_retry(&rest, &rate_limiter, RequestKind::Query, |rest| {
4190 let symbols_param = symbols_param.clone();
4191 Box::pin(async move {
4192 let params = QueryIsolatedMarginAccountInfoParams::builder()
4193 .symbols(symbols_param)
4194 .build()?;
4195 rest.query_isolated_margin_account_info(params).await
4196 })
4197 })
4198 .await
4199 .map_err(|e| classify_rest_query_error(&e, None))?;
4200
4201 let info = response.data().await.map_err(response_decode_error)?;
4202 Ok::<_, UnindexedClientError>(info.assets.unwrap_or_default())
4203 }
4204 }))
4205 .buffer_unordered(8)
4206 .try_fold(
4207 Vec::with_capacity(symbols.len()),
4208 |mut acc, assets| async move {
4209 acc.extend(assets);
4210 Ok(acc)
4211 },
4212 )
4213 .await
4214}
4215
4216fn convert_margin_trade(
4223 t: &QueryMarginAccountsTradeListResponseInner,
4224 instrument: &InstrumentNameExchange,
4225) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
4226 let trade_id_raw = match t.id {
4227 Some(id) => id,
4228 None => {
4229 warn!(%instrument, "BinanceMargin trade missing id");
4230 return None;
4231 }
4232 };
4233 let trade_id = TradeId(format_smolstr!("{trade_id_raw}"));
4234 let order_id = match t.order_id {
4235 Some(id) => OrderId(format_smolstr!("{id}")),
4236 None => {
4237 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing orderId");
4238 return None;
4239 }
4240 };
4241 let side = match t.is_buyer {
4242 Some(true) => Side::Buy,
4243 Some(false) => Side::Sell,
4244 None => {
4245 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing isBuyer");
4246 return None;
4247 }
4248 };
4249 let price = match t.price.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
4250 Some(v) => v,
4251 None => {
4252 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable price");
4253 return None;
4254 }
4255 };
4256 let quantity = match t.qty.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
4257 Some(v) => v,
4258 None => {
4259 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable qty");
4260 return None;
4261 }
4262 };
4263 let commission = t
4264 .commission
4265 .as_deref()
4266 .and_then(|s| Decimal::from_str(s).ok())
4267 .unwrap_or(Decimal::ZERO);
4268 let time_exchange = match t.time.and_then(|ms| Utc.timestamp_millis_opt(ms).single()) {
4269 Some(ts) => ts,
4270 None => {
4271 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable time, using now");
4272 Utc::now()
4273 }
4274 };
4275
4276 let fee_asset = t
4281 .commission_asset
4282 .as_deref()
4283 .map(AssetNameExchange::from)
4284 .unwrap_or_else(|| AssetNameExchange::from("UNKNOWN"));
4285
4286 Some(Trade::new(
4287 trade_id,
4288 order_id,
4289 instrument.clone(),
4290 StrategyId::unknown(), time_exchange,
4292 side,
4293 price,
4294 quantity,
4295 None,
4297 AssetFees::new(fee_asset, commission, None),
4298 ))
4299}
4300
4301#[derive(Debug)]
4307enum BuildOrderError {
4308 Unsupported,
4310 Build(String),
4312 Trailing(TrailingOrderError),
4314}
4315
4316trait IsolatedFlag {
4327 fn from_flag(is_isolated: bool) -> Self;
4329}
4330
4331macro_rules! impl_isolated_flag {
4336 ($($t:ty),* $(,)?) => {
4337 $(
4338 impl IsolatedFlag for $t {
4339 fn from_flag(is_isolated: bool) -> Self {
4340 if is_isolated { Self::True } else { Self::False }
4341 }
4342 }
4343 )*
4344 };
4345}
4346
4347impl_isolated_flag!(
4348 MarginAccountNewOrderIsIsolatedEnum,
4349 MarginAccountCancelOrderIsIsolatedEnum,
4350 QueryMarginAccountsOpenOrdersIsIsolatedEnum,
4351 QueryMarginAccountsOrderIsIsolatedEnum,
4352 QueryMarginAccountsTradeListIsIsolatedEnum,
4353 GetFutureHourlyInterestRateIsIsolatedEnum,
4354);
4355
4356fn build_cancel_order_params(
4364 symbol: String,
4365 id: Option<&OrderId>,
4366 cid: &ClientOrderId,
4367 is_isolated: bool,
4368) -> Result<MarginAccountCancelOrderParams, String> {
4369 let mut builder = MarginAccountCancelOrderParams::builder(symbol).is_isolated(
4370 MarginAccountCancelOrderIsIsolatedEnum::from_flag(is_isolated),
4371 );
4372
4373 match id {
4374 Some(order_id) => match order_id.0.parse::<i64>() {
4375 Ok(id) => builder = builder.order_id(id),
4376 Err(_) => {
4377 error!(
4379 order_id = %order_id.0,
4380 "BinanceMargin cancel: exchange orderId not parseable as i64, falling back to clientOrderId"
4381 );
4382 builder = builder.orig_client_order_id(cid.0.to_string());
4383 }
4384 },
4385 None => builder = builder.orig_client_order_id(cid.0.to_string()),
4386 }
4387
4388 builder.build().map_err(|e| e.to_string())
4389}
4390
4391#[allow(clippy::too_many_arguments)] fn build_new_order_params(
4400 symbol: String,
4401 side: Side,
4402 price: Option<Decimal>,
4403 quantity: Decimal,
4404 kind: OrderKind,
4405 time_in_force: TimeInForce,
4406 new_client_order_id: String,
4407 side_effect: MarginSideEffect,
4408 is_isolated: bool,
4409) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
4410 let binance_side = match side {
4411 Side::Buy => MarginAccountNewOrderSideEnum::Buy,
4412 Side::Sell => MarginAccountNewOrderSideEnum::Sell,
4413 };
4414
4415 let (binance_type, binance_tif) =
4416 convert_order_kind_tif_margin(kind, time_in_force).ok_or(BuildOrderError::Unsupported)?;
4417
4418 let mut builder = MarginAccountNewOrderParams::builder(symbol, binance_side, binance_type)
4420 .quantity(quantity)
4421 .is_isolated(MarginAccountNewOrderIsIsolatedEnum::from_flag(is_isolated))
4422 .side_effect_type(side_effect.as_sdk_enum())
4423 .new_client_order_id(new_client_order_id)
4424 .new_order_resp_type(MarginAccountNewOrderNewOrderRespTypeEnum::Full);
4425
4426 if side_effect == MarginSideEffect::AutoBorrowRepay {
4429 builder = builder.auto_repay_at_cancel(true);
4430 }
4431
4432 if let Some(tif) = binance_tif {
4433 builder = builder.time_in_force(tif);
4434 }
4435
4436 if let Some(trailing) = trailing_params(&kind, price).map_err(BuildOrderError::Trailing)? {
4439 builder = builder.trailing_delta(i64::from(trailing.trailing_delta));
4440 if let Some(limit) = trailing.price {
4441 builder = builder.price(limit);
4442 }
4443 if let Some(activation_price) = trailing.stop_price {
4444 builder = builder.stop_price(activation_price);
4445 }
4446 }
4447
4448 match kind {
4452 OrderKind::Limit => {
4453 builder = builder.price(price);
4454 }
4455 OrderKind::Stop { trigger_price } | OrderKind::TakeProfit { trigger_price } => {
4456 builder = builder.stop_price(trigger_price);
4457 }
4458 OrderKind::StopLimit { trigger_price } | OrderKind::TakeProfitLimit { trigger_price } => {
4459 builder = builder.price(price).stop_price(trigger_price);
4460 }
4461 OrderKind::TrailingStop { .. }
4464 | OrderKind::TrailingStopFixedLimit { .. }
4465 | OrderKind::Market
4466 | OrderKind::TrailingStopLimit { .. } => {}
4467 }
4468
4469 builder
4470 .build()
4471 .map_err(|e| BuildOrderError::Build(e.to_string()))
4472}
4473
4474fn convert_order_kind_tif_margin(
4485 kind: OrderKind,
4486 tif: TimeInForce,
4487) -> Option<(
4488 MarginAccountNewOrderTypeEnum,
4489 Option<MarginAccountNewOrderTimeInForceEnum>,
4490)> {
4491 let (binance_type, binance_tif) = classify_order_kind_tif(kind, tif)?;
4492 let margin_type = match binance_type {
4493 BinanceOrderType::Market => MarginAccountNewOrderTypeEnum::Market,
4494 BinanceOrderType::Limit => MarginAccountNewOrderTypeEnum::Limit,
4495 BinanceOrderType::LimitMaker => MarginAccountNewOrderTypeEnum::LimitMaker,
4496 BinanceOrderType::StopLoss => MarginAccountNewOrderTypeEnum::StopLoss,
4497 BinanceOrderType::StopLossLimit => MarginAccountNewOrderTypeEnum::StopLossLimit,
4498 BinanceOrderType::TakeProfit => MarginAccountNewOrderTypeEnum::TakeProfit,
4499 BinanceOrderType::TakeProfitLimit => MarginAccountNewOrderTypeEnum::TakeProfitLimit,
4500 };
4501 let margin_tif = binance_tif.map(|t| match t {
4502 BinanceTimeInForce::Gtc => MarginAccountNewOrderTimeInForceEnum::Gtc,
4503 BinanceTimeInForce::Ioc => MarginAccountNewOrderTimeInForceEnum::Ioc,
4504 BinanceTimeInForce::Fok => MarginAccountNewOrderTimeInForceEnum::Fok,
4505 });
4506 Some((margin_type, margin_tif))
4507}
4508
4509#[cfg(test)]
4510#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
4512 use super::*;
4513 use crate::client::binance::shared::{INITIAL_BACKOFF_MS, convert_open_order, unix_ms};
4514 use crate::error::StreamTerminationReason;
4515 use crate::notice::{AccountNotice, NoticeKind};
4516 use crate::order::state::ActiveOrderState;
4517 use binance_sdk::margin_trading::rest_api::QueryMarginAccountsOpenOrdersResponseInner;
4518
4519 #[tokio::test]
4522 async fn a_banned_listen_token_subscribe_starts_the_cooldown() {
4523 let until = i64::try_from(unix_ms()).unwrap() + 120_000;
4524 let error = subscribe_error(WebsocketError::ResponseError {
4525 code: -1003,
4526 message: "Way too much request weight used".to_string(),
4527 retry_after: Some(until),
4528 });
4529 assert!(
4530 error
4531 .to_string()
4532 .starts_with("userDataStream.subscribe.listenToken failed: "),
4533 "{error}"
4534 );
4535 let tracker = RateLimitTracker::new(WeightPool::Sapi);
4536
4537 assert!(tracker.on_stream_refused(&error));
4538 assert!(
4539 tracker
4540 .cooldown_remaining()
4541 .is_some_and(|left| left > Duration::from_secs(115)),
4542 "{:?}",
4543 tracker.cooldown_remaining()
4544 );
4545 }
4546
4547 struct ManagerRig {
4551 ws_config: ConfigurationWebsocketApi,
4552 rest_config: Arc<ConfigurationRestApi>,
4553 rest: Arc<RestApi>,
4554 rate_limiter: Arc<RateLimitTracker>,
4555 requests: Arc<parking_lot::Mutex<Vec<tokio::time::Instant>>>,
4557 _rest_server: wiremock::MockServer,
4558 }
4559
4560 impl ManagerRig {
4561 async fn new(
4562 answer: fn(&serde_json::Value) -> serde_json::Value,
4563 close_after: Option<Duration>,
4564 ) -> Self {
4565 use futures::{SinkExt as _, StreamExt as _};
4566 use tokio_tungstenite::tungstenite::Message;
4567
4568 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4569 let ws_url = format!("ws://{}", listener.local_addr().unwrap());
4570 let requests = Arc::new(parking_lot::Mutex::new(Vec::new()));
4571 let seen = Arc::clone(&requests);
4572 tokio::spawn(async move {
4573 while let Ok((stream, _)) = listener.accept().await {
4574 let seen = Arc::clone(&seen);
4575 tokio::spawn(async move {
4576 let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else {
4577 return;
4578 };
4579 while let Some(Ok(message)) = ws.next().await {
4580 let Message::Text(text) = message else {
4581 continue;
4582 };
4583 seen.lock().push(tokio::time::Instant::now());
4584 let request: serde_json::Value = serde_json::from_str(&text).unwrap();
4585 let response = answer(&request).to_string();
4586 if ws.send(Message::text(response)).await.is_err() {
4587 return;
4588 }
4589 if let Some(after) = close_after {
4590 tokio::time::sleep(after).await;
4591 let _ = ws.close(None).await;
4592 return;
4593 }
4594 }
4595 });
4596 }
4597 });
4598
4599 let rest_server = wiremock::MockServer::start().await;
4600 wiremock::Mock::given(wiremock::matchers::method("POST"))
4601 .and(wiremock::matchers::path("/sapi/v1/userListenToken"))
4602 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
4603 serde_json::json!({
4604 "token": "token",
4605 "expirationTime": Utc::now().timestamp_millis() + 86_400_000,
4606 }),
4607 ))
4608 .mount(&rest_server)
4609 .await;
4610 let rest_config = Arc::new(
4611 ConfigurationRestApi::builder()
4612 .api_key("key")
4613 .api_secret("secret")
4614 .base_path(rest_server.uri())
4615 .build()
4616 .unwrap(),
4617 );
4618 Self {
4619 ws_config: ConfigurationWebsocketApi::builder()
4620 .api_key("key")
4621 .api_secret("secret")
4622 .ws_url(ws_url)
4623 .build()
4624 .unwrap(),
4625 rest: Arc::new(MarginTradingRestApi::from_config((*rest_config).clone())),
4626 rest_config,
4627 rate_limiter: Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
4628 requests,
4629 _rest_server: rest_server,
4630 }
4631 }
4632
4633 fn token() -> UserListenToken {
4635 UserListenToken {
4636 token: "token".to_string(),
4637 expiration_time_ms: Utc::now().timestamp_millis() + 86_400_000,
4638 }
4639 }
4640
4641 async fn cross(&self) -> mpsc::UnboundedReceiver<UnindexedAccountEvent> {
4643 let (tx, rx) = mpsc::unbounded_channel();
4644 let ws = connect_margin_ws(&self.ws_config).await.unwrap();
4645 tokio::spawn(margin_connection_manager(
4646 tx,
4647 new_dedup_cache(),
4648 self.ws_config.clone(),
4649 Arc::clone(&self.rest_config),
4650 Arc::clone(&self.rest),
4651 Arc::clone(&self.rate_limiter),
4652 KnownLiveOrders::shared(ExchangeId::BinanceMargin),
4653 Vec::new(),
4654 Some((ws, Self::token())),
4655 ));
4656 rx
4657 }
4658
4659 async fn isolated(&self) -> mpsc::UnboundedReceiver<UnindexedAccountEvent> {
4662 let (tx, rx) = mpsc::unbounded_channel();
4663 let (dedup, known) = (
4664 new_dedup_cache(),
4665 KnownLiveOrders::shared(ExchangeId::BinanceMargin),
4666 );
4667 let symbol = InstrumentNameExchange::new("BTCUSDT");
4668 let base_quote = Arc::new(HashMap::from([(
4669 symbol.clone(),
4670 (
4671 AssetNameExchange::new("BTC"),
4672 AssetNameExchange::new("USDT"),
4673 ),
4674 )]));
4675 let tokens = [(symbol.clone(), Self::token())];
4676 let live = isolated_connect_and_subscribe(
4677 &self.ws_config,
4678 &tx,
4679 &dedup,
4680 &known,
4681 &base_quote,
4682 &tokens,
4683 )
4684 .await
4685 .unwrap();
4686 tokio::spawn(isolated_connection_manager(
4687 tx,
4688 dedup,
4689 self.ws_config.clone(),
4690 Arc::clone(&self.rest_config),
4691 Arc::clone(&self.rest),
4692 Arc::clone(&self.rate_limiter),
4693 known,
4694 vec![symbol],
4695 base_quote,
4696 Some(live),
4697 ));
4698 rx
4699 }
4700 }
4701
4702 fn subscribed(request: &serde_json::Value) -> serde_json::Value {
4704 serde_json::json!({
4705 "id": request["id"],
4706 "status": 200,
4707 "result": { "subscriptionId": 0 },
4708 })
4709 }
4710
4711 fn banned_for_two_minutes(request: &serde_json::Value) -> serde_json::Value {
4713 let until = unix_ms() + 120_000;
4714 serde_json::json!({
4715 "id": request["id"],
4716 "status": 418,
4717 "error": {
4718 "code": -1003,
4719 "msg": format!("Way too much request weight used; IP banned until {until}."),
4720 "data": { "serverTime": unix_ms(), "retryAfter": until },
4721 },
4722 })
4723 }
4724
4725 async fn rate_limited_end(rx: &mut mpsc::UnboundedReceiver<UnindexedAccountEvent>) -> Duration {
4728 let event = tokio::time::timeout(Duration::from_secs(5), rx.recv())
4729 .await
4730 .expect("the stream ends rate-limited within the timeout")
4731 .expect("it reports why before it ends");
4732 let AccountEventKind::StreamTerminated(StreamTerminationReason::RateLimited {
4733 retry_after,
4734 }) = event.kind
4735 else {
4736 panic!("expected a rate-limited termination, got {:?}", event.kind);
4737 };
4738 assert!(rx.recv().await.is_none(), "the stream has ended");
4739 retry_after
4740 }
4741
4742 #[tokio::test]
4745 async fn a_banned_subscribe_ends_the_cross_margin_stream_rate_limited() {
4746 let rig = ManagerRig::new(banned_for_two_minutes, None).await;
4747 let mut events = rig.cross().await;
4748
4749 let wait = rate_limited_end(&mut events).await;
4750
4751 assert!(wait > Duration::from_secs(115), "{wait:?}");
4752 assert_eq!(rig.requests.lock().len(), 1, "no second subscribe");
4753 }
4754
4755 #[tokio::test]
4758 async fn a_banned_isolated_subscribe_is_recognised_as_a_refusal() {
4759 let rig = ManagerRig::new(banned_for_two_minutes, None).await;
4760 let (tx, _rx) = mpsc::unbounded_channel();
4761 let symbol = InstrumentNameExchange::new("BTCUSDT");
4762
4763 let Err(error) = isolated_connect_and_subscribe(
4764 &rig.ws_config,
4765 &tx,
4766 &new_dedup_cache(),
4767 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
4768 &Arc::new(HashMap::new()),
4769 &[(symbol, ManagerRig::token())],
4770 )
4771 .await
4772 else {
4773 panic!("a banned subscribe fails");
4774 };
4775
4776 assert!(rig.rate_limiter.on_stream_refused(&error), "{error:#}");
4777 assert!(
4778 rig.rate_limiter
4779 .cooldown_remaining()
4780 .is_some_and(|left| left > Duration::from_secs(115))
4781 );
4782 }
4783
4784 #[tokio::test]
4787 async fn a_drop_during_a_long_cooldown_ends_the_isolated_margin_stream_rate_limited() {
4788 let rig = ManagerRig::new(subscribed, Some(Duration::from_secs(1))).await;
4790 let mut events = rig.isolated().await;
4791 rig.rate_limiter
4792 .on_rate_limited(Some(Duration::from_secs(60)));
4793
4794 let wait = rate_limited_end(&mut events).await;
4795
4796 assert!(wait > Duration::from_secs(55), "{wait:?}");
4797 assert_eq!(rig.requests.lock().len(), 1, "no second subscribe");
4798 }
4799
4800 #[tokio::test]
4803 async fn a_dropped_margin_connection_backs_off_before_reconnecting() {
4804 for isolated in [false, true] {
4805 let rig = ManagerRig::new(subscribed, Some(Duration::from_millis(100))).await;
4806 let _events = if isolated {
4807 rig.isolated().await
4808 } else {
4809 rig.cross().await
4810 };
4811
4812 let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
4813 while rig.requests.lock().len() < 2 {
4814 assert!(
4815 tokio::time::Instant::now() < deadline,
4816 "isolated {isolated}: no reconnect"
4817 );
4818 tokio::time::sleep(Duration::from_millis(20)).await;
4819 }
4820
4821 let requests = rig.requests.lock();
4822 let between = requests[1] - requests[0];
4823 assert!(
4825 between >= Duration::from_millis(INITIAL_BACKOFF_MS),
4826 "isolated {isolated}: reconnected {between:?} after the first subscribe"
4827 );
4828 }
4829 }
4830
4831 #[tokio::test]
4834 async fn a_dropped_margin_connection_is_closed() {
4835 use futures::StreamExt as _;
4836 use tokio_tungstenite::tungstenite::Message;
4837
4838 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4839 let url = format!("ws://{}", listener.local_addr().unwrap());
4840 let (closed_tx, mut closed) = mpsc::unbounded_channel();
4841 tokio::spawn(async move {
4842 let (stream, _) = listener.accept().await.unwrap();
4843 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
4844 while let Some(Ok(message)) = ws.next().await {
4845 if let Message::Close(_) = message {
4846 let _ = closed_tx.send(());
4847 return;
4848 }
4849 }
4850 });
4851 let config = ConfigurationWebsocketApi::builder()
4852 .api_key("key")
4853 .api_secret("secret")
4854 .ws_url(url)
4855 .build()
4856 .unwrap();
4857
4858 drop(connect_margin_ws(&config).await.unwrap());
4859
4860 tokio::time::timeout(Duration::from_secs(5), closed.recv())
4861 .await
4862 .expect("the dropped connection is closed")
4863 .unwrap();
4864 }
4865
4866 #[test]
4867 fn margin_side_effect_default_is_auto_borrow_repay() {
4868 assert_eq!(
4869 MarginSideEffect::default(),
4870 MarginSideEffect::AutoBorrowRepay
4871 );
4872 }
4873
4874 #[test]
4875 fn margin_side_effect_wire_strings() {
4876 assert_eq!(
4877 MarginSideEffect::AutoBorrowRepay.as_binance_str(),
4878 "AUTO_BORROW_REPAY"
4879 );
4880 assert_eq!(
4881 MarginSideEffect::NoBorrow.as_binance_str(),
4882 "NO_SIDE_EFFECT"
4883 );
4884 }
4885
4886 #[test]
4887 fn margin_side_effect_serde_round_trip() {
4888 assert_eq!(
4890 serde_json::to_string(&MarginSideEffect::AutoBorrowRepay).unwrap(),
4891 r#""auto_borrow_repay""#
4892 );
4893 assert_eq!(
4894 serde_json::to_string(&MarginSideEffect::NoBorrow).unwrap(),
4895 r#""no_borrow""#
4896 );
4897 assert_eq!(
4898 serde_json::from_str::<MarginSideEffect>(r#""auto_borrow_repay""#).unwrap(),
4899 MarginSideEffect::AutoBorrowRepay
4900 );
4901 assert_eq!(
4902 serde_json::from_str::<MarginSideEffect>(r#""no_borrow""#).unwrap(),
4903 MarginSideEffect::NoBorrow
4904 );
4905 }
4906
4907 #[test]
4908 fn config_debug_redacts_secrets() {
4909 let config = BinanceMarginConfig::new(
4910 "my_api_key".to_string(),
4911 "my_secret_key".to_string(),
4912 false,
4913 false,
4914 Vec::new(),
4915 MarginSideEffect::default(),
4916 );
4917 let debug = format!("{config:?}");
4918 assert!(!debug.contains("my_api_key"));
4919 assert!(!debug.contains("my_secret_key"));
4920 assert!(debug.contains("***"));
4921 }
4922
4923 #[test]
4924 fn cross_margin_uses_common_case_defaults() {
4925 let config = BinanceMarginConfig::cross_margin("k".to_string(), "s".to_string());
4926 assert!(!config.testnet);
4927 assert!(!config.is_isolated);
4928 assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
4929 assert_eq!(config.api_key(), "k");
4930 }
4931
4932 #[test]
4933 fn config_deserializes_with_defaults() {
4934 let config: BinanceMarginConfig =
4936 serde_json::from_str(r#"{"api_key":"k","secret_key":"s","testnet":false}"#)
4937 .expect("deserialize");
4938 assert!(!config.is_isolated);
4939 assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
4940 }
4941
4942 #[test]
4943 fn config_deserializes_explicit_side_effect() {
4944 let config: BinanceMarginConfig = serde_json::from_str(
4945 r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true,"side_effect":"no_borrow"}"#,
4946 )
4947 .expect("deserialize");
4948 assert!(config.is_isolated);
4949 assert_eq!(config.side_effect, MarginSideEffect::NoBorrow);
4950 }
4951
4952 #[test]
4953 fn isolated_ctor_sets_symbols_and_flag() {
4954 let symbols = vec![
4955 InstrumentNameExchange::new("BTCUSDT"),
4956 InstrumentNameExchange::new("ETHUSDT"),
4957 ];
4958 let config =
4959 BinanceMarginConfig::isolated("k".to_string(), "s".to_string(), symbols.clone());
4960 assert!(config.is_isolated);
4961 assert_eq!(config.isolated_symbols, symbols);
4962 assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
4963 }
4964
4965 #[test]
4966 fn isolated_config_with_symbols_constructs() {
4967 let config = BinanceMarginConfig::isolated(
4969 "k".to_string(),
4970 "s".to_string(),
4971 vec![InstrumentNameExchange::new("BTCUSDT")],
4972 );
4973 let _client = BinanceMargin::new(config); }
4975
4976 #[test]
4977 #[should_panic(expected = "non-empty isolated_symbols")]
4978 fn isolated_empty_symbols_panics_at_new() {
4979 let config = BinanceMarginConfig::new(
4981 "k".to_string(),
4982 "s".to_string(),
4983 false,
4984 true,
4985 Vec::new(),
4986 MarginSideEffect::default(),
4987 );
4988 let _ = BinanceMargin::new(config);
4989 }
4990
4991 #[test]
4992 #[should_panic(expected = "non-empty isolated_symbols")]
4993 fn isolated_empty_symbols_panics_via_deserialize_path() {
4994 let config: BinanceMarginConfig = serde_json::from_str(
4997 r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true}"#,
4998 )
4999 .expect("deserialize");
5000 assert!(config.isolated_symbols.is_empty());
5001 let _ = BinanceMargin::new(config);
5002 }
5003
5004 #[test]
5005 fn isolated_flag_maps_mode() {
5006 assert_eq!(
5009 MarginAccountNewOrderIsIsolatedEnum::from_flag(true).as_str(),
5010 "TRUE"
5011 );
5012 assert_eq!(
5013 MarginAccountNewOrderIsIsolatedEnum::from_flag(false).as_str(),
5014 "FALSE"
5015 );
5016 assert_eq!(
5017 MarginAccountCancelOrderIsIsolatedEnum::from_flag(true).as_str(),
5018 "TRUE"
5019 );
5020 assert_eq!(
5021 MarginAccountCancelOrderIsIsolatedEnum::from_flag(false).as_str(),
5022 "FALSE"
5023 );
5024 assert_eq!(
5025 QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(true).as_str(),
5026 "TRUE"
5027 );
5028 assert_eq!(
5029 QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(false).as_str(),
5030 "FALSE"
5031 );
5032 assert_eq!(
5033 QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(true).as_str(),
5034 "TRUE"
5035 );
5036 assert_eq!(
5037 QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(false).as_str(),
5038 "FALSE"
5039 );
5040 }
5041
5042 #[test]
5047 fn cancel_params_cross_with_parseable_order_id() {
5048 let p = build_cancel_order_params(
5049 "BTCUSDT".to_string(),
5050 Some(&OrderId::new("12345")),
5051 &ClientOrderId::new("cid-1"),
5052 false,
5053 )
5054 .expect("build");
5055 assert_eq!(p.symbol, "BTCUSDT");
5056 assert_eq!(p.is_isolated.as_ref().map(|i| i.as_str()), Some("FALSE"));
5057 assert_eq!(p.order_id, Some(12345));
5058 assert_eq!(p.orig_client_order_id, None);
5059 }
5060
5061 #[test]
5062 fn cancel_params_isolated_is_config_driven() {
5063 let p = build_cancel_order_params(
5064 "BTCUSDT".to_string(),
5065 Some(&OrderId::new("12345")),
5066 &ClientOrderId::new("cid-1"),
5067 true,
5068 )
5069 .expect("build");
5070 assert_eq!(p.is_isolated.as_ref().map(|i| i.as_str()), Some("TRUE"));
5071 assert_eq!(p.order_id, Some(12345));
5072 }
5073
5074 #[test]
5075 fn cancel_params_non_i64_order_id_falls_back_to_cid() {
5076 let p = build_cancel_order_params(
5077 "BTCUSDT".to_string(),
5078 Some(&OrderId::new("not-an-i64")),
5079 &ClientOrderId::new("cid-1"),
5080 false,
5081 )
5082 .expect("build");
5083 assert_eq!(p.order_id, None);
5084 assert_eq!(p.orig_client_order_id.as_deref(), Some("cid-1"));
5085 }
5086
5087 #[test]
5088 fn cancel_params_absent_order_id_uses_cid() {
5089 let p = build_cancel_order_params(
5090 "BTCUSDT".to_string(),
5091 None,
5092 &ClientOrderId::new("cid-1"),
5093 false,
5094 )
5095 .expect("build");
5096 assert_eq!(p.order_id, None);
5097 assert_eq!(p.orig_client_order_id.as_deref(), Some("cid-1"));
5098 }
5099
5100 use crate::order::TrailingOffsetType;
5105
5106 fn params(
5108 side: Side,
5109 price: Option<Decimal>,
5110 kind: OrderKind,
5111 tif: TimeInForce,
5112 side_effect: MarginSideEffect,
5113 ) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
5114 build_new_order_params(
5115 "BTCUSDT".to_string(),
5116 side,
5117 price,
5118 Decimal::from(2),
5119 kind,
5120 tif,
5121 "cid-1".to_string(),
5122 side_effect,
5123 false, )
5125 }
5126
5127 fn gtc() -> TimeInForce {
5128 TimeInForce::GoodUntilCancelled { post_only: false }
5129 }
5130
5131 #[test]
5132 fn new_order_limit_maps_core_fields() {
5133 let p = params(
5134 Side::Buy,
5135 Some(Decimal::from(50_000)),
5136 OrderKind::Limit,
5137 gtc(),
5138 MarginSideEffect::AutoBorrowRepay,
5139 )
5140 .expect("build");
5141
5142 assert_eq!(p.symbol, "BTCUSDT");
5143 assert_eq!(p.side.as_str(), "BUY");
5144 assert_eq!(p.r#type.as_str(), "LIMIT");
5145 assert_eq!(p.quantity, Some(Decimal::from(2)));
5146 assert_eq!(p.price, Some(Decimal::from(50_000)));
5147 assert_eq!(p.stop_price, None);
5148 assert_eq!(p.time_in_force.as_ref().map(|t| t.as_str()), Some("GTC"));
5149 assert_eq!(p.new_client_order_id.as_deref(), Some("cid-1"));
5150 assert_eq!(
5152 p.new_order_resp_type.as_ref().map(|r| r.as_str()),
5153 Some("FULL")
5154 );
5155 }
5156
5157 #[test]
5158 fn new_order_is_isolated_is_config_driven() {
5159 let cross = build_new_order_params(
5161 "BTCUSDT".to_string(),
5162 Side::Sell,
5163 Some(Decimal::from(10)),
5164 Decimal::from(2),
5165 OrderKind::Limit,
5166 gtc(),
5167 "cid-1".to_string(),
5168 MarginSideEffect::AutoBorrowRepay,
5169 false,
5170 )
5171 .expect("build");
5172 assert_eq!(
5173 cross.is_isolated.as_ref().map(|i| i.as_str()),
5174 Some("FALSE")
5175 );
5176 assert_eq!(cross.side.as_str(), "SELL");
5177
5178 let isolated = build_new_order_params(
5180 "BTCUSDT".to_string(),
5181 Side::Sell,
5182 Some(Decimal::from(10)),
5183 Decimal::from(2),
5184 OrderKind::Limit,
5185 gtc(),
5186 "cid-1".to_string(),
5187 MarginSideEffect::AutoBorrowRepay,
5188 true,
5189 )
5190 .expect("build");
5191 assert_eq!(
5192 isolated.is_isolated.as_ref().map(|i| i.as_str()),
5193 Some("TRUE")
5194 );
5195 }
5196
5197 #[test]
5198 fn side_effect_and_auto_repay_gating() {
5199 let auto = params(
5201 Side::Buy,
5202 Some(Decimal::from(1)),
5203 OrderKind::Limit,
5204 gtc(),
5205 MarginSideEffect::AutoBorrowRepay,
5206 )
5207 .expect("build");
5208 assert_eq!(
5209 auto.side_effect_type.as_ref().map(|s| s.as_str()),
5210 Some("AUTO_BORROW_REPAY")
5211 );
5212 assert_eq!(auto.auto_repay_at_cancel, Some(true));
5213
5214 let no_borrow = params(
5216 Side::Buy,
5217 Some(Decimal::from(1)),
5218 OrderKind::Limit,
5219 gtc(),
5220 MarginSideEffect::NoBorrow,
5221 )
5222 .expect("build");
5223 assert_eq!(
5224 no_borrow.side_effect_type.as_ref().map(|s| s.as_str()),
5225 Some("NO_SIDE_EFFECT")
5226 );
5227 assert_eq!(no_borrow.auto_repay_at_cancel, None);
5228 }
5229
5230 #[test]
5231 fn new_order_market_has_no_price_or_tif() {
5232 let p = params(
5233 Side::Buy,
5234 None,
5235 OrderKind::Market,
5236 gtc(),
5237 MarginSideEffect::AutoBorrowRepay,
5238 )
5239 .expect("build");
5240 assert_eq!(p.r#type.as_str(), "MARKET");
5241 assert_eq!(p.price, None);
5242 assert_eq!(p.stop_price, None);
5243 assert!(p.time_in_force.is_none());
5244 }
5245
5246 #[test]
5247 fn post_only_limit_maps_to_limit_maker() {
5248 let p = params(
5249 Side::Buy,
5250 Some(Decimal::from(10)),
5251 OrderKind::Limit,
5252 TimeInForce::GoodUntilCancelled { post_only: true },
5253 MarginSideEffect::AutoBorrowRepay,
5254 )
5255 .expect("build");
5256 assert_eq!(p.r#type.as_str(), "LIMIT_MAKER");
5257 assert!(p.time_in_force.is_none());
5259 }
5260
5261 #[test]
5262 fn conditional_kinds_set_stop_price() {
5263 let trigger = Decimal::from(48_000);
5264
5265 let stop = params(
5266 Side::Sell,
5267 None,
5268 OrderKind::Stop {
5269 trigger_price: trigger,
5270 },
5271 gtc(),
5272 MarginSideEffect::AutoBorrowRepay,
5273 )
5274 .expect("build");
5275 assert_eq!(stop.r#type.as_str(), "STOP_LOSS");
5276 assert_eq!(stop.stop_price, Some(trigger));
5277 assert_eq!(stop.price, None);
5278
5279 let stop_limit = params(
5280 Side::Sell,
5281 Some(Decimal::from(47_900)),
5282 OrderKind::StopLimit {
5283 trigger_price: trigger,
5284 },
5285 gtc(),
5286 MarginSideEffect::AutoBorrowRepay,
5287 )
5288 .expect("build");
5289 assert_eq!(stop_limit.r#type.as_str(), "STOP_LOSS_LIMIT");
5290 assert_eq!(stop_limit.stop_price, Some(trigger));
5291 assert_eq!(stop_limit.price, Some(Decimal::from(47_900)));
5292 assert_eq!(
5293 stop_limit.time_in_force.as_ref().map(|t| t.as_str()),
5294 Some("GTC")
5295 );
5296
5297 let take_profit = params(
5298 Side::Sell,
5299 None,
5300 OrderKind::TakeProfit {
5301 trigger_price: trigger,
5302 },
5303 gtc(),
5304 MarginSideEffect::AutoBorrowRepay,
5305 )
5306 .expect("build");
5307 assert_eq!(take_profit.r#type.as_str(), "TAKE_PROFIT");
5308 assert_eq!(take_profit.stop_price, Some(trigger));
5309 }
5310
5311 #[test]
5314 fn a_trailing_stop_maps_to_stop_loss_with_trailing_delta() {
5315 let cases = [
5316 (Decimal::from(25), TrailingOffsetType::BasisPoints, 25),
5317 (Decimal::new(15, 1), TrailingOffsetType::Percentage, 150),
5318 ];
5319 for (offset, offset_type, trailing_delta) in cases {
5320 let p = params(
5321 Side::Sell,
5322 None,
5323 OrderKind::TrailingStop {
5324 offset,
5325 offset_type,
5326 activation_price: None,
5327 },
5328 gtc(),
5329 MarginSideEffect::NoBorrow,
5330 )
5331 .expect("build");
5332 assert_eq!(p.r#type.as_str(), "STOP_LOSS");
5333 assert_eq!(p.trailing_delta, Some(trailing_delta), "{offset_type:?}");
5334 assert_eq!(p.stop_price, None);
5335 assert_eq!(p.price, None);
5336 assert!(p.time_in_force.is_none());
5337 }
5338 }
5339
5340 #[test]
5344 fn trailing_orders_carry_their_activation_price_and_fixed_limit() {
5345 let activation = Some(Decimal::from(50_000));
5346 let trailing_stop = params(
5347 Side::Sell,
5348 None,
5349 OrderKind::TrailingStop {
5350 offset: Decimal::from(25),
5351 offset_type: TrailingOffsetType::BasisPoints,
5352 activation_price: activation,
5353 },
5354 gtc(),
5355 MarginSideEffect::NoBorrow,
5356 )
5357 .expect("build");
5358 assert_eq!(trailing_stop.r#type.as_str(), "STOP_LOSS");
5359 assert_eq!(trailing_stop.trailing_delta, Some(25));
5360 assert_eq!(trailing_stop.stop_price, activation);
5361 assert_eq!(trailing_stop.price, None);
5362
5363 for activation_price in [None, activation] {
5364 let fixed_limit = params(
5365 Side::Sell,
5366 Some(Decimal::from(49_000)),
5367 OrderKind::TrailingStopFixedLimit {
5368 offset: Decimal::new(15, 1),
5369 offset_type: TrailingOffsetType::Percentage,
5370 activation_price,
5371 },
5372 TimeInForce::ImmediateOrCancel,
5373 MarginSideEffect::NoBorrow,
5374 )
5375 .expect("build");
5376 assert_eq!(fixed_limit.r#type.as_str(), "STOP_LOSS_LIMIT");
5377 assert_eq!(fixed_limit.trailing_delta, Some(150));
5378 assert_eq!(fixed_limit.price, Some(Decimal::from(49_000)));
5379 assert_eq!(fixed_limit.stop_price, activation_price);
5380 assert_eq!(
5381 fixed_limit.time_in_force.as_ref().map(|t| t.as_str()),
5382 Some("IOC")
5383 );
5384 }
5385
5386 let absolute = params(
5387 Side::Sell,
5388 Some(Decimal::from(49_000)),
5389 OrderKind::TrailingStopFixedLimit {
5390 offset: Decimal::from(100),
5391 offset_type: TrailingOffsetType::Absolute,
5392 activation_price: None,
5393 },
5394 gtc(),
5395 MarginSideEffect::NoBorrow,
5396 );
5397 assert!(matches!(absolute, Err(BuildOrderError::Unsupported)));
5398
5399 let no_limit = params(
5401 Side::Sell,
5402 None,
5403 OrderKind::TrailingStopFixedLimit {
5404 offset: Decimal::from(25),
5405 offset_type: TrailingOffsetType::BasisPoints,
5406 activation_price: None,
5407 },
5408 gtc(),
5409 MarginSideEffect::NoBorrow,
5410 );
5411 assert!(matches!(
5412 no_limit,
5413 Err(BuildOrderError::Trailing(TrailingOrderError::LimitPrice(
5414 None
5415 )))
5416 ));
5417 let zero_activation = params(
5418 Side::Sell,
5419 None,
5420 OrderKind::TrailingStop {
5421 offset: Decimal::from(25),
5422 offset_type: TrailingOffsetType::BasisPoints,
5423 activation_price: Some(Decimal::ZERO),
5424 },
5425 gtc(),
5426 MarginSideEffect::NoBorrow,
5427 );
5428 assert!(matches!(
5429 zero_activation,
5430 Err(BuildOrderError::Trailing(
5431 TrailingOrderError::NonPositiveActivation(_)
5432 ))
5433 ));
5434 }
5435
5436 #[test]
5437 fn absolute_trailing_offsets_and_trailing_stop_limits_are_unsupported() {
5438 let absolute = params(
5439 Side::Sell,
5440 None,
5441 OrderKind::TrailingStop {
5442 offset: Decimal::from(100),
5443 offset_type: TrailingOffsetType::Absolute,
5444 activation_price: None,
5445 },
5446 gtc(),
5447 MarginSideEffect::AutoBorrowRepay,
5448 );
5449 assert!(matches!(absolute, Err(BuildOrderError::Unsupported)));
5450
5451 let trailing_stop_limit = params(
5452 Side::Sell,
5453 Some(Decimal::from(100)),
5454 OrderKind::TrailingStopLimit {
5455 offset: Decimal::from(100),
5456 offset_type: TrailingOffsetType::BasisPoints,
5457 stop_price: Decimal::from(95),
5458 limit_offset: Decimal::from(10),
5459 },
5460 gtc(),
5461 MarginSideEffect::AutoBorrowRepay,
5462 );
5463 assert!(matches!(
5464 trailing_stop_limit,
5465 Err(BuildOrderError::Unsupported)
5466 ));
5467 }
5468
5469 #[test]
5471 fn a_fractional_trailing_offset_is_refused() {
5472 let fractional = params(
5473 Side::Sell,
5474 None,
5475 OrderKind::TrailingStop {
5476 offset: Decimal::new(155, 3),
5477 offset_type: TrailingOffsetType::Percentage,
5478 activation_price: None,
5479 },
5480 gtc(),
5481 MarginSideEffect::AutoBorrowRepay,
5482 );
5483 assert!(matches!(
5484 fractional,
5485 Err(BuildOrderError::Trailing(TrailingOrderError::Delta(
5486 crate::client::binance::shared::TrailingDeltaError::Invalid(_)
5487 )))
5488 ));
5489 }
5490
5491 #[test]
5492 fn tif_variants_map_to_margin_enum() {
5493 for (tif, expected) in [
5494 (TimeInForce::ImmediateOrCancel, "IOC"),
5495 (TimeInForce::FillOrKill, "FOK"),
5496 (gtc(), "GTC"),
5497 ] {
5498 let p = params(
5499 Side::Buy,
5500 Some(Decimal::from(1)),
5501 OrderKind::Limit,
5502 tif,
5503 MarginSideEffect::AutoBorrowRepay,
5504 )
5505 .expect("build");
5506 assert_eq!(p.time_in_force.as_ref().map(|t| t.as_str()), Some(expected));
5507 }
5508 }
5509
5510 fn user_asset(
5515 asset: &str,
5516 free: &str,
5517 locked: &str,
5518 borrowed: &str,
5519 interest: &str,
5520 ) -> QueryCrossMarginAccountDetailsResponseUserAssetsInner {
5521 let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
5522 a.asset = Some(asset.to_string());
5523 a.free = Some(free.to_string());
5524 a.locked = Some(locked.to_string());
5525 a.borrowed = Some(borrowed.to_string());
5526 a.interest = Some(interest.to_string());
5527 a
5528 }
5529
5530 #[test]
5531 fn margin_balance_entry_maps_debt() {
5532 let ab =
5534 convert_margin_balance_entry(user_asset("USDT", "10", "2", "3", "0.5"), Utc::now())
5535 .expect("convert");
5536 assert_eq!(ab.asset.name().as_str(), "USDT");
5537 assert_eq!(ab.balance.total, Decimal::from(12));
5538 assert_eq!(ab.balance.free, Decimal::from(10));
5539 assert!(
5540 ab.balance.margin.is_some(),
5541 "REST snapshot must populate margin"
5542 );
5543 assert_eq!(ab.balance.net_asset(), Decimal::from(9));
5545 }
5546
5547 #[test]
5548 fn margin_balance_short_is_negative_net() {
5549 let ab = convert_margin_balance_entry(user_asset("BTC", "1", "0", "5", "0"), Utc::now())
5551 .expect("convert");
5552 assert_eq!(ab.balance.net_asset(), Decimal::from(-4));
5553 }
5554
5555 #[test]
5556 fn margin_balance_missing_debt_defaults_to_zero() {
5557 let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
5560 a.asset = Some("ETH".to_string());
5561 a.free = Some("4".to_string());
5562 a.locked = Some("0".to_string());
5563 let ab = convert_margin_balance_entry(a, Utc::now()).expect("convert");
5564 assert!(ab.balance.margin.is_some());
5565 assert_eq!(ab.balance.net_asset(), Decimal::from(4));
5566 }
5567
5568 #[test]
5569 fn margin_balance_missing_free_is_dropped() {
5570 let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
5572 a.asset = Some("USDT".to_string());
5573 a.locked = Some("0".to_string());
5574 assert!(convert_margin_balance_entry(a, Utc::now()).is_none());
5575 }
5576
5577 #[test]
5578 fn margin_balance_filtering() {
5579 let assets = vec![
5580 user_asset("USDT", "10", "0", "0", "0"),
5581 user_asset("BTC", "1", "0", "0", "0"),
5582 user_asset("ETH", "5", "0", "0", "0"),
5583 ];
5584 assert_eq!(
5586 filter_and_convert_margin_balances(assets.clone(), &[]).len(),
5587 3
5588 );
5589 let filtered = filter_and_convert_margin_balances(
5591 assets,
5592 &[AssetNameExchange::new("BTC"), AssetNameExchange::new("ETH")],
5593 );
5594 assert_eq!(filtered.len(), 2);
5595 assert!(filtered.iter().all(|b| b.asset.name().as_str() != "USDT"));
5596 }
5597
5598 #[test]
5599 fn margin_balance_filtering_large_slice_uses_hashset_branch() {
5600 let assets = vec![
5602 user_asset("BTC", "1", "0", "0", "0"),
5603 user_asset("ETH", "5", "0", "0", "0"),
5604 user_asset("DOGE", "9", "0", "0", "0"), ];
5606 let mut requested: Vec<AssetNameExchange> = (0..17)
5608 .map(|i| AssetNameExchange::new(format!("A{i:02}")))
5609 .collect();
5610 requested.push(AssetNameExchange::new("BTC"));
5611 requested.push(AssetNameExchange::new("ETH"));
5612 assert!(
5613 requested.len() > 16,
5614 "must exceed the linear-scan threshold"
5615 );
5616
5617 let filtered = filter_and_convert_margin_balances(assets, &requested);
5618 assert_eq!(filtered.len(), 2);
5619 assert!(filtered.iter().all(|b| b.asset.name().as_str() != "DOGE"));
5620 }
5621
5622 use binance_sdk::margin_trading::rest_api::{
5627 QueryIsolatedMarginAccountInfoResponseAssetsInnerBaseAsset as IsoBase,
5628 QueryIsolatedMarginAccountInfoResponseAssetsInnerQuoteAsset as IsoQuote,
5629 };
5630
5631 fn d(s: &str) -> Decimal {
5632 Decimal::from_str(s).unwrap()
5633 }
5634
5635 fn iso_base(asset: &str, free: &str, locked: &str, borrowed: &str, interest: &str) -> IsoBase {
5636 let mut b = IsoBase::new();
5637 b.asset = Some(asset.to_string());
5638 b.free = Some(free.to_string());
5639 b.locked = Some(locked.to_string());
5640 b.borrowed = Some(borrowed.to_string());
5641 b.interest = Some(interest.to_string());
5642 b
5643 }
5644
5645 fn iso_quote(
5646 asset: &str,
5647 free: &str,
5648 locked: &str,
5649 borrowed: &str,
5650 interest: &str,
5651 ) -> IsoQuote {
5652 let mut q = IsoQuote::new();
5653 q.asset = Some(asset.to_string());
5654 q.free = Some(free.to_string());
5655 q.locked = Some(locked.to_string());
5656 q.borrowed = Some(borrowed.to_string());
5657 q.interest = Some(interest.to_string());
5658 q
5659 }
5660
5661 fn iso_entry(
5662 symbol: &str,
5663 base: IsoBase,
5664 quote: IsoQuote,
5665 margin_level: Option<&str>,
5666 margin_ratio: Option<&str>,
5667 liquidate_price: Option<&str>,
5668 ) -> QueryIsolatedMarginAccountInfoResponseAssetsInner {
5669 let mut e = QueryIsolatedMarginAccountInfoResponseAssetsInner::new();
5670 e.symbol = Some(symbol.to_string());
5671 e.base_asset = Some(Box::new(base));
5672 e.quote_asset = Some(Box::new(quote));
5673 e.margin_level = margin_level.map(str::to_string);
5674 e.margin_ratio = margin_ratio.map(str::to_string);
5675 e.liquidate_price = liquidate_price.map(str::to_string);
5676 e
5677 }
5678
5679 fn isolated_client(symbols: &[&str]) -> BinanceMargin {
5680 let config = BinanceMarginConfig::isolated(
5681 "k".to_string(),
5682 "s".to_string(),
5683 symbols
5684 .iter()
5685 .map(|s| InstrumentNameExchange::new(*s))
5686 .collect(),
5687 );
5688 BinanceMargin::new(config)
5689 }
5690
5691 #[test]
5692 fn isolated_assets_map_base_quote_and_risk() {
5693 let entry = iso_entry(
5696 "BTCUSDT",
5697 iso_base("BTC", "0.5", "0.1", "0.2", "0.001"),
5698 iso_quote("USDT", "1000", "50", "300", "1.5"),
5699 Some("3.5"),
5700 Some("0.12"),
5701 Some("48000"),
5702 );
5703 let map = convert_isolated_margin_assets(vec![entry]);
5704 let state = map
5705 .get(&InstrumentNameExchange::new("BTCUSDT"))
5706 .expect("entry present");
5707
5708 assert_eq!(state.base.asset.name().as_str(), "BTC");
5709 assert_eq!(state.base.balance.total, d("0.6"));
5710 assert_eq!(state.base.balance.free, d("0.5"));
5711 assert_eq!(state.base.balance.net_asset(), d("0.4"));
5712
5713 assert_eq!(state.quote.asset.name().as_str(), "USDT");
5714 assert_eq!(state.quote.balance.total, d("1050"));
5715 assert_eq!(state.quote.balance.net_asset(), d("750"));
5716
5717 assert_eq!(state.risk.margin_level, Some(d("3.5")));
5718 assert_eq!(state.risk.margin_ratio, Some(d("0.12")));
5719 assert_eq!(state.risk.liquidation_price, Some(d("48000")));
5720 }
5721
5722 #[test]
5723 fn isolated_base_short_is_negative_net() {
5724 let entry = iso_entry(
5726 "BTCUSDT",
5727 iso_base("BTC", "0", "0", "1.5", "0.001"),
5728 iso_quote("USDT", "100", "0", "0", "0"),
5729 None,
5730 None,
5731 None,
5732 );
5733 let map = convert_isolated_margin_assets(vec![entry]);
5734 let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
5735 assert_eq!(state.base.balance.net_asset(), d("-1.5"));
5736 }
5737
5738 #[test]
5739 fn isolated_missing_debt_defaults_zero_and_risk_optional() {
5740 let mut base = IsoBase::new();
5743 base.asset = Some("BTC".to_string());
5744 base.free = Some("0.5".to_string());
5745 base.locked = Some("0".to_string());
5746 let mut quote = IsoQuote::new();
5747 quote.asset = Some("USDT".to_string());
5748 quote.free = Some("100".to_string());
5749 quote.locked = Some("0".to_string());
5750
5751 let map = convert_isolated_margin_assets(vec![iso_entry(
5752 "BTCUSDT", base, quote, None, None, None,
5753 )]);
5754 let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
5755 assert!(state.base.balance.margin.is_some());
5756 assert_eq!(state.base.balance.net_asset(), d("0.5"));
5757 assert_eq!(state.risk.margin_level, None);
5758 assert_eq!(state.risk.liquidation_price, None);
5759 }
5760
5761 #[test]
5762 fn isolated_unparseable_risk_is_none_but_entry_kept() {
5763 let entry = iso_entry(
5764 "BTCUSDT",
5765 iso_base("BTC", "1", "0", "0", "0"),
5766 iso_quote("USDT", "1", "0", "0", "0"),
5767 Some("not_a_number"),
5768 None,
5769 None,
5770 );
5771 let map = convert_isolated_margin_assets(vec![entry]);
5772 let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
5773 assert_eq!(state.risk.margin_level, None);
5774 }
5775
5776 #[test]
5777 fn isolated_missing_base_free_drops_entry() {
5778 let mut base = IsoBase::new();
5780 base.asset = Some("BTC".to_string());
5781 base.locked = Some("0".to_string()); let entry = iso_entry(
5783 "BTCUSDT",
5784 base,
5785 iso_quote("USDT", "100", "0", "0", "0"),
5786 None,
5787 None,
5788 None,
5789 );
5790 assert!(convert_isolated_margin_assets(vec![entry]).is_empty());
5791 }
5792
5793 #[test]
5794 fn isolated_missing_symbol_drops_entry() {
5795 let mut e = QueryIsolatedMarginAccountInfoResponseAssetsInner::new();
5796 e.base_asset = Some(Box::new(iso_base("BTC", "1", "0", "0", "0")));
5797 e.quote_asset = Some(Box::new(iso_quote("USDT", "1", "0", "0", "0")));
5798 assert!(convert_isolated_margin_assets(vec![e]).is_empty());
5800 }
5801
5802 #[test]
5803 fn chunk_symbols_batches_by_five() {
5804 let syms: Vec<InstrumentNameExchange> = (0..12)
5805 .map(|i| InstrumentNameExchange::new(format!("S{i:02}")))
5806 .collect();
5807 let chunks = chunk_symbols(&syms);
5808 assert_eq!(chunks.len(), 3, "12 symbols → 5 + 5 + 2");
5809 assert_eq!(chunks[0].split(',').count(), 5);
5810 assert_eq!(chunks[1].split(',').count(), 5);
5811 assert_eq!(chunks[2].split(',').count(), 2);
5812
5813 let three: Vec<_> = (0..3)
5815 .map(|i| InstrumentNameExchange::new(format!("S{i}")))
5816 .collect();
5817 assert_eq!(chunk_symbols(&three).len(), 1);
5818
5819 let five: Vec<_> = (0..5)
5821 .map(|i| InstrumentNameExchange::new(format!("S{i}")))
5822 .collect();
5823 let c = chunk_symbols(&five);
5824 assert_eq!(c.len(), 1);
5825 assert_eq!(c[0].split(',').count(), 5);
5826
5827 assert!(chunk_symbols(&[]).is_empty());
5829 }
5830
5831 #[test]
5832 fn effective_isolated_set_empty_returns_all_configured() {
5833 let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
5834 assert_eq!(
5835 client.effective_isolated_set(&[]),
5836 vec![
5837 InstrumentNameExchange::new("BTCUSDT"),
5838 InstrumentNameExchange::new("ETHUSDT"),
5839 ]
5840 );
5841 }
5842
5843 #[test]
5844 fn effective_isolated_set_intersects_requested() {
5845 let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
5846 assert_eq!(
5847 client.effective_isolated_set(&[InstrumentNameExchange::new("ETHUSDT")]),
5848 vec![InstrumentNameExchange::new("ETHUSDT")]
5849 );
5850 }
5851
5852 #[test]
5853 fn effective_isolated_set_skips_out_of_set() {
5854 let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
5855 assert_eq!(
5857 client.effective_isolated_set(&[
5858 InstrumentNameExchange::new("BTCUSDT"),
5859 InstrumentNameExchange::new("DOGEUSDT"),
5860 ]),
5861 vec![InstrumentNameExchange::new("BTCUSDT")]
5862 );
5863 }
5864
5865 #[tokio::test]
5866 async fn fetch_balances_isolated_returns_empty() {
5867 let client = isolated_client(&["BTCUSDT"]);
5870 assert!(client.fetch_balances(&[]).await.expect("ok").is_empty());
5871 }
5872
5873 fn open_order(order_id: Option<i64>, side: &str) -> QueryMarginAccountsOpenOrdersResponseInner {
5874 let mut o = QueryMarginAccountsOpenOrdersResponseInner::new();
5875 o.order_id = order_id;
5876 o.client_order_id = Some("cid-9".to_string());
5877 o.side = Some(side.to_string());
5878 o.price = Some("50000".to_string());
5879 o.orig_qty = Some("2".to_string());
5880 o.executed_qty = Some("0.5".to_string());
5881 o.r#type = Some("LIMIT".to_string());
5882 o.time_in_force = Some("GTC".to_string());
5883 o.time = Some(1_700_000_000_000);
5884 o.status = Some("PARTIALLY_FILLED".to_string());
5885 o
5886 }
5887
5888 #[test]
5889 fn margin_open_order_converts() {
5890 let inst = InstrumentNameExchange::new("BTCUSDT");
5891 let order = convert_open_order(
5892 &open_order(Some(42), "BUY"),
5893 ExchangeId::BinanceMargin,
5894 &inst,
5895 )
5896 .expect("convert");
5897 assert_eq!(order.key.exchange, ExchangeId::BinanceMargin);
5898 assert_eq!(
5899 order.state.id.assigned().map(|id| id.0.as_str()),
5900 Some("42")
5901 );
5902 assert_eq!(order.key.cid.0.as_str(), "cid-9");
5903 assert_eq!(order.side, Side::Buy);
5904 assert_eq!(order.price, Some(Decimal::from(50_000)));
5905 assert_eq!(order.quantity, Decimal::from(2));
5906 assert_eq!(order.state.filled_quantity, Decimal::new(5, 1));
5907 assert_eq!(order.kind, OrderKind::Limit);
5908 }
5909
5910 #[test]
5913 fn margin_open_order_that_is_not_live_is_dropped() {
5914 let inst = InstrumentNameExchange::new("BTCUSDT");
5915 let mut o = open_order(Some(42), "BUY");
5916 o.status = Some("CANCELED".to_string());
5917 assert!(convert_open_order(&o, ExchangeId::BinanceMargin, &inst).is_none());
5918 }
5919
5920 #[test]
5921 fn margin_open_order_missing_order_id_is_dropped() {
5922 let inst = InstrumentNameExchange::new("BTCUSDT");
5923 assert!(
5924 convert_open_order(&open_order(None, "BUY"), ExchangeId::BinanceMargin, &inst)
5925 .is_none()
5926 );
5927 }
5928
5929 #[test]
5935 fn margin_open_order_prefers_update_time_over_creation_time() {
5936 let inst = InstrumentNameExchange::new("BTCUSDT");
5937
5938 let mut updated = open_order(Some(42), "BUY");
5939 updated.update_time = Some(1_700_000_005_000);
5940 let order =
5941 convert_open_order(&updated, ExchangeId::BinanceMargin, &inst).expect("convert");
5942 assert_eq!(
5943 order.state.time_exchange,
5944 Utc.timestamp_millis_opt(1_700_000_005_000)
5945 .single()
5946 .unwrap(),
5947 "update_time must win over time"
5948 );
5949 }
5950
5951 #[test]
5954 fn margin_open_order_falls_back_to_creation_time_when_update_time_absent() {
5955 let inst = InstrumentNameExchange::new("BTCUSDT");
5956
5957 let base = open_order(Some(42), "BUY");
5958 assert!(base.update_time.is_none(), "fixture must omit update_time");
5959 let order = convert_open_order(&base, ExchangeId::BinanceMargin, &inst).expect("convert");
5960 assert_eq!(
5961 order.state.time_exchange,
5962 Utc.timestamp_millis_opt(1_700_000_000_000)
5963 .single()
5964 .unwrap(),
5965 "absent update_time falls back to time"
5966 );
5967 }
5968
5969 #[test]
5970 fn margin_open_order_owned_symbol_recovers_instrument() {
5971 let mut o = open_order(Some(42), "SELL");
5973 o.symbol = Some("ETHUSDT".to_string());
5974 let order =
5975 convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin).expect("convert");
5976 assert_eq!(order.key.instrument.name().as_str(), "ETHUSDT");
5977 assert_eq!(order.key.exchange, ExchangeId::BinanceMargin);
5978 assert_eq!(order.side, Side::Sell);
5979 }
5980
5981 #[test]
5982 fn margin_open_order_owned_symbol_missing_symbol_is_dropped() {
5983 let o = open_order(Some(42), "BUY");
5985 assert!(o.symbol.is_none());
5986 assert!(convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin).is_none());
5987 }
5988
5989 fn trade(id: Option<i64>, is_buyer: Option<bool>) -> QueryMarginAccountsTradeListResponseInner {
5990 let mut t = QueryMarginAccountsTradeListResponseInner::new();
5991 t.id = id;
5992 t.order_id = Some(7);
5993 t.is_buyer = is_buyer;
5994 t.price = Some("48000".to_string());
5995 t.qty = Some("0.25".to_string());
5996 t.commission = Some("0.001".to_string());
5997 t.commission_asset = Some("BNB".to_string());
5998 t.time = Some(1_700_000_000_000);
5999 t
6000 }
6001
6002 #[test]
6003 fn margin_trade_converts() {
6004 let inst = InstrumentNameExchange::new("BTCUSDT");
6005 let tr = convert_margin_trade(&trade(Some(11), Some(false)), &inst).expect("convert");
6006 assert_eq!(tr.id.0.as_str(), "11");
6007 assert_eq!(tr.order_id.0.as_str(), "7");
6008 assert_eq!(tr.side, Side::Sell);
6009 assert_eq!(tr.price, Decimal::from(48_000));
6010 assert_eq!(tr.quantity, Decimal::new(25, 2));
6011 assert_eq!(tr.fees.asset.name().as_str(), "BNB");
6013 assert_eq!(tr.fees.fees, Decimal::new(1, 3));
6014 }
6015
6016 #[test]
6017 fn margin_trade_missing_fields_are_dropped() {
6018 let inst = InstrumentNameExchange::new("BTCUSDT");
6019 assert!(convert_margin_trade(&trade(None, Some(true)), &inst).is_none());
6021 assert!(convert_margin_trade(&trade(Some(11), None), &inst).is_none());
6022 }
6023
6024 #[test]
6027 fn token_renew_after_applies_margin_and_floor() {
6028 let now_ms = Utc::now().timestamp_millis();
6029 let far = token_renew_after(now_ms + 86_400_000).as_secs();
6031 assert!(
6032 far > 80_000 && far <= 86_400,
6033 "expected ~24h-minus-margin, got {far}"
6034 );
6035 assert_eq!(
6037 token_renew_after(now_ms - 10_000).as_secs(),
6038 TOKEN_MIN_LIFETIME_SECS
6039 );
6040 assert_eq!(
6042 token_renew_after(now_ms + 60_000).as_secs(),
6043 TOKEN_MIN_LIFETIME_SECS
6044 );
6045 }
6046
6047 fn push(event: serde_json::Value) -> String {
6052 serde_json::json!({ "subscriptionId": 1, "event": event }).to_string()
6053 }
6054
6055 #[test]
6056 fn margin_ws_rpc_ack_yields_no_events() {
6057 let frame = serde_json::json!({ "id": "abc", "status": 200, "result": {} }).to_string();
6059 let mut buf = Vec::new();
6060 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6061 assert!(buf.is_empty());
6062 }
6063
6064 #[test]
6067 fn margin_ws_unrecognised_frames_yield_no_events() {
6068 for frame in [
6069 r#"{"e":"eventStreamTerminated","E":1}"#,
6070 r#"{"subscriptionId":1,"event":{"E":1}}"#,
6071 ] {
6072 let mut buf = Vec::new();
6073 assert!(!convert_margin_user_data_events(frame, &mut buf), "{frame}");
6074 assert!(buf.is_empty(), "{frame}");
6075 }
6076 }
6077
6078 #[test]
6081 fn margin_ws_trailing_stop_report_decodes_as_a_trailing_stop() {
6082 for trailing_delta in [serde_json::json!(25), serde_json::json!("25")] {
6083 let frame = push(serde_json::json!({
6084 "e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "STOP_LOSS",
6085 "x": "NEW", "X": "NEW", "i": 12_345_i64, "c": "cid-1", "t": -1_i64,
6086 "p": "0.00000000", "P": "0.00000000", "d": trailing_delta, "D": 1_700_000_000_000_i64,
6087 "q": "2", "f": "GTC", "z": "0", "T": 1_700_000_000_000_i64,
6088 }));
6089 let mut buf = Vec::new();
6090 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6091 let [event] = buf.as_slice() else {
6092 panic!("one snapshot expected for {frame}, got {buf:?}");
6093 };
6094 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
6095 panic!("expected an order snapshot, got {event:?}");
6096 };
6097 assert_eq!(
6098 snapshot.0.kind,
6099 OrderKind::TrailingStop {
6100 offset: Decimal::from(25),
6101 offset_type: TrailingOffsetType::BasisPoints,
6102 activation_price: None,
6103 }
6104 );
6105 assert_eq!(snapshot.0.price, None);
6106 }
6107 }
6108
6109 #[test]
6117 fn margin_ws_execution_report_trade_maps_to_trade_and_order_snapshot() {
6118 let frame = push(serde_json::json!({
6119 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6120 "x": "TRADE", "X": "PARTIALLY_FILLED", "i": 12_345_i64, "c": "cid-1",
6121 "t": 99_i64, "l": "0.5", "L": "48000", "z": "0.5",
6122 "p": "48000", "q": "2", "f": "GTC",
6123 "n": "0.001", "N": "BNB", "T": 1_700_000_000_000_i64,
6124 }));
6125 let mut buf = Vec::new();
6126 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6127 assert_eq!(
6128 buf.len(),
6129 2,
6130 "a fill must produce the execution and the order snapshot, got: {buf:?}"
6131 );
6132 assert_eq!(buf[0].exchange, ExchangeId::BinanceMargin);
6133 match &buf[0].kind {
6134 AccountEventKind::Trade(t) => {
6135 assert_eq!(t.instrument.name().as_str(), "BTCUSDT");
6136 assert_eq!(t.side, Side::Buy);
6137 assert_eq!(t.price, Decimal::from(48_000));
6138 assert_eq!(t.quantity, Decimal::new(5, 1));
6139 assert_eq!(t.fees.asset.name().as_str(), "BNB"); assert_eq!(t.fees.fees, Decimal::new(1, 3));
6141 assert_eq!(t.client_order_id, Some(ClientOrderId::new("cid-1")));
6142 }
6143 other => panic!("expected Trade, got {other:?}"),
6144 }
6145 assert_eq!(buf[1].exchange, ExchangeId::BinanceMargin);
6146 match &buf[1].kind {
6147 AccountEventKind::OrderSnapshot(snap) => {
6148 assert_eq!(
6149 snap.0.quantity,
6150 Decimal::from(2),
6151 "the order's total quantity"
6152 );
6153 match &snap.0.state {
6154 OrderState::Active(ActiveOrderState::Open(open)) => {
6155 assert_eq!(open.id.assigned().map(|id| id.0.as_str()), Some("12345"));
6156 assert_eq!(
6157 open.filled_quantity,
6158 Decimal::new(5, 1),
6159 "the snapshot must carry the order's cumulative filled quantity (z)"
6160 );
6161 assert_eq!(
6162 open.time_exchange.timestamp_millis(),
6163 1_700_000_000_000,
6164 "stamped with the execution's transaction time (T), which is what \
6165 gives the engine's staleness gate an ordering key"
6166 );
6167 }
6168 other => panic!("expected an Open order, got {other:?}"),
6169 }
6170 }
6171 other => panic!("expected OrderSnapshot, got {other:?}"),
6172 }
6173 }
6174
6175 #[test]
6179 fn margin_trade_for_an_order_no_longer_working_emits_no_snapshot() {
6180 for status in ["CANCELED", "EXPIRED", "REJECTED", "PENDING_CANCEL", "NEW"] {
6181 let frame = push(serde_json::json!({
6182 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6183 "x": "TRADE", "X": status, "i": 12_345_i64, "c": "cid-1",
6184 "t": 99_i64, "l": "0.5", "L": "48000", "z": "0.5",
6185 "p": "48000", "q": "2", "f": "GTC", "T": 1_700_000_000_000_i64,
6186 }));
6187 let mut buf = Vec::new();
6188 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6189 assert_eq!(
6190 buf.len(),
6191 1,
6192 "status {status} must produce the execution alone, got: {buf:?}"
6193 );
6194 assert!(
6195 matches!(buf[0].kind, AccountEventKind::Trade(_)),
6196 "the execution must still be reported for status {status}"
6197 );
6198 }
6199 }
6200
6201 #[test]
6202 fn margin_ws_new_report_maps_to_active_order_snapshot() {
6203 let frame = push(serde_json::json!({
6204 "e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "LIMIT",
6205 "x": "NEW", "X": "NEW", "i": 7_i64, "c": "cid-2",
6206 "p": "48000", "q": "1", "f": "GTC", "z": "0", "T": 1_700_000_000_000_i64,
6207 }));
6208 let mut buf = Vec::new();
6209 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6210 assert_eq!(buf.len(), 1);
6211 match &buf[0].kind {
6212 AccountEventKind::OrderSnapshot(snap) => {
6213 assert_eq!(snap.0.side, Side::Sell);
6214 assert_eq!(snap.0.price, Some(Decimal::from(48_000)));
6215 assert_eq!(snap.0.quantity, Decimal::from(1));
6216 assert!(
6217 matches!(snap.0.state, OrderState::Active(_)),
6218 "NEW should be an active (resting) order"
6219 );
6220 }
6221 other => panic!("expected OrderSnapshot, got {other:?}"),
6222 }
6223 }
6224
6225 #[test]
6226 fn margin_ws_canceled_report_maps_to_order_cancelled() {
6227 let frame = push(serde_json::json!({
6228 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6229 "x": "CANCELED", "X": "CANCELED", "i": 8_i64, "c": "cid-3",
6230 "z": "0", "T": 1_700_000_000_000_i64,
6231 }));
6232 let mut buf = Vec::new();
6233 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6234 assert_eq!(buf.len(), 1);
6235 match &buf[0].kind {
6236 AccountEventKind::OrderCancelled(resp) => assert!(resp.state.is_ok()),
6237 other => panic!("expected OrderCancelled, got {other:?}"),
6238 }
6239 }
6240
6241 #[test]
6244 fn margin_ws_trade_without_a_client_order_id_reports_none() {
6245 let frame = push(serde_json::json!({
6246 "e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "MARKET",
6247 "x": "TRADE", "X": "FILLED", "i": 12_345_i64,
6248 "t": 99_i64, "l": "1", "L": "48000", "z": "1",
6249 "q": "1", "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
6250 }));
6251 let mut buf = Vec::new();
6252 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6253 let AccountEventKind::Trade(trade) = &buf[0].kind else {
6254 panic!("expected Trade, got {:?}", buf[0].kind);
6255 };
6256 assert_eq!(trade.order_id.0.as_str(), "12345");
6257 assert_eq!(trade.client_order_id, None);
6258 }
6259
6260 #[test]
6261 fn margin_ws_execution_report_missing_symbol_is_dropped() {
6262 let frame = push(serde_json::json!({
6264 "e": "executionReport", "S": "BUY", "x": "TRADE", "i": 1_i64, "t": 1_i64,
6265 "l": "1", "L": "100", "T": 1_700_000_000_000_i64,
6266 }));
6267 let mut buf = Vec::new();
6268 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6269 assert!(buf.is_empty());
6270 }
6271
6272 #[test]
6273 fn margin_ws_outbound_account_position_maps_to_balance_stream_updates() {
6274 let frame = push(serde_json::json!({
6275 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6276 "B": [
6277 { "a": "USDT", "f": "100.0", "l": "5.0" },
6278 { "a": "BTC", "f": "0.5", "l": "0" },
6279 ],
6280 }));
6281 let mut buf = Vec::new();
6282 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6283 assert_eq!(buf.len(), 2);
6284 for ev in &buf {
6286 assert!(matches!(ev.kind, AccountEventKind::BalanceStreamUpdate(_)));
6287 }
6288 }
6289
6290 #[test]
6291 fn margin_ws_stream_terminated_signals_reconnect() {
6292 let frame = push(serde_json::json!({ "e": "eventStreamTerminated" }));
6293 let mut buf = Vec::new();
6294 assert!(
6295 convert_margin_user_data_events(&frame, &mut buf),
6296 "eventStreamTerminated must signal reconnect"
6297 );
6298 assert!(buf.is_empty());
6299 }
6300
6301 #[test]
6302 fn margin_ws_liability_change_is_observable_only() {
6303 let mut buf = Vec::new();
6306 assert!(!convert_margin_user_data_events(
6307 &push(documented_liability_change("USER_LIABILITY_CHANGE")),
6308 &mut buf
6309 ));
6310 assert!(buf.is_empty());
6311 }
6312
6313 fn only_notice(buf: &[UnindexedAccountEvent]) -> &AccountNotice<InstrumentNameExchange> {
6315 match buf {
6316 [
6317 UnindexedAccountEvent {
6318 exchange: ExchangeId::BinanceMargin,
6319 kind: AccountEventKind::Notice(notice),
6320 },
6321 ] => notice,
6322 other => panic!("expected one BinanceMargin notice, got {other:?}"),
6323 }
6324 }
6325
6326 #[test]
6327 fn margin_ws_margin_level_change_is_forwarded_as_a_notice() {
6328 let mut buf = Vec::new();
6329 assert!(!convert_margin_user_data_events(
6330 &push(documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE")),
6331 &mut buf
6332 ));
6333 assert_eq!(
6334 only_notice(&buf),
6335 &AccountNotice::new(
6336 NoticeKind::MarginCall,
6337 None,
6338 Utc.timestamp_millis_opt(1_573_200_697_110).unwrap(),
6339 Some(Decimal::new(125, 2)),
6340 SmolStr::new("MARGIN_CALL"),
6341 ),
6342 "cross names no instrument"
6343 );
6344 }
6345
6346 #[test]
6347 fn margin_level_statuses_map_to_notice_kinds() {
6348 for (status, kind) in [
6349 ("MARGIN_CALL", NoticeKind::MarginCall),
6350 ("PRE_LIQUIDATION", NoticeKind::LiquidationWarning),
6351 ("FORCE_LIQUIDATION", NoticeKind::Liquidation),
6352 ("NORMAL", NoticeKind::MarginRestored),
6353 ("EXCESSIVE", NoticeKind::MarginRestored),
6354 ("SOME_FUTURE_STATUS", NoticeKind::Other),
6355 ] {
6356 let mut buf = Vec::new();
6357 let frame = push(serde_json::json!({
6358 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "1.5",
6359 "s": status,
6360 }));
6361 convert_margin_user_data_events(&frame, &mut buf);
6362 let notice = only_notice(&buf);
6363 assert_eq!(notice.kind, kind, "{status}");
6364 assert_eq!(notice.status, status, "the raw status is kept");
6365 }
6366 }
6367
6368 #[test]
6369 fn a_margin_level_change_missing_parts_is_still_delivered_where_it_can_be() {
6370 let mut buf = Vec::new();
6372 convert_margin_user_data_events(
6373 &push(serde_json::json!({
6374 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "n/a",
6375 "s": "FORCE_LIQUIDATION",
6376 })),
6377 &mut buf,
6378 );
6379 let notice = only_notice(&buf);
6380 assert_eq!(notice.kind, NoticeKind::Liquidation);
6381 assert_eq!(notice.margin_level, None);
6382
6383 let before = Utc::now();
6385 let mut buf = Vec::new();
6386 convert_margin_user_data_events(
6387 &push(serde_json::json!({ "e": "MARGIN_LEVEL_STATUS_CHANGE", "s": "MARGIN_CALL" })),
6388 &mut buf,
6389 );
6390 assert!(only_notice(&buf).time_exchange >= before);
6391
6392 let mut buf = Vec::new();
6394 convert_margin_user_data_events(
6395 &push(serde_json::json!({
6396 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "1.5",
6397 })),
6398 &mut buf,
6399 );
6400 assert!(buf.is_empty());
6401 }
6402
6403 fn documented_liability_change(e: &str) -> serde_json::Value {
6405 serde_json::json!({
6406 "e": e, "E": 1_573_200_697_110_i64, "a": "BTC", "t": "BORROW",
6407 "p": "1.03453430", "i": "0",
6408 })
6409 }
6410
6411 fn documented_margin_level_change(e: &str) -> serde_json::Value {
6413 serde_json::json!({
6414 "e": e, "E": 1_573_200_697_110_i64, "l": "1.25", "s": "MARGIN_CALL",
6415 })
6416 }
6417
6418 #[test]
6419 fn margin_ws_margin_specific_events_match_their_documented_names() {
6420 for (e, frame, expected) in [
6425 (
6426 "USER_LIABILITY_CHANGE",
6427 documented_liability_change("USER_LIABILITY_CHANGE"),
6428 "USER_LIABILITY_CHANGE (observable",
6429 ),
6430 (
6431 "MARGIN_LEVEL_STATUS_CHANGE",
6432 documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE"),
6433 "MARGIN_LEVEL_STATUS_CHANGE received",
6434 ),
6435 (
6437 "userLiabilityChange",
6438 documented_liability_change("userLiabilityChange"),
6439 "USER_LIABILITY_CHANGE (observable",
6440 ),
6441 (
6442 "marginLevelStatusChange",
6443 documented_margin_level_change("marginLevelStatusChange"),
6444 "MARGIN_LEVEL_STATUS_CHANGE received",
6445 ),
6446 ] {
6447 let mut buf = Vec::new();
6448 let (terminated, messages) =
6449 log_messages(|| convert_margin_user_data_events(&push(frame), &mut buf));
6450 assert!(!terminated, "{e}");
6451 assert_eq!(
6453 buf.len(),
6454 usize::from(expected.starts_with("MARGIN")),
6455 "{e}"
6456 );
6457 assert!(
6458 messages.iter().any(|m| m.contains(expected)),
6459 "{e}: expected its own arm's log, got {messages:?}"
6460 );
6461 assert!(
6462 !messages.iter().any(|m| m.contains("undeserializable")
6463 || m.contains("unhandled")
6464 || m.contains("does not handle")),
6465 "{e}: {messages:?}"
6466 );
6467 }
6468 }
6469
6470 fn log_messages<T>(f: impl FnOnce() -> T) -> (T, Vec<String>) {
6474 use std::sync::{Arc, Mutex};
6475 use tracing_subscriber::layer::SubscriberExt;
6476
6477 #[derive(Default)]
6478 struct Messages(Arc<Mutex<Vec<String>>>);
6479
6480 struct Message<'a>(&'a mut String);
6481
6482 impl tracing::field::Visit for Message<'_> {
6483 fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
6484 if field.name() == "message" {
6485 *self.0 = format!("{value:?}");
6486 }
6487 }
6488 }
6489
6490 impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Messages {
6491 fn on_event(
6492 &self,
6493 event: &tracing::Event<'_>,
6494 _: tracing_subscriber::layer::Context<'_, S>,
6495 ) {
6496 let mut message = String::new();
6497 event.record(&mut Message(&mut message));
6498 self.0.lock().unwrap().push(message);
6499 }
6500 }
6501
6502 let layer = Messages::default();
6503 let messages = Arc::clone(&layer.0);
6504 let value =
6506 tracing::subscriber::with_default(tracing_subscriber::registry().with(layer), f);
6507 let messages = messages.lock().unwrap().clone();
6508 (value, messages)
6509 }
6510
6511 #[test]
6512 fn margin_ws_unhandled_event_types_are_ignored_without_teardown() {
6513 for (e, expected) in [
6516 ("listStatus", "ignoring listStatus"),
6517 ("someFutureEvent", "of a type this client does not handle"),
6519 ] {
6520 let mut buf = Vec::new();
6521 let frame = push(serde_json::json!({ "e": e, "E": 1_700_000_000_000_i64 }));
6522 let (terminated, messages) =
6523 log_messages(|| convert_margin_user_data_events(&frame, &mut buf));
6524 assert!(!terminated, "{e}");
6525 assert!(buf.is_empty(), "{e}");
6526 assert!(
6527 messages.iter().any(|m| m.contains(expected)),
6528 "{e}: {messages:?}"
6529 );
6530 }
6531 }
6532
6533 #[test]
6536 fn margin_trade_event_dedup_by_key() {
6537 let frame = push(serde_json::json!({
6540 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6541 "x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 4_242_i64,
6542 "l": "1", "L": "100", "z": "1", "p": "100", "q": "1", "f": "GTC",
6543 "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
6544 }));
6545 let mut buf = Vec::new();
6546 convert_margin_user_data_events(&frame, &mut buf);
6547 let event = buf.first().expect("a Trade event");
6550 assert!(
6551 matches!(event.kind, AccountEventKind::Trade(_)),
6552 "the execution is the first of the two events a fill produces"
6553 );
6554
6555 let cache = new_dedup_cache();
6556 let first = dedup_key_from_event(event).expect("Trade events have a dedup key");
6558 assert!(!is_duplicate(&cache, first), "first sighting is fresh");
6559 let second = dedup_key_from_event(event).expect("Trade events have a dedup key");
6560 assert!(
6561 is_duplicate(&cache, second),
6562 "second sighting is a duplicate"
6563 );
6564 }
6565
6566 fn drive_ws_pipeline(frames: &[&str]) -> Vec<UnindexedAccountEvent> {
6574 let cache = new_dedup_cache();
6575 let mut delivered = Vec::new();
6576 let mut buf = Vec::new();
6577 for frame in frames {
6578 let _ = convert_margin_user_data_events(frame, &mut buf);
6579 for ev in buf.drain(..) {
6580 if let Some(key) = dedup_key_from_event(&ev)
6581 && is_duplicate(&cache, key)
6582 {
6583 continue;
6584 }
6585 delivered.push(ev);
6586 }
6587 }
6588 delivered
6589 }
6590
6591 fn ws_new(cumulative_filled: &str) -> String {
6592 push(serde_json::json!({
6593 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6594 "x": "NEW", "X": "NEW", "i": 12_345_i64, "c": "client-1",
6595 "q": "2", "p": "100", "f": "GTC", "z": cumulative_filled,
6596 "T": 1_700_000_000_000_i64,
6597 }))
6598 }
6599
6600 fn ws_partial_fill(cumulative_filled: &str) -> String {
6601 push(serde_json::json!({
6602 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6603 "x": "TRADE", "X": "PARTIALLY_FILLED", "i": 12_345_i64, "c": "client-1",
6604 "q": "2", "p": "100", "f": "GTC", "z": cumulative_filled,
6605 "l": "1", "L": "100", "t": 555_i64, "n": "0.1", "N": "USDT",
6606 "T": 1_700_000_001_000_i64,
6607 }))
6608 }
6609
6610 fn open_filled_quantities(events: &[UnindexedAccountEvent]) -> Vec<Decimal> {
6611 events
6612 .iter()
6613 .filter_map(|ev| match &ev.kind {
6614 AccountEventKind::OrderSnapshot(snap) => match &snap.0.state {
6615 OrderState::Active(ActiveOrderState::Open(open)) => Some(open.filled_quantity),
6616 _ => None,
6617 },
6618 _ => None,
6619 })
6620 .collect()
6621 }
6622
6623 #[test]
6631 fn margin_ws_fill_snapshot_survives_the_dedup_gate() {
6632 let delivered = drive_ws_pipeline(&[&ws_new("0"), &ws_partial_fill("1")]);
6633
6634 assert_eq!(
6635 delivered.len(),
6636 3,
6637 "expected the ack snapshot, then the fill's execution and its snapshot, got: {delivered:?}"
6638 );
6639 assert_eq!(
6640 open_filled_quantities(&delivered),
6641 vec![Decimal::ZERO, Decimal::ONE],
6642 "the fill's snapshot must reach the consumer carrying the advanced filled quantity"
6643 );
6644 }
6645
6646 #[test]
6649 fn margin_ws_replayed_frame_is_still_deduplicated() {
6650 let delivered = drive_ws_pipeline(&[
6651 &ws_new("0"),
6652 &ws_partial_fill("1"),
6653 &ws_new("0"),
6654 &ws_partial_fill("1"),
6655 ]);
6656
6657 assert_eq!(
6658 delivered.len(),
6659 3,
6660 "re-delivering both frames must add nothing, got: {delivered:?}"
6661 );
6662 }
6663
6664 #[test]
6670 fn margin_ws_an_ack_and_its_fill_reporting_one_state_deliver_that_state_once() {
6671 let delivered = drive_ws_pipeline(&[&ws_new("1"), &ws_partial_fill("1")]);
6672
6673 assert_eq!(
6674 delivered.len(),
6675 2,
6676 "the ack snapshot and the execution, with the fill's duplicate snapshot collapsed, \
6677 got: {delivered:?}"
6678 );
6679 assert!(
6680 delivered
6681 .iter()
6682 .any(|ev| matches!(ev.kind, AccountEventKind::Trade(_))),
6683 "the execution must reach the consumer regardless"
6684 );
6685 assert_eq!(
6686 open_filled_quantities(&delivered),
6687 vec![Decimal::ONE],
6688 "one snapshot, carrying the state both reports agree on"
6689 );
6690 }
6691
6692 fn push_with_sub(subscription_id: i64, event: serde_json::Value) -> String {
6697 serde_json::json!({ "subscriptionId": subscription_id, "event": event }).to_string()
6698 }
6699
6700 fn isolated_handler(
6703 sub_map: Arc<Mutex<HashMap<i64, InstrumentNameExchange>>>,
6704 base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
6705 ) -> impl FnMut(OutboundAccountPosition, Option<i64>, &mut Vec<UnindexedAccountEvent>) {
6706 move |position, subscription_id, buf| {
6707 route_isolated_account_position(position, subscription_id, &sub_map, &base_quote, buf);
6708 }
6709 }
6710
6711 fn btcusdt() -> InstrumentNameExchange {
6712 InstrumentNameExchange::new("BTCUSDT")
6713 }
6714
6715 fn isolated_balances(buf: &[UnindexedAccountEvent]) -> Vec<(&str, &str, Decimal, Decimal)> {
6718 buf.iter()
6719 .map(|event| match &event.kind {
6720 AccountEventKind::InstrumentBalanceUpdate(ibu) => (
6721 ibu.instrument.name().as_str(),
6722 ibu.balance.asset.name().as_str(),
6723 ibu.balance.update.free,
6724 ibu.balance.update.locked,
6725 ),
6726 other => panic!("expected InstrumentBalanceUpdate, got {other:?}"),
6727 })
6728 .collect()
6729 }
6730
6731 fn token(expiration_time_ms: i64) -> UserListenToken {
6732 UserListenToken {
6733 token: "t".to_string(),
6734 expiration_time_ms,
6735 }
6736 }
6737
6738 #[test]
6739 fn listen_token_query_cross_vs_isolated() {
6740 assert!(build_listen_token_query(None).is_empty());
6742
6743 let q = build_listen_token_query(Some(&btcusdt()));
6745 assert_eq!(q.get("isIsolated").and_then(|v| v.as_str()), Some("TRUE"));
6746 assert_eq!(q.get("symbol").and_then(|v| v.as_str()), Some("BTCUSDT"));
6747 assert_eq!(q.len(), 2);
6748 }
6749
6750 #[test]
6751 fn earliest_token_expiry_picks_min() {
6752 let tokens = vec![
6753 (btcusdt(), token(3_000)),
6754 (InstrumentNameExchange::new("ETHUSDT"), token(1_000)),
6755 (InstrumentNameExchange::new("BNBUSDT"), token(2_000)),
6756 ];
6757 assert_eq!(earliest_token_expiry_ms(&tokens), 1_000);
6758 assert_eq!(earliest_token_expiry_ms(&[]), i64::MAX);
6760 }
6761
6762 #[test]
6763 fn base_quote_map_extracts_base_and_quote() {
6764 let entries = vec![
6765 iso_entry(
6766 "BTCUSDT",
6767 iso_base("BTC", "0", "0", "0", "0"),
6768 iso_quote("USDT", "0", "0", "0", "0"),
6769 None,
6770 None,
6771 None,
6772 ),
6773 iso_entry(
6774 "ETHBTC",
6775 iso_base("ETH", "0", "0", "0", "0"),
6776 iso_quote("BTC", "0", "0", "0", "0"),
6777 None,
6778 None,
6779 None,
6780 ),
6781 ];
6782 let map = build_base_quote_map(&entries);
6783 assert_eq!(
6784 map.get(&btcusdt())
6785 .map(|(b, q)| (b.name().as_str(), q.name().as_str())),
6786 Some(("BTC", "USDT"))
6787 );
6788 assert_eq!(
6791 map.get(&InstrumentNameExchange::new("ETHBTC"))
6792 .map(|(b, q)| (b.name().as_str(), q.name().as_str())),
6793 Some(("ETH", "BTC"))
6794 );
6795 }
6796
6797 #[test]
6798 fn isolated_margin_level_change_names_its_subscriptions_pair_or_none() {
6799 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6800 let notice_on = |subscription_id: i64| {
6801 let mut buf = Vec::new();
6802 let frame = push_with_sub(
6803 subscription_id,
6804 documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE"),
6805 );
6806 assert!(!convert_margin_user_data_events_with(
6807 &frame,
6808 &mut buf,
6809 &mut cross_account_position_handler,
6810 &mut |id| isolated_notice_instrument(id, &sub_map),
6811 ));
6812 only_notice(&buf).instrument.clone()
6813 };
6814
6815 assert_eq!(notice_on(7), Some(btcusdt()));
6816 assert_eq!(notice_on(99), None);
6818 assert_eq!(isolated_notice_instrument(None, &sub_map), None);
6820 }
6821
6822 #[test]
6823 fn the_two_streams_key_one_notice_alike() {
6824 use crate::client::binance::margin_risk::{RiskFrame, convert_risk_frame};
6826
6827 let change = documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE");
6828 let mut from_user_data = Vec::new();
6829 convert_margin_user_data_events(&push(change.clone()), &mut from_user_data);
6830 let mut from_risk = Vec::new();
6831 let risk_frame = serde_json::json!({ "stream": "listen-key", "data": change });
6832 assert_eq!(
6833 convert_risk_frame(&risk_frame.to_string(), "listen-key", &mut from_risk),
6834 RiskFrame::Continue
6835 );
6836
6837 let dedup = new_dedup_cache();
6838 let key = |events: &[UnindexedAccountEvent]| dedup_key_from_event(&events[0]).unwrap();
6839 assert!(!is_duplicate(&dedup, key(&from_user_data)));
6840 assert!(
6841 is_duplicate(&dedup, key(&from_risk)),
6842 "the risk data stream's copy is dropped"
6843 );
6844 }
6845
6846 #[test]
6847 fn a_notice_from_both_streams_is_delivered_once() {
6848 let dedup = new_dedup_cache();
6852 let notice = |time_ms: i64| {
6853 let mut buf = Vec::new();
6854 convert_margin_user_data_events(
6855 &push(serde_json::json!({
6856 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": time_ms, "l": "1.1",
6857 "s": "MARGIN_CALL",
6858 })),
6859 &mut buf,
6860 );
6861 buf.pop().unwrap()
6862 };
6863 let is_new = |event: &UnindexedAccountEvent| {
6864 !is_duplicate(&dedup, dedup_key_from_event(event).unwrap())
6865 };
6866
6867 assert!(is_new(¬ice(1_700_000_000_000)));
6868 assert!(!is_new(¬ice(1_700_000_000_000)), "second copy dropped");
6869 assert!(is_new(¬ice(1_700_000_060_000)), "a later notice is new");
6870 }
6871
6872 #[test]
6873 fn isolated_outbound_position_routes_to_instrument_balance_update() {
6874 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6875 let base_quote = Arc::new(HashMap::from([(
6876 btcusdt(),
6877 (
6878 AssetNameExchange::new("BTC"),
6879 AssetNameExchange::new("USDT"),
6880 ),
6881 )]));
6882 let mut handler = isolated_handler(sub_map, base_quote);
6883
6884 let frame = push_with_sub(
6886 7,
6887 serde_json::json!({
6888 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6889 "B": [
6890 { "a": "USDT", "f": "1000.0", "l": "50.0" },
6891 { "a": "BTC", "f": "0.5", "l": "0.1" },
6892 ],
6893 }),
6894 );
6895 let mut buf = Vec::new();
6896 assert!(!convert_margin_user_data_events_with(
6897 &frame,
6898 &mut buf,
6899 &mut handler,
6900 &mut cross_notice_instrument,
6901 ));
6902 assert_eq!(
6905 isolated_balances(&buf),
6906 vec![
6907 ("BTCUSDT", "USDT", Decimal::from(1000), Decimal::from(50)),
6908 ("BTCUSDT", "BTC", Decimal::new(5, 1), Decimal::new(1, 1)),
6909 ]
6910 );
6911 let time = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
6912 assert!(buf.iter().all(|event| matches!(
6913 &event.kind,
6914 AccountEventKind::InstrumentBalanceUpdate(ibu) if ibu.balance.time_exchange == time
6915 )));
6916 }
6917
6918 #[test]
6919 fn isolated_outbound_position_unknown_subscription_dropped() {
6920 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6922 let base_quote = Arc::new(HashMap::from([(
6923 btcusdt(),
6924 (
6925 AssetNameExchange::new("BTC"),
6926 AssetNameExchange::new("USDT"),
6927 ),
6928 )]));
6929 let mut handler = isolated_handler(sub_map, base_quote);
6930
6931 let frame = push_with_sub(
6932 99,
6933 serde_json::json!({
6934 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6935 "B": [{ "a": "BTC", "f": "1", "l": "0" }, { "a": "USDT", "f": "1", "l": "0" }],
6936 }),
6937 );
6938 let mut buf = Vec::new();
6939 assert!(!convert_margin_user_data_events_with(
6940 &frame,
6941 &mut buf,
6942 &mut handler,
6943 &mut cross_notice_instrument,
6944 ));
6945 assert!(buf.is_empty(), "unmapped subscriptionId frame is dropped");
6946 }
6947
6948 #[test]
6949 fn isolated_outbound_position_with_one_side_emits_that_side() {
6950 for (listed, free) in [("BTC", Decimal::new(5, 1)), ("USDT", Decimal::from(20))] {
6953 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6954 let base_quote = Arc::new(HashMap::from([(
6955 btcusdt(),
6956 (
6957 AssetNameExchange::new("BTC"),
6958 AssetNameExchange::new("USDT"),
6959 ),
6960 )]));
6961 let mut handler = isolated_handler(sub_map, base_quote);
6962
6963 let frame = push_with_sub(
6964 7,
6965 serde_json::json!({
6966 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6967 "B": [{ "a": listed, "f": free.to_string(), "l": "0" }],
6968 }),
6969 );
6970 let mut buf = Vec::new();
6971 assert!(!convert_margin_user_data_events_with(
6972 &frame,
6973 &mut buf,
6974 &mut handler,
6975 &mut cross_notice_instrument,
6976 ));
6977 assert_eq!(
6978 isolated_balances(&buf),
6979 vec![("BTCUSDT", listed, free, Decimal::ZERO)],
6980 "only {listed} listed"
6981 );
6982 }
6983 }
6984
6985 #[test]
6986 fn isolated_outbound_position_skips_assets_outside_the_pair() {
6987 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6990 let base_quote = Arc::new(HashMap::from([(
6991 btcusdt(),
6992 (
6993 AssetNameExchange::new("BTC"),
6994 AssetNameExchange::new("USDT"),
6995 ),
6996 )]));
6997 let mut handler = isolated_handler(sub_map, base_quote);
6998
6999 let mixed = push_with_sub(
7000 7,
7001 serde_json::json!({
7002 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
7003 "B": [{ "a": "BNB", "f": "3", "l": "0" }, { "a": "USDT", "f": "20", "l": "1" }],
7004 }),
7005 );
7006 let mut buf = Vec::new();
7007 assert!(!convert_margin_user_data_events_with(
7008 &mixed,
7009 &mut buf,
7010 &mut handler,
7011 &mut cross_notice_instrument,
7012 ));
7013 assert_eq!(
7014 isolated_balances(&buf),
7015 vec![("BTCUSDT", "USDT", Decimal::from(20), Decimal::ONE)]
7016 );
7017
7018 let stray_only = push_with_sub(
7019 7,
7020 serde_json::json!({
7021 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
7022 "B": [{ "a": "BNB", "f": "3", "l": "0" }],
7023 }),
7024 );
7025 let mut buf = Vec::new();
7026 assert!(!convert_margin_user_data_events_with(
7027 &stray_only,
7028 &mut buf,
7029 &mut handler,
7030 &mut cross_notice_instrument,
7031 ));
7032 assert!(buf.is_empty(), "no asset of the pair → nothing emitted");
7033 }
7034
7035 #[test]
7036 fn isolated_outbound_position_unparseable_entry_keeps_the_other_side() {
7037 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
7039 let base_quote = Arc::new(HashMap::from([(
7040 btcusdt(),
7041 (
7042 AssetNameExchange::new("BTC"),
7043 AssetNameExchange::new("USDT"),
7044 ),
7045 )]));
7046 let mut handler = isolated_handler(sub_map, base_quote);
7047
7048 let frame = push_with_sub(
7049 7,
7050 serde_json::json!({
7051 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
7052 "B": [{ "a": "BTC", "f": "x", "l": "0" }, { "a": "USDT", "f": "20", "l": "1" }],
7053 }),
7054 );
7055 let mut buf = Vec::new();
7056 assert!(!convert_margin_user_data_events_with(
7057 &frame,
7058 &mut buf,
7059 &mut handler,
7060 &mut cross_notice_instrument,
7061 ));
7062 assert_eq!(
7063 isolated_balances(&buf),
7064 vec![("BTCUSDT", "USDT", Decimal::from(20), Decimal::ONE)]
7065 );
7066 }
7067
7068 #[test]
7069 fn isolated_execution_report_routes_by_inner_symbol_independent_of_map() {
7070 let sub_map = Arc::new(Mutex::new(HashMap::<i64, InstrumentNameExchange>::new()));
7074 let base_quote = Arc::new(HashMap::new());
7075 let mut handler = isolated_handler(sub_map, base_quote);
7076
7077 let frame = push_with_sub(
7078 42,
7079 serde_json::json!({
7080 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
7081 "x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 5_i64,
7082 "l": "1", "L": "100", "z": "1", "p": "100", "q": "1", "f": "GTC",
7083 "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
7084 }),
7085 );
7086 let mut buf = Vec::new();
7087 assert!(!convert_margin_user_data_events_with(
7088 &frame,
7089 &mut buf,
7090 &mut handler,
7091 &mut cross_notice_instrument,
7092 ));
7093 assert_eq!(buf.len(), 2, "the execution and its order snapshot");
7094 match &buf[0].kind {
7095 AccountEventKind::Trade(t) => assert_eq!(t.instrument.name().as_str(), "BTCUSDT"),
7096 other => panic!("expected Trade, got {other:?}"),
7097 }
7098 match &buf[1].kind {
7099 AccountEventKind::OrderSnapshot(snap) => {
7100 assert_eq!(snap.0.key.instrument.name().as_str(), "BTCUSDT");
7101 }
7102 other => panic!("expected OrderSnapshot, got {other:?}"),
7103 }
7104 }
7105
7106 struct MarginMyTradesVenue {
7117 trades: Vec<(i64, i64, i64)>,
7119 end_exclusive: bool,
7120 }
7121
7122 impl wiremock::Respond for MarginMyTradesVenue {
7123 fn respond(&self, request: &wiremock::Request) -> wiremock::ResponseTemplate {
7124 const DAY_MS: i64 = 24 * 60 * 60 * 1000;
7125 let query: std::collections::HashMap<String, String> =
7126 request.url.query_pairs().into_owned().collect();
7127 let param = |name: &str| query.get(name).map(|v| v.parse::<i64>().unwrap());
7128 let limit = usize::try_from(param("limit").unwrap_or(500)).unwrap();
7129 let order = param("orderId");
7130
7131 let mut selected: Vec<&(i64, i64, i64)> = self
7132 .trades
7133 .iter()
7134 .filter(|(_, _, order_id)| order.is_none_or(|order| *order_id == order))
7135 .collect();
7136 if let Some(from_id) = param("fromId") {
7137 selected.retain(|(id, _, _)| *id >= from_id);
7138 } else if order.is_none() {
7139 let start = param("startTime").expect("a query by time sends startTime");
7140 let end = match param("endTime") {
7141 Some(end) if end - start >= DAY_MS => {
7142 return wiremock::ResponseTemplate::new(400).set_body_json(
7143 serde_json::json!({"code": -1127, "msg": "More than 24 hours between startTime and endTime."}),
7144 );
7145 }
7146 Some(end) if self.end_exclusive => end - 1,
7147 Some(end) => end,
7148 None => start + DAY_MS - 1,
7149 };
7150 selected.retain(|(_, time, _)| (start..=end).contains(time));
7151 }
7152
7153 let page: Vec<serde_json::Value> = selected
7154 .into_iter()
7155 .take(limit)
7156 .map(|(id, time, order_id)| {
7157 serde_json::json!({
7158 "id": id, "symbol": "BTCUSDT", "time": time, "orderId": order_id,
7159 "price": "50000", "qty": "0.01", "commission": "0",
7160 "commissionAsset": "USDT", "isBuyer": true, "isMaker": false,
7161 "isBestMatch": true, "isIsolated": true,
7162 })
7163 })
7164 .collect();
7165 wiremock::ResponseTemplate::new(200).set_body_json(page)
7166 }
7167 }
7168
7169 fn since(start: i64) -> MyTradesFrom {
7171 MyTradesFrom::Span {
7172 start,
7173 end: i64::MAX,
7174 }
7175 }
7176
7177 async fn read_margin_trades(
7180 venue: MarginMyTradesVenue,
7181 from: MyTradesFrom,
7182 ) -> (Vec<i64>, Vec<std::collections::HashMap<String, String>>) {
7183 let server = wiremock::MockServer::start().await;
7184 wiremock::Mock::given(wiremock::matchers::method("GET"))
7185 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7186 .respond_with(venue)
7187 .mount(&server)
7188 .await;
7189 let rest = Arc::new(MarginTradingRestApi::from_config(
7190 ConfigurationRestApi::builder()
7191 .api_key("key")
7192 .api_secret("secret")
7193 .base_path(server.uri())
7194 .build()
7195 .unwrap(),
7196 ));
7197
7198 let read = paginate_margin_my_trades(
7199 &rest,
7200 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7201 &InstrumentNameExchange::new("BTCUSDT"),
7202 from,
7203 true,
7204 RequestKind::Query,
7205 )
7206 .await
7207 .unwrap();
7208
7209 let requests = server
7210 .received_requests()
7211 .await
7212 .unwrap()
7213 .iter()
7214 .map(|request| request.url.query_pairs().into_owned().collect())
7215 .collect();
7216 (read.into_iter().map(|t| t.id.unwrap()).collect(), requests)
7217 }
7218
7219 #[tokio::test]
7222 async fn recovered_margin_fills_carry_their_orders_client_order_ids() {
7223 recovered_margin_fills_carry_client_order_ids(true).await;
7224 recovered_margin_fills_carry_client_order_ids(false).await;
7225 }
7226
7227 async fn recovered_margin_fills_carry_client_order_ids(is_isolated: bool) {
7228 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7229 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7230 let fill_time = (disconnect + chrono::Duration::minutes(1)).timestamp_millis();
7231 let server = wiremock::MockServer::start().await;
7232 wiremock::Mock::given(wiremock::matchers::method("GET"))
7233 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7234 .respond_with(MarginMyTradesVenue {
7235 trades: vec![(1, fill_time, 1), (2, fill_time, 2)],
7236 end_exclusive: false,
7237 })
7238 .mount(&server)
7239 .await;
7240 wiremock::Mock::given(wiremock::matchers::method("GET"))
7241 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7242 .and(wiremock::matchers::query_param("orderId", "1"))
7243 .and(wiremock::matchers::query_param(
7244 "isIsolated",
7245 if is_isolated { "TRUE" } else { "FALSE" },
7246 ))
7247 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
7248 serde_json::json!({"symbol": "BTCUSDT", "orderId": 1, "clientOrderId": "repay-1"}),
7249 ))
7250 .mount(&server)
7251 .await;
7252 wiremock::Mock::given(wiremock::matchers::method("GET"))
7253 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7254 .and(wiremock::matchers::query_param("orderId", "2"))
7255 .respond_with(
7256 wiremock::ResponseTemplate::new(400).set_body_json(
7257 serde_json::json!({"code": -2013, "msg": "Order does not exist."}),
7258 ),
7259 )
7260 .mount(&server)
7261 .await;
7262 let rest = Arc::new(MarginTradingRestApi::from_config(
7263 ConfigurationRestApi::builder()
7264 .api_key("key")
7265 .api_secret("secret")
7266 .base_path(server.uri())
7267 .build()
7268 .unwrap(),
7269 ));
7270 let (tx, mut rx) = mpsc::unbounded_channel();
7271 let mut unrecovered = UnrecoveredFills::default();
7272 unrecovered.open(
7273 &[InstrumentNameExchange::new("BTCUSDT")],
7274 disconnect,
7275 reconnect,
7276 );
7277
7278 recover_margin_fills(
7279 &rest,
7280 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7281 &mut unrecovered,
7282 &tx,
7283 &new_dedup_cache(),
7284 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7285 is_isolated,
7286 )
7287 .await;
7288
7289 let mut cids: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
7290 .filter_map(|event| match event.kind {
7291 AccountEventKind::Trade(trade) => Some((
7292 trade.id.0.to_string(),
7293 trade.client_order_id,
7294 trade.order_filled_quantity.is_some(),
7295 )),
7296 _ => None,
7297 })
7298 .collect();
7299 cids.sort();
7300 assert_eq!(
7302 cids,
7303 [
7304 ("1".to_string(), Some(ClientOrderId::new("repay-1")), true),
7305 ("2".to_string(), None, true),
7306 ],
7307 "is_isolated: {is_isolated}"
7308 );
7309 let order_reads = server
7310 .received_requests()
7311 .await
7312 .unwrap()
7313 .iter()
7314 .filter(|request| request.url.path() == "/sapi/v1/margin/order")
7315 .count();
7316 assert_eq!(order_reads, 2, "one read per order");
7317 }
7318
7319 fn venue(trades: Vec<(i64, i64)>) -> MarginMyTradesVenue {
7321 MarginMyTradesVenue {
7322 trades: trades
7323 .into_iter()
7324 .map(|(id, time)| (id, time, id))
7325 .collect(),
7326 end_exclusive: false,
7327 }
7328 }
7329
7330 const HOUR_MS: i64 = 60 * 60 * 1000;
7331
7332 #[tokio::test]
7335 async fn margin_trades_read_a_lone_fill_beyond_the_first_day() {
7336 let now = Utc::now().timestamp_millis();
7337 let (ids, _) = read_margin_trades(
7338 venue(vec![(7, now - 72 * HOUR_MS)]),
7339 since(now - 120 * HOUR_MS),
7340 )
7341 .await;
7342
7343 assert_eq!(ids, vec![7]);
7344 }
7345
7346 #[tokio::test]
7349 async fn margin_trades_read_fills_across_several_windows_once_each() {
7350 let now = Utc::now().timestamp_millis();
7351 let trades = vec![
7352 (1, now - 119 * HOUR_MS),
7353 (2, now - 72 * HOUR_MS),
7354 (3, now - HOUR_MS),
7355 ];
7356 let (ids, _) = read_margin_trades(venue(trades), since(now - 144 * HOUR_MS)).await;
7357
7358 assert_eq!(ids, vec![1, 2, 3]);
7359 }
7360
7361 #[tokio::test]
7364 async fn margin_trades_read_a_trade_on_a_window_boundary_once() {
7365 for end_exclusive in [false, true] {
7366 let now = Utc::now().timestamp_millis();
7367 let from = now - 48 * HOUR_MS;
7368 let venue = MarginMyTradesVenue {
7369 end_exclusive,
7370 ..venue(vec![(5, from + MARGIN_MY_TRADES_WINDOW_MS)])
7371 };
7372 let (ids, _) = read_margin_trades(venue, since(from)).await;
7373
7374 assert_eq!(ids, vec![5], "end_exclusive: {end_exclusive}");
7375 }
7376 }
7377
7378 #[tokio::test]
7380 async fn margin_trades_page_on_by_id_past_a_full_window() {
7381 let now = Utc::now().timestamp_millis();
7382 let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
7383 let trades = (1..=count)
7384 .map(|id| (id, now - 48 * HOUR_MS + id))
7385 .collect();
7386 let (ids, _) = read_margin_trades(venue(trades), since(now - 50 * HOUR_MS)).await;
7387
7388 assert_eq!(ids, (1..=count).collect::<Vec<_>>());
7389 }
7390
7391 #[tokio::test]
7394 async fn margin_trades_page_on_past_a_full_final_window() {
7395 let now = Utc::now().timestamp_millis();
7396 let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
7397 let trades = (1..=count).map(|id| (id, now - HOUR_MS + id)).collect();
7398 let (ids, requests) = read_margin_trades(venue(trades), since(now - 2 * HOUR_MS)).await;
7399
7400 assert_eq!(ids, (1..=count).collect::<Vec<_>>());
7401 assert_eq!(requests.len(), 2);
7402 }
7403
7404 #[tokio::test]
7407 async fn margin_trades_quiet_lookback_sends_one_request_per_window() {
7408 let now = Utc::now().timestamp_millis();
7409 let (ids, requests) =
7410 read_margin_trades(venue(Vec::new()), since(now - 72 * HOUR_MS)).await;
7411
7412 assert!(ids.is_empty());
7413 assert_eq!(requests.len(), 4);
7415 }
7416
7417 #[tokio::test]
7420 async fn margin_trades_read_a_short_gap_in_one_request() {
7421 let now = Utc::now().timestamp_millis();
7422 let (ids, requests) =
7423 read_margin_trades(venue(vec![(9, now - HOUR_MS)]), since(now - 2 * HOUR_MS)).await;
7424
7425 assert_eq!(ids, vec![9]);
7426 assert_eq!(requests.len(), 1);
7427 }
7428
7429 #[tokio::test]
7431 async fn margin_trades_from_the_future_send_no_request() {
7432 let now = Utc::now().timestamp_millis();
7433 let (ids, requests) = read_margin_trades(venue(Vec::new()), since(now + HOUR_MS)).await;
7434
7435 assert!(ids.is_empty());
7436 assert!(requests.is_empty());
7437 }
7438
7439 #[tokio::test]
7442 async fn margin_trades_by_order_page_by_id_without_time_bounds() {
7443 let now = Utc::now().timestamp_millis();
7444 let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
7445 let trades = (1..=2 * count)
7447 .map(|id| {
7448 (
7449 id,
7450 now - 48 * HOUR_MS + id,
7451 if id % 2 == 0 { 42 } else { 7 },
7452 )
7453 })
7454 .collect();
7455 let venue = MarginMyTradesVenue {
7456 trades,
7457 end_exclusive: false,
7458 };
7459 let (ids, requests) = read_margin_trades(venue, MyTradesFrom::Order(42)).await;
7460
7461 assert_eq!(ids, (1..=count).map(|n| 2 * n).collect::<Vec<_>>());
7462 assert_eq!(requests.len(), 2);
7463 assert!(
7464 requests.iter().all(
7465 |query| query.get("orderId").map(String::as_str) == Some("42")
7466 && !query.contains_key("startTime")
7467 && !query.contains_key("endTime")
7468 )
7469 );
7470 }
7471
7472 async fn margin_trades_error(response: wiremock::ResponseTemplate) -> UnindexedClientError {
7475 let server = wiremock::MockServer::start().await;
7476 wiremock::Mock::given(wiremock::matchers::method("GET"))
7477 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7478 .respond_with(response)
7479 .mount(&server)
7480 .await;
7481 let rest = Arc::new(MarginTradingRestApi::from_config(
7482 ConfigurationRestApi::builder()
7483 .api_key("key")
7484 .api_secret("secret")
7485 .base_path(server.uri())
7486 .build()
7487 .unwrap(),
7488 ));
7489
7490 paginate_margin_my_trades(
7491 &rest,
7492 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7493 &InstrumentNameExchange::new("BTCUSDT"),
7494 since(Utc::now().timestamp_millis() - HOUR_MS),
7495 true,
7496 RequestKind::Query,
7497 )
7498 .await
7499 .unwrap_err()
7500 }
7501
7502 #[tokio::test]
7505 async fn margin_trades_report_a_venue_rejection_as_a_non_transient_request_rejection() {
7506 let err = margin_trades_error(wiremock::ResponseTemplate::new(400).set_body_json(
7507 serde_json::json!({"code": -1127, "msg": "More than 24 hours between startTime and endTime."}),
7508 ))
7509 .await;
7510
7511 let UnindexedClientError::Api(ApiError::RequestRejected(msg)) = &err else {
7512 panic!("expected RequestRejected, got {err:?}");
7513 };
7514 assert!(msg.contains("-1127"), "code missing from {msg:?}");
7515 assert!(!err.is_transient());
7516 }
7517
7518 #[tokio::test]
7520 async fn margin_trades_report_an_undecodable_response_as_internal() {
7521 let err = margin_trades_error(
7522 wiremock::ResponseTemplate::new(200)
7523 .set_body_json(serde_json::json!({"unexpected": 1})),
7524 )
7525 .await;
7526
7527 assert!(
7528 matches!(err, UnindexedClientError::Internal(_)),
7529 "got {err:?}"
7530 );
7531 assert!(!err.is_transient());
7532 }
7533
7534 #[tokio::test]
7537 async fn sapi_used_ip_weight_near_the_limit_pauses_queries() {
7538 for (used, pauses) in [("10799", false), ("10800", true), ("not a number", false)] {
7539 let server = wiremock::MockServer::start().await;
7540 wiremock::Mock::given(wiremock::matchers::method("GET"))
7541 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
7542 .respond_with(
7543 wiremock::ResponseTemplate::new(200)
7544 .insert_header("x-sapi-used-ip-weight-1m", used)
7545 .set_body_json(serde_json::json!([])),
7546 )
7547 .mount(&server)
7548 .await;
7549 let rest = Arc::new(MarginTradingRestApi::from_config(
7550 ConfigurationRestApi::builder()
7551 .api_key("key")
7552 .api_secret("secret")
7553 .base_path(server.uri())
7554 .build()
7555 .unwrap(),
7556 ));
7557 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7558
7559 fetch_margin_all_open_orders(rest, Arc::clone(&tracker), false)
7560 .await
7561 .unwrap();
7562
7563 let query_waits = tokio::time::timeout(
7564 Duration::from_millis(50),
7565 tracker.wait_if_blocked(RequestKind::Query),
7566 )
7567 .await
7568 .is_err();
7569 assert_eq!(query_waits, pauses, "used IP weight {used}");
7570 tokio::time::timeout(
7571 Duration::from_millis(50),
7572 tracker.wait_if_blocked(RequestKind::Order),
7573 )
7574 .await
7575 .expect("orders never wait on the pause")
7576 .expect("no cooldown too long to sleep through");
7577 }
7578 }
7579
7580 #[tokio::test]
7582 async fn margin_fill_recovery_does_not_wait_for_the_weight_pause() {
7583 let server = wiremock::MockServer::start().await;
7584 wiremock::Mock::given(wiremock::matchers::method("GET"))
7585 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7586 .respond_with(
7588 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([{
7589 "symbol": "BTCUSDT",
7590 "id": 1,
7591 "orderId": 7,
7592 "price": "100",
7593 "qty": "1",
7594 "commission": "0",
7595 "commissionAsset": "USDT",
7596 "time": Utc::now().timestamp_millis(),
7597 "isBuyer": true,
7598 "isMaker": false,
7599 }])),
7600 )
7601 .expect(2)
7602 .mount(&server)
7603 .await;
7604 let rest = Arc::new(MarginTradingRestApi::from_config(
7605 ConfigurationRestApi::builder()
7606 .api_key("key")
7607 .api_secret("secret")
7608 .base_path(server.uri())
7609 .build()
7610 .unwrap(),
7611 ));
7612 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7613 tracker.throttle(Duration::from_secs(60));
7614 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
7615 let mut unrecovered = UnrecoveredFills::default();
7616 unrecovered.open(
7617 &[InstrumentNameExchange::new("BTCUSDT")],
7618 Utc::now() - chrono::Duration::minutes(1),
7619 Utc::now(),
7620 );
7621
7622 tokio::time::timeout(
7624 Duration::from_millis(800),
7625 recover_margin_fills(
7626 &rest,
7627 &tracker,
7628 &mut unrecovered,
7629 &tx,
7630 &new_dedup_cache(),
7631 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7632 false,
7633 ),
7634 )
7635 .await
7636 .expect("recovery does not wait for the pause");
7637 }
7638
7639 #[tokio::test]
7642 async fn margin_trades_read_over_a_span_stop_at_its_end() {
7643 let now = Utc::now().timestamp_millis();
7644 let trades = vec![
7645 (1, now - 50 * HOUR_MS),
7646 (2, now - 49 * HOUR_MS),
7647 (3, now - 10 * HOUR_MS),
7648 ];
7649 let (ids, _) = read_margin_trades(
7650 venue(trades),
7651 MyTradesFrom::Span {
7652 start: now - 60 * HOUR_MS,
7653 end: now - 20 * HOUR_MS,
7654 },
7655 )
7656 .await;
7657
7658 assert_eq!(ids, vec![1, 2]);
7659 }
7660
7661 #[tokio::test]
7664 async fn a_margin_gap_whose_every_read_fails_is_reported_when_given_up() {
7665 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7666 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7667 let server = wiremock::MockServer::start().await;
7668 wiremock::Mock::given(wiremock::matchers::method("GET"))
7669 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7670 .respond_with(
7671 wiremock::ResponseTemplate::new(400)
7672 .set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
7673 )
7674 .mount(&server)
7675 .await;
7676 let rest = Arc::new(MarginTradingRestApi::from_config(
7677 ConfigurationRestApi::builder()
7678 .api_key("key")
7679 .api_secret("secret")
7680 .base_path(server.uri())
7681 .build()
7682 .expect("valid config"),
7683 ));
7684 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7685 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
7686 let dedup = new_dedup_cache();
7687 let btc = InstrumentNameExchange::new("BTCUSDT");
7688 let mut unrecovered = UnrecoveredFills::default();
7689 unrecovered.open(std::slice::from_ref(&btc), disconnect, reconnect);
7690
7691 for _ in 0..=crate::client::order_recovery::MAX_GAP_RETRIES {
7692 recover_margin_fills(
7693 &rest,
7694 &tracker,
7695 &mut unrecovered,
7696 &tx,
7697 &dedup,
7698 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7699 false,
7700 )
7701 .await;
7702 unrecovered.make_due();
7703 }
7704
7705 assert!(unrecovered.is_empty(), "the gap is given up");
7706 let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7707 let [event] = forwarded.as_slice() else {
7708 panic!("only the give-up is sent: {forwarded:?}");
7709 };
7710 let AccountEventKind::FillRecoveryGaveUp(gave_up) = &event.kind else {
7711 panic!("expected FillRecoveryGaveUp, got {event:?}");
7712 };
7713 assert_eq!(event.exchange, ExchangeId::BinanceMargin);
7714 assert_eq!(
7715 gave_up.scope,
7716 crate::FillRecoveryScope::Instruments(vec![btc])
7717 );
7718 assert_eq!(
7719 gave_up.start.timestamp_millis(),
7720 disconnect.timestamp_millis()
7721 );
7722 assert_eq!(
7723 gave_up.attempts,
7724 crate::client::order_recovery::MAX_GAP_RETRIES + 1
7725 );
7726 assert!(
7727 matches!(&gave_up.reason, FillRecoveryFailure::Request(reason) if reason.contains("rejected")),
7728 "the last read's error: {:?}",
7729 gave_up.reason
7730 );
7731 }
7732
7733 #[tokio::test]
7736 async fn a_failed_margin_recovery_keeps_its_closed_gap_for_the_retry() {
7737 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7738 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7739 let server = wiremock::MockServer::start().await;
7740 wiremock::Mock::given(wiremock::matchers::method("GET"))
7741 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7742 .respond_with(
7743 wiremock::ResponseTemplate::new(400)
7744 .set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
7745 )
7746 .up_to_n_times(1)
7747 .with_priority(1)
7748 .mount(&server)
7749 .await;
7750 wiremock::Mock::given(wiremock::matchers::method("GET"))
7751 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7752 .respond_with(venue(vec![
7753 (
7754 1,
7755 (disconnect + chrono::Duration::minutes(1)).timestamp_millis(),
7756 ),
7757 (
7758 2,
7759 (Utc::now() - chrono::Duration::minutes(1)).timestamp_millis(),
7760 ),
7761 ]))
7762 .mount(&server)
7763 .await;
7764 let rest = Arc::new(MarginTradingRestApi::from_config(
7765 ConfigurationRestApi::builder()
7766 .api_key("key")
7767 .api_secret("secret")
7768 .base_path(server.uri())
7769 .build()
7770 .unwrap(),
7771 ));
7772 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7773 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
7774 let dedup = new_dedup_cache();
7775 let mut unrecovered = UnrecoveredFills::default();
7776 unrecovered.open(
7777 &[InstrumentNameExchange::new("BTCUSDT")],
7778 disconnect,
7779 reconnect,
7780 );
7781
7782 recover_margin_fills(
7783 &rest,
7784 &tracker,
7785 &mut unrecovered,
7786 &tx,
7787 &dedup,
7788 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7789 false,
7790 )
7791 .await;
7792 assert!(!unrecovered.is_empty(), "the failed gap is kept");
7793 assert!(rx.try_recv().is_err(), "nothing forwarded");
7794
7795 unrecovered.make_due();
7796 recover_margin_fills(
7797 &rest,
7798 &tracker,
7799 &mut unrecovered,
7800 &tx,
7801 &dedup,
7802 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7803 false,
7804 )
7805 .await;
7806 assert!(unrecovered.is_empty(), "the retry recovered it");
7807 let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7808 assert_eq!(
7809 forwarded.len(),
7810 1,
7811 "only the fill inside the gap: {forwarded:?}"
7812 );
7813 }
7814
7815 #[tokio::test]
7818 async fn a_timed_out_margin_recovery_charges_only_the_gaps_it_started() {
7819 let server = wiremock::MockServer::start().await;
7820 wiremock::Mock::given(wiremock::matchers::method("GET"))
7821 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7822 .respond_with(
7823 wiremock::ResponseTemplate::new(200)
7824 .set_body_json(serde_json::json!([]))
7825 .set_delay(Duration::from_secs(3_600)),
7826 )
7827 .mount(&server)
7828 .await;
7829 let rest = Arc::new(MarginTradingRestApi::from_config(
7830 ConfigurationRestApi::builder()
7831 .api_key("key")
7832 .api_secret("secret")
7833 .base_path(server.uri())
7834 .timeout(3_600_000_u64)
7835 .retries(0_u32)
7836 .build()
7837 .unwrap(),
7838 ));
7839 let instruments: Vec<_> = (0..9)
7840 .map(|i| InstrumentNameExchange::new(format!("SYM{i}USDT")))
7841 .collect();
7842 let mut unrecovered = UnrecoveredFills::default();
7843 unrecovered.open(
7844 &instruments,
7845 Utc::now() - chrono::Duration::minutes(10),
7846 Utc::now(),
7847 );
7848 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
7849 tokio::time::pause();
7851
7852 recover_margin_fills(
7853 &rest,
7854 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7855 &mut unrecovered,
7856 &tx,
7857 &new_dedup_cache(),
7858 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7859 false,
7860 )
7861 .await;
7862
7863 let now = tokio::time::Instant::now();
7864 assert_eq!(
7865 unrecovered.due(now).len(),
7866 1,
7867 "the gap never started stays due"
7868 );
7869 assert_eq!(
7870 unrecovered
7871 .due(now + Duration::from_secs(crate::client::order_recovery::GAP_RETRY_BASE_SECS))
7872 .len(),
7873 9,
7874 "the eight started wait for their first retry"
7875 );
7876 }
7877
7878 fn margin_order_row(cid: &str, status: &str) -> serde_json::Value {
7884 serde_json::json!({
7885 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid, "price": "100",
7886 "origQty": "2", "executedQty": "1", "cummulativeQuoteQty": "105", "status": status,
7887 "timeInForce": "GTC", "type": "LIMIT", "side": "BUY", "isIsolated": false,
7888 "time": 1_700_000_000_000_i64, "updateTime": 1_700_000_060_000_i64,
7889 })
7890 }
7891
7892 fn margin_key(instrument: &str, cid: &str) -> UnindexedOrderKey {
7893 OrderKey::new(
7894 ExchangeId::BinanceMargin,
7895 InstrumentNameExchange::new(instrument),
7896 StrategyId::new("strategy"),
7897 ClientOrderId::new(cid),
7898 )
7899 }
7900
7901 #[test]
7902 fn a_margin_order_row_says_how_the_order_ended() {
7903 use crate::order::state::InactiveOrderState;
7904 use binance_sdk::margin_trading::rest_api::QueryMarginAccountsOrderResponse;
7905 let key = margin_key("BTCUSDT", "a");
7906 let ended = |row: serde_json::Value| {
7907 let row: QueryMarginAccountsOrderResponse = serde_json::from_value(row).unwrap();
7908 convert_ended_order(&row, ExchangeId::BinanceMargin, &key, None)
7909 .map(|order| order.state)
7910 };
7911
7912 let mut filled = margin_order_row("a", "FILLED");
7913 filled["executedQty"] = "2".into();
7914 filled["cummulativeQuoteQty"] = "210".into();
7915 let Some(InactiveOrderState::FullyFilled(filled)) = ended(filled) else {
7916 panic!("FILLED is fully filled");
7917 };
7918 assert_eq!(filled.avg_price, Some(Decimal::from(105)));
7919 assert!(matches!(
7920 ended(margin_order_row("a", "CANCELED")),
7921 Some(InactiveOrderState::Cancelled(cancelled)) if cancelled.filled_quantity == Some(Decimal::ONE)
7922 ));
7923 assert_eq!(ended(margin_order_row("a", "NEW")), None);
7924
7925 let mut unknown_quote = margin_order_row("a", "FILLED");
7927 unknown_quote["executedQty"] = "2".into();
7928 unknown_quote["cummulativeQuoteQty"] = "-1".into();
7929 let Some(InactiveOrderState::FullyFilled(filled)) = ended(unknown_quote) else {
7930 panic!("still fully filled");
7931 };
7932 assert_eq!(filled.avg_price, None, "no negative average price");
7933 }
7934
7935 fn margin_rest_at(server: &wiremock::MockServer) -> Arc<RestApi> {
7936 Arc::new(MarginTradingRestApi::from_config(
7937 ConfigurationRestApi::builder()
7938 .api_key("key")
7939 .api_secret("secret")
7940 .base_path(server.uri())
7941 .build()
7942 .unwrap(),
7943 ))
7944 }
7945
7946 pub(super) fn margin_client_at(
7948 server: &wiremock::MockServer,
7949 is_isolated: bool,
7950 ) -> BinanceMargin {
7951 let config = if is_isolated {
7952 BinanceMarginConfig::isolated(
7953 "key".into(),
7954 "secret".into(),
7955 vec![InstrumentNameExchange::new("BTCUSDT")],
7956 )
7957 } else {
7958 BinanceMarginConfig::cross_margin("key".into(), "secret".into())
7959 };
7960 let mut client = <BinanceMargin as ExecutionClient>::new(config);
7961 client.rest = margin_rest_at(server);
7962 client
7963 }
7964
7965 async fn mount_margin_order(
7967 server: &wiremock::MockServer,
7968 cid: &str,
7969 response: wiremock::ResponseTemplate,
7970 ) {
7971 wiremock::Mock::given(wiremock::matchers::method("GET"))
7972 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7973 .and(wiremock::matchers::query_param("origClientOrderId", cid))
7974 .respond_with(response)
7975 .mount(server)
7976 .await;
7977 }
7978
7979 fn margin_venue_error(code: i64) -> wiremock::ResponseTemplate {
7980 wiremock::ResponseTemplate::new(400)
7981 .set_body_json(serde_json::json!({"code": code, "msg": "refused"}))
7982 }
7983
7984 fn margin_row_response(cid: &str, status: &str) -> wiremock::ResponseTemplate {
7985 wiremock::ResponseTemplate::new(200).set_body_json(margin_order_row(cid, status))
7986 }
7987
7988 async fn margin_requests(server: &wiremock::MockServer) -> Vec<(String, Option<String>)> {
7990 server
7991 .received_requests()
7992 .await
7993 .unwrap()
7994 .iter()
7995 .map(|request| {
7996 let isolated = request
7997 .url
7998 .query_pairs()
7999 .find(|(key, _)| key == "isIsolated")
8000 .map(|(_, value)| value.into_owned());
8001 (request.url.path().to_owned(), isolated)
8002 })
8003 .collect()
8004 }
8005
8006 #[tokio::test]
8007 async fn margin_fetch_ended_orders_asks_the_configured_account_and_omits_the_rest() {
8008 use crate::order::state::InactiveOrderState;
8009 for (is_isolated, flag) in [(false, "FALSE"), (true, "TRUE")] {
8010 let server = wiremock::MockServer::start().await;
8011 mount_margin_order(&server, "filled", margin_row_response("filled", "FILLED")).await;
8012 mount_margin_order(&server, "live", margin_row_response("live", "NEW")).await;
8013 mount_margin_order(&server, "unknown", margin_venue_error(-2013)).await;
8014 mount_margin_order(&server, "no-symbol", margin_venue_error(-1121)).await;
8015 let client = margin_client_at(&server, is_isolated);
8016 let keys = ["filled", "live", "unknown"].map(|cid| margin_key("BTCUSDT", cid));
8017
8018 let ended = client
8019 .fetch_ended_orders(
8020 &[keys.to_vec(), vec![margin_key("NOSUCH", "no-symbol")]].concat(),
8021 )
8022 .await
8023 .unwrap();
8024
8025 let [order] = ended.as_slice() else {
8026 panic!("one order ended: {ended:?}");
8027 };
8028 assert_eq!(order.key, keys[0], "the strategy asked survives");
8029 assert!(matches!(order.state, InactiveOrderState::FullyFilled(_)));
8030 let requests = margin_requests(&server).await;
8031 assert_eq!(requests.len(), 4);
8032 for (path, isolated) in requests {
8033 assert_eq!(path, "/sapi/v1/margin/order");
8034 assert_eq!(
8035 isolated.as_deref(),
8036 Some(flag),
8037 "is_isolated = {is_isolated}"
8038 );
8039 }
8040 }
8041 }
8042
8043 #[tokio::test]
8044 async fn margin_fetch_ended_orders_fails_when_any_lookup_fails() {
8045 let server = wiremock::MockServer::start().await;
8046 mount_margin_order(
8047 &server,
8048 "cancelled",
8049 margin_row_response("cancelled", "CANCELED"),
8050 )
8051 .await;
8052 mount_margin_order(&server, "refused", margin_venue_error(-1100)).await;
8053 let client = margin_client_at(&server, false);
8054
8055 let result = client
8056 .fetch_ended_orders(&[
8057 margin_key("BTCUSDT", "cancelled"),
8058 margin_key("BTCUSDT", "refused"),
8059 ])
8060 .await;
8061
8062 assert!(
8063 matches!(
8064 result,
8065 Err(UnindexedClientError::Api(ApiError::RequestRejected(_)))
8066 ),
8067 "{result:?}"
8068 );
8069 }
8070
8071 fn held_margin_orders() -> SharedKnownLiveOrders {
8073 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
8074 for (n, cid) in ["live", "gone", "unknown"].into_iter().enumerate() {
8075 known.lock().live(
8076 &margin_key("BTCUSDT", cid),
8077 Decimal::TWO,
8078 OrderKind::Limit,
8079 &Open::new(
8080 VenueOrderId::Assigned(OrderId::new(n.to_string())),
8081 Utc::now(),
8082 Decimal::ZERO,
8083 ),
8084 );
8085 }
8086 known
8087 }
8088
8089 #[tokio::test]
8094 async fn an_ended_margin_trailing_stop_is_reported_with_the_kind_it_was_held_with() {
8095 let server = wiremock::MockServer::start().await;
8096 wiremock::Mock::given(wiremock::matchers::method("GET"))
8097 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
8098 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
8099 .mount(&server)
8100 .await;
8101 let mut row = margin_order_row("trailing", "CANCELED");
8102 row["type"] = "STOP_LOSS".into();
8103 row["price"] = "0.00000000".into();
8104 row["stopPrice"] = "0.00000000".into();
8105 mount_margin_order(
8106 &server,
8107 "trailing",
8108 wiremock::ResponseTemplate::new(200).set_body_json(row),
8109 )
8110 .await;
8111 let trailing_stop = OrderKind::TrailingStop {
8112 offset: Decimal::from(25),
8113 offset_type: TrailingOffsetType::BasisPoints,
8114 activation_price: None,
8115 };
8116 let hold = |known: &SharedKnownLiveOrders| {
8117 known.lock().live(
8118 &margin_key("BTCUSDT", "trailing"),
8119 Decimal::TWO,
8120 trailing_stop,
8121 &Open::new(
8122 VenueOrderId::Assigned(OrderId::new("7")),
8123 Utc::now(),
8124 Decimal::ZERO,
8125 ),
8126 );
8127 };
8128
8129 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
8130 hold(&known);
8131 let mut unchecked = UncheckedOrders::default();
8132 unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
8133 let (tx, mut rx) = mpsc::unbounded_channel();
8134 recover_margin_ended_orders(
8135 &margin_rest_at(&server),
8136 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
8137 &known,
8138 &mut unchecked,
8139 &UnrecoveredFills::default(),
8140 &tx,
8141 false,
8142 )
8143 .await;
8144 let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
8145 let [event] = sent.as_slice() else {
8146 panic!("the trailing stop ended: {sent:?}");
8147 };
8148 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
8149 panic!("an order snapshot: {event:?}");
8150 };
8151 assert_eq!(snapshot.0.kind, trailing_stop);
8152 assert!(matches!(
8153 snapshot.0.state,
8154 OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(_))
8155 ));
8156 assert!(!known.lock().contains(&ClientOrderId::new("trailing")));
8157
8158 let client = margin_client_at(&server, false);
8159 hold(&client.known_live);
8160 let ended = client
8161 .fetch_ended_orders(&[margin_key("BTCUSDT", "trailing")])
8162 .await
8163 .expect("lookup");
8164 let [order] = ended.as_slice() else {
8165 panic!("the trailing stop ended: {ended:?}");
8166 };
8167 assert_eq!(order.kind, trailing_stop);
8168 }
8169
8170 #[tokio::test]
8171 async fn a_margin_reconnect_reports_how_an_unlisted_order_ended_in_the_configured_account() {
8172 for (is_isolated, flag) in [(false, "FALSE"), (true, "TRUE")] {
8173 let server = wiremock::MockServer::start().await;
8174 wiremock::Mock::given(wiremock::matchers::method("GET"))
8175 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
8176 .and(wiremock::matchers::query_param("symbol", "BTCUSDT"))
8177 .respond_with(
8178 wiremock::ResponseTemplate::new(200)
8179 .set_body_json(serde_json::json!([margin_order_row("live", "NEW")])),
8180 )
8181 .mount(&server)
8182 .await;
8183 mount_margin_order(&server, "gone", margin_row_response("gone", "CANCELED")).await;
8184 mount_margin_order(&server, "unknown", margin_venue_error(-2013)).await;
8185 let known = held_margin_orders();
8186 let mut unchecked = UncheckedOrders::default();
8187 unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
8188 let (tx, mut rx) = mpsc::unbounded_channel();
8189
8190 recover_margin_ended_orders(
8191 &margin_rest_at(&server),
8192 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
8193 &known,
8194 &mut unchecked,
8195 &UnrecoveredFills::default(),
8196 &tx,
8197 is_isolated,
8198 )
8199 .await;
8200
8201 let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
8202 let [event] = sent.as_slice() else {
8203 panic!("one order ended: {sent:?}");
8204 };
8205 assert_eq!(event.exchange, ExchangeId::BinanceMargin);
8206 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
8207 panic!("an order snapshot: {event:?}");
8208 };
8209 assert_eq!(snapshot.0.key.cid, ClientOrderId::new("gone"));
8210 assert_eq!(snapshot.0.key.strategy, StrategyId::unknown());
8211 assert!(matches!(
8212 snapshot.0.state,
8213 OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(_))
8214 ));
8215 let mut requests = margin_requests(&server).await;
8216 requests.sort();
8217 let expected = |path: &str| (path.to_owned(), Some(flag.to_owned()));
8218 assert_eq!(
8219 requests,
8220 [
8221 expected("/sapi/v1/margin/openOrders"),
8222 expected("/sapi/v1/margin/order"),
8223 expected("/sapi/v1/margin/order"),
8224 ],
8225 "one listing, and a lookup of each unlisted order only"
8226 );
8227 let known = known.lock();
8228 assert!(known.contains(&ClientOrderId::new("live")));
8229 assert!(!known.contains(&ClientOrderId::new("gone")));
8230 assert!(
8231 !known.contains(&ClientOrderId::new("unknown")),
8232 "asking again cannot help"
8233 );
8234 assert!(unchecked.is_empty());
8235 }
8236 }
8237
8238 fn margin_open_request<'a>(
8239 instrument: &'a InstrumentNameExchange,
8240 cid: &str,
8241 ) -> OrderRequestOpen<ExchangeId, &'a InstrumentNameExchange> {
8242 OrderRequestOpen {
8243 key: OrderKey::new(
8244 ExchangeId::BinanceMargin,
8245 instrument,
8246 StrategyId::new("strategy"),
8247 ClientOrderId::new(cid),
8248 ),
8249 state: crate::order::request::RequestOpen {
8250 side: Side::Buy,
8251 price: Some(Decimal::from(100)),
8252 quantity: Decimal::TWO,
8253 kind: OrderKind::Limit,
8254 time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
8255 position_id: None,
8256 reduce_only: false,
8257 market: None,
8258 },
8259 }
8260 }
8261
8262 #[tokio::test]
8265 async fn a_margin_trailing_stop_is_sent_with_trailing_delta() {
8266 let server = wiremock::MockServer::start().await;
8267 let sent_as_trailing_stop = |request: &wiremock::Request| {
8268 let mut body_as_query = request.url.clone();
8270 body_as_query.set_query(Some(&String::from_utf8_lossy(&request.body)));
8271 let params: Vec<(String, String)> = request
8272 .url
8273 .query_pairs()
8274 .chain(body_as_query.query_pairs())
8275 .map(|(k, v)| (k.into_owned(), v.into_owned()))
8276 .collect();
8277 let has = |key: &str, value: &str| params.iter().any(|(k, v)| k == key && v == value);
8278 has("type", "STOP_LOSS")
8279 && has("trailingDelta", "25")
8280 && has("sideEffectType", "AUTO_BORROW_REPAY")
8281 && !params.iter().any(|(k, _)| k == "stopPrice" || k == "price")
8282 };
8283 wiremock::Mock::given(wiremock::matchers::method("POST"))
8284 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8285 .and(sent_as_trailing_stop)
8286 .respond_with(
8287 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
8288 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "trailing",
8289 "transactTime": 1_700_000_000_000_i64, "price": "0", "origQty": "2",
8290 "executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
8291 "timeInForce": "GTC", "type": "STOP_LOSS", "side": "SELL",
8292 })),
8293 )
8294 .expect(1)
8295 .mount(&server)
8296 .await;
8297 let client = margin_client_at(&server, false);
8298 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
8299 let trailing_request = |cid, offset| {
8300 let mut request = margin_open_request(&btcusdt, cid);
8301 request.state.side = Side::Sell;
8302 request.state.price = None;
8303 request.state.kind = OrderKind::TrailingStop {
8304 offset,
8305 offset_type: TrailingOffsetType::BasisPoints,
8306 activation_price: None,
8307 };
8308 request
8309 };
8310
8311 let placed = client
8312 .open_order(trailing_request("trailing", Decimal::from(25)))
8313 .await;
8314 assert!(
8315 matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
8316 "{placed:?}"
8317 );
8318
8319 let refused = client
8320 .open_order(trailing_request("fractional", Decimal::new(25, 1)))
8321 .await;
8322 assert!(
8323 matches!(
8324 refused.state,
8325 OrderState::Inactive(crate::order::state::InactiveOrderState::OpenFailed(
8326 OrderError::InvalidPrecision(_)
8327 ))
8328 ),
8329 "{refused:?}"
8330 );
8331 }
8332
8333 #[tokio::test]
8334 async fn placing_listing_and_cancelling_margin_orders_feed_the_known_live_set() {
8335 let server = wiremock::MockServer::start().await;
8336 wiremock::Mock::given(wiremock::matchers::method("POST"))
8337 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8338 .respond_with(
8339 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
8340 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "placed",
8341 "transactTime": 1_700_000_000_000_i64, "price": "100", "origQty": "2",
8342 "executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
8343 "timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
8344 })),
8345 )
8346 .mount(&server)
8347 .await;
8348 wiremock::Mock::given(wiremock::matchers::method("DELETE"))
8349 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8350 .respond_with(
8351 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
8352 "symbol": "BTCUSDT", "orderId": "7", "origClientOrderId": "placed",
8353 "clientOrderId": "cancel", "price": "100", "origQty": "2", "executedQty": "0",
8354 "cummulativeQuoteQty": "0", "status": "CANCELED", "timeInForce": "GTC",
8355 "type": "LIMIT", "side": "BUY",
8356 })),
8357 )
8358 .mount(&server)
8359 .await;
8360 wiremock::Mock::given(wiremock::matchers::method("GET"))
8361 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
8362 .respond_with(
8363 wiremock::ResponseTemplate::new(200)
8364 .set_body_json(serde_json::json!([margin_order_row("listed", "NEW")])),
8365 )
8366 .mount(&server)
8367 .await;
8368 let client = margin_client_at(&server, false);
8369 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
8370
8371 let placed = client
8372 .open_order(margin_open_request(&btcusdt, "placed"))
8373 .await;
8374 assert!(
8375 matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
8376 "{placed:?}"
8377 );
8378 assert!(
8379 client
8380 .known_live
8381 .lock()
8382 .contains(&ClientOrderId::new("placed"))
8383 );
8384
8385 client.fetch_open_orders(&[]).await.unwrap();
8386 assert!(
8387 client
8388 .known_live
8389 .lock()
8390 .contains(&ClientOrderId::new("listed"))
8391 );
8392
8393 let cancelled = client
8394 .cancel_order(OrderRequestCancel {
8395 key: OrderKey::new(
8396 ExchangeId::BinanceMargin,
8397 &btcusdt,
8398 StrategyId::new("strategy"),
8399 ClientOrderId::new("placed"),
8400 ),
8401 state: crate::order::request::RequestCancel {
8402 id: Some(VenueOrderId::Assigned(OrderId::new("7"))),
8403 },
8404 })
8405 .await;
8406 assert!(cancelled.state.is_ok(), "{cancelled:?}");
8407 assert!(
8408 !client
8409 .known_live
8410 .lock()
8411 .contains(&ClientOrderId::new("placed"))
8412 );
8413 }
8414
8415 #[tokio::test]
8417 async fn a_cancel_response_without_executed_qty_reports_the_fill_unknown() {
8418 let server = wiremock::MockServer::start().await;
8419 wiremock::Mock::given(wiremock::matchers::method("DELETE"))
8420 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8421 .respond_with(
8422 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
8423 "symbol": "BTCUSDT", "orderId": "7", "origClientOrderId": "placed",
8424 "clientOrderId": "cancel", "price": "100", "origQty": "2",
8425 "status": "CANCELED", "timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
8426 })),
8427 )
8428 .mount(&server)
8429 .await;
8430 let client = margin_client_at(&server, false);
8431
8432 let cancelled = client
8433 .cancel_order(OrderRequestCancel {
8434 key: OrderKey::new(
8435 ExchangeId::BinanceMargin,
8436 &InstrumentNameExchange::new("BTCUSDT"),
8437 StrategyId::new("strategy"),
8438 ClientOrderId::new("placed"),
8439 ),
8440 state: crate::order::request::RequestCancel {
8441 id: Some(VenueOrderId::Assigned(OrderId::new("7"))),
8442 },
8443 })
8444 .await;
8445 let Ok(cancelled) = cancelled.state else {
8446 panic!("expected Ok, got {cancelled:?}");
8447 };
8448 assert_eq!(cancelled.filled_quantity, None);
8449 }
8450
8451 #[test]
8452 fn a_reconnect_opens_a_gap_on_each_stream_instrument_and_checks_those_with_orders_held() {
8453 let [btc, eth, xrp] = ["BTCUSDT", "ETHUSDT", "XRPUSDT"].map(InstrumentNameExchange::new);
8454 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
8455 for (instrument, cid) in [(&btc, "btc"), (&xrp, "xrp")] {
8456 known.lock().live(
8457 &margin_key(instrument.name(), cid),
8458 Decimal::TWO,
8459 OrderKind::Limit,
8460 &Open::new(
8461 VenueOrderId::Assigned(OrderId::new(cid)),
8462 Utc::now(),
8463 Decimal::ZERO,
8464 ),
8465 );
8466 }
8467 let mut unrecovered = UnrecoveredFills::default();
8468 let mut unchecked = UncheckedOrders::default();
8469
8470 open_reconnect_recovery(
8471 &mut unrecovered,
8472 &mut unchecked,
8473 &known,
8474 &[btc.clone(), eth.clone()],
8475 Utc::now() - chrono::Duration::minutes(1),
8476 );
8477
8478 assert!(unrecovered.covers(&btc) && unrecovered.covers(ð));
8479 assert!(!unrecovered.covers(&xrp), "not the stream's");
8480 assert!(unchecked.contains(&btc));
8481 assert!(!unchecked.contains(ð), "no order held");
8482 assert!(!unchecked.contains(&xrp), "its fills are not recovered");
8483 }
8484
8485 #[tokio::test]
8488 async fn a_recovered_margin_fill_that_completes_a_held_order_ends_it() {
8489 let disconnect = Utc::now() - chrono::Duration::minutes(10);
8490 let server = wiremock::MockServer::start().await;
8491 wiremock::Mock::given(wiremock::matchers::method("GET"))
8492 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
8493 .respond_with(venue(vec![
8494 (
8495 1,
8496 (disconnect + chrono::Duration::minutes(1)).timestamp_millis(),
8497 ),
8498 (
8499 2,
8500 (disconnect + chrono::Duration::minutes(2)).timestamp_millis(),
8501 ),
8502 ]))
8503 .mount(&server)
8504 .await;
8505 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
8508 for (cid, order_id, quantity) in [("one", "1", "0.01"), ("two", "2", "0.02")] {
8509 known.lock().live(
8510 &margin_key("BTCUSDT", cid),
8511 Decimal::from_str(quantity).unwrap(),
8512 OrderKind::Limit,
8513 &Open::new(
8514 VenueOrderId::Assigned(OrderId::new(order_id)),
8515 Utc::now(),
8516 Decimal::ZERO,
8517 ),
8518 );
8519 }
8520 let mut unrecovered = UnrecoveredFills::default();
8521 unrecovered.open(
8522 &[InstrumentNameExchange::new("BTCUSDT")],
8523 disconnect,
8524 Utc::now(),
8525 );
8526 let (tx, mut rx) = mpsc::unbounded_channel();
8527
8528 recover_margin_fills(
8529 &margin_rest_at(&server),
8530 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
8531 &mut unrecovered,
8532 &tx,
8533 &new_dedup_cache(),
8534 &known,
8535 false,
8536 )
8537 .await;
8538
8539 assert_eq!(
8540 std::iter::from_fn(|| rx.try_recv().ok()).count(),
8541 2,
8542 "both fills"
8543 );
8544 let known = known.lock();
8545 assert!(!known.contains(&ClientOrderId::new("one")), "completed");
8546 assert!(known.contains(&ClientOrderId::new("two")), "part-filled");
8547 }
8548
8549 #[tokio::test]
8552 async fn a_margin_order_that_expired_on_placement_is_not_reported_or_held_open() {
8553 use rust_decimal_macros::dec;
8554
8555 for (cid, filled, quote) in [("unfilled", "0", "0"), ("partly", "1", "100")] {
8556 let server = wiremock::MockServer::start().await;
8557 wiremock::Mock::given(wiremock::matchers::method("POST"))
8558 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8559 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
8560 serde_json::json!({
8561 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid,
8562 "transactTime": 1_700_000_000_000_i64, "price": "100", "origQty": "2",
8563 "executedQty": filled, "cummulativeQuoteQty": quote,
8564 "status": "EXPIRED", "timeInForce": "IOC", "type": "LIMIT", "side": "BUY",
8565 }),
8566 ))
8567 .mount(&server)
8568 .await;
8569 let client = margin_client_at(&server, false);
8570 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
8571
8572 let placed = client.open_order(margin_open_request(&btcusdt, cid)).await;
8573
8574 let OrderState::Inactive(crate::order::state::InactiveOrderState::Expired(expired)) =
8575 &placed.state
8576 else {
8577 panic!("{cid}: expired, not open: {placed:?}");
8578 };
8579 assert_eq!(expired.filled_quantity, Decimal::from_str(filled).ok());
8580 let late = Open::new(
8581 VenueOrderId::Assigned(OrderId::new("7")),
8582 Utc::now(),
8583 Decimal::ZERO,
8584 );
8585 let mut known = client.known_live.lock();
8586 known.live(&placed.key, dec!(2), OrderKind::Limit, &late);
8587 assert!(
8588 !known.contains(&ClientOrderId::new(cid)),
8589 "{cid}: not held, even after a late live report"
8590 );
8591 }
8592 }
8593
8594 async fn mount_borrow_terms(
8598 server: &wiremock::MockServer,
8599 pair: serde_json::Value,
8600 asset: serde_json::Value,
8601 rate: serde_json::Value,
8602 max_borrow: serde_json::Value,
8603 ) {
8604 use wiremock::{
8605 Mock, ResponseTemplate,
8606 matchers::{method, path},
8607 };
8608 for (route, body) in [
8609 ("/sapi/v1/margin/allPairs", pair.clone()),
8610 ("/sapi/v1/margin/isolated/allPairs", pair),
8611 ("/sapi/v1/margin/allAssets", asset),
8612 ("/sapi/v1/margin/next-hourly-interest-rate", rate),
8613 ("/sapi/v1/margin/maxBorrowable", max_borrow),
8614 ("/sapi/v1/margin/delist-schedule", serde_json::json!([])),
8615 ] {
8616 Mock::given(method("GET"))
8617 .and(path(route))
8618 .respond_with(ResponseTemplate::new(200).set_body_json(body))
8619 .mount(server)
8620 .await;
8621 }
8622 }
8623
8624 async fn mount_delist_schedule(server: &wiremock::MockServer, schedule: serde_json::Value) {
8627 use wiremock::{
8628 Mock, ResponseTemplate,
8629 matchers::{method, path},
8630 };
8631 Mock::given(method("GET"))
8632 .and(path("/sapi/v1/margin/delist-schedule"))
8633 .respond_with(ResponseTemplate::new(200).set_body_json(schedule))
8634 .with_priority(1)
8635 .mount(server)
8636 .await;
8637 }
8638
8639 fn btcusdt_pair(is_margin_trade: bool, is_sell_allowed: bool) -> serde_json::Value {
8640 serde_json::json!([{
8641 "base": "BTC", "quote": "USDT", "symbol": "BTCUSDT",
8642 "isMarginTrade": is_margin_trade, "isSellAllowed": is_sell_allowed, "isBuyAllowed": true,
8643 }])
8644 }
8645
8646 fn btc_asset(is_borrowable: bool) -> serde_json::Value {
8647 serde_json::json!([{ "assetName": "BTC", "isBorrowable": is_borrowable }])
8648 }
8649
8650 fn btc_rate(rate: &str) -> serde_json::Value {
8651 serde_json::json!([{ "asset": "BTC", "nextHourlyInterestRate": rate }])
8652 }
8653
8654 async fn query_value(server: &wiremock::MockServer, route: &str, key: &str) -> Option<String> {
8656 server
8657 .received_requests()
8658 .await
8659 .unwrap()
8660 .iter()
8661 .find(|request| request.url.path() == route)
8662 .and_then(|request| {
8663 request
8664 .url
8665 .query_pairs()
8666 .find(|(name, _)| name == key)
8667 .map(|(_, value)| value.into_owned())
8668 })
8669 }
8670
8671 #[tokio::test]
8672 async fn margin_shortability_reads_the_pair_the_asset_and_the_annualised_rate() {
8673 use crate::client::ShortabilityClient;
8674 use rust_decimal_macros::dec;
8675 for (is_isolated, pairs_route, flag) in [
8676 (false, "/sapi/v1/margin/allPairs", "FALSE"),
8677 (true, "/sapi/v1/margin/isolated/allPairs", "TRUE"),
8678 ] {
8679 let server = wiremock::MockServer::start().await;
8680 mount_borrow_terms(
8681 &server,
8682 btcusdt_pair(true, true),
8683 btc_asset(true),
8684 btc_rate("0.00000571"),
8685 serde_json::json!({}),
8686 )
8687 .await;
8688
8689 let shortability = margin_client_at(&server, is_isolated)
8690 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8691 .await
8692 .unwrap();
8693
8694 assert_eq!(
8695 shortability,
8696 Shortability::new()
8697 .with_shortable(true)
8698 .with_fee_rate(dec!(0.00000571) * dec!(8760)),
8699 "isolated: {is_isolated}"
8700 );
8701 assert_eq!(
8702 query_value(&server, pairs_route, "symbol").await.as_deref(),
8703 Some("BTCUSDT"),
8704 "the pair list is asked for the one symbol"
8705 );
8706 assert_eq!(
8707 query_value(&server, "/sapi/v1/margin/allAssets", "asset")
8708 .await
8709 .as_deref(),
8710 Some("BTC"),
8711 "the base asset is the one borrowed"
8712 );
8713 let rate = "/sapi/v1/margin/next-hourly-interest-rate";
8714 assert_eq!(
8715 query_value(&server, rate, "assets").await.as_deref(),
8716 Some("BTC")
8717 );
8718 assert_eq!(
8719 query_value(&server, rate, "isIsolated").await.as_deref(),
8720 Some(flag)
8721 );
8722 }
8723 }
8724
8725 #[tokio::test]
8726 async fn margin_shortability_is_refused_by_any_flag_and_unknown_without_them() {
8727 use crate::client::ShortabilityClient;
8728 let cases = [
8729 (btcusdt_pair(false, true), btc_asset(true), Some(false)),
8730 (btcusdt_pair(true, false), btc_asset(true), Some(false)),
8731 (btcusdt_pair(true, true), btc_asset(false), Some(false)),
8732 (btcusdt_pair(true, true), serde_json::json!([]), None),
8734 ];
8735 for (pair, asset, shortable) in cases {
8736 let server = wiremock::MockServer::start().await;
8737 mount_borrow_terms(
8738 &server,
8739 pair.clone(),
8740 asset.clone(),
8741 serde_json::json!([]),
8742 serde_json::json!({}),
8743 )
8744 .await;
8745
8746 let shortability = margin_client_at(&server, false)
8747 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8748 .await
8749 .unwrap();
8750
8751 assert_eq!(shortability.shortable, shortable, "{pair} {asset}");
8752 assert_eq!(
8753 shortability.fee_rate, None,
8754 "no rate returned is an unknown fee"
8755 );
8756 assert_eq!(shortability.easy_to_borrow, None);
8757 assert_eq!(shortability.available, None);
8758 }
8759 }
8760
8761 #[tokio::test]
8765 async fn margin_shortability_reads_the_delisting_schedule_for_this_margin() {
8766 use crate::client::ShortabilityClient;
8767 let past = Utc::now() - chrono::Duration::hours(1);
8768 let soon = Utc::now() + chrono::Duration::days(3);
8769 let later = soon + chrono::Duration::days(7);
8770 let entry = |time: DateTime<Utc>, assets: &[&str], symbols: &[&str]| {
8771 serde_json::json!({
8772 "delistTime": time.timestamp_millis(),
8773 "crossMarginAssets": assets,
8774 "isolatedMarginSymbols": symbols,
8775 })
8776 };
8777 let millis = |time: DateTime<Utc>| DateTime::from_timestamp_millis(time.timestamp_millis());
8778 let cases = [
8779 (
8781 false,
8782 vec![entry(soon, &["BTC"], &[])],
8783 millis(soon),
8784 Some(true),
8785 ),
8786 (
8787 false,
8788 vec![entry(soon, &["USDT"], &[])],
8789 millis(soon),
8790 Some(true),
8791 ),
8792 (
8793 false,
8794 vec![
8795 entry(later, &["BTC"], &[]),
8796 entry(soon, &["ETH", "USDT"], &[]),
8797 ],
8798 millis(soon),
8799 Some(true),
8800 ),
8801 (
8802 false,
8803 vec![entry(soon, &["ETH"], &["BTCUSDT"])],
8804 None,
8805 Some(true),
8806 ),
8807 (
8808 false,
8809 vec![entry(past, &["BTC"], &[])],
8810 millis(past),
8811 Some(false),
8812 ),
8813 (
8815 true,
8816 vec![entry(soon, &[], &["BTCUSDT"])],
8817 millis(soon),
8818 Some(true),
8819 ),
8820 (
8821 true,
8822 vec![entry(soon, &["BTC", "USDT"], &["ETHUSDT"])],
8823 None,
8824 Some(true),
8825 ),
8826 (
8827 true,
8828 vec![entry(past, &[], &["BTCUSDT"])],
8829 millis(past),
8830 Some(false),
8831 ),
8832 ];
8833 for (is_isolated, schedule, delisting_at, shortable) in cases {
8834 let server = wiremock::MockServer::start().await;
8835 mount_borrow_terms(
8836 &server,
8837 btcusdt_pair(true, true),
8838 btc_asset(true),
8839 serde_json::json!([]),
8840 serde_json::json!({}),
8841 )
8842 .await;
8843 let schedule = serde_json::Value::from(schedule);
8844 mount_delist_schedule(&server, schedule.clone()).await;
8845
8846 let shortability = margin_client_at(&server, is_isolated)
8847 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8848 .await
8849 .unwrap();
8850
8851 let case = format!("isolated: {is_isolated}, {schedule}");
8852 assert_eq!(shortability.delisting_at, delisting_at, "{case}");
8853 assert_eq!(shortability.shortable, shortable, "{case}");
8854 }
8855 }
8856
8857 #[tokio::test]
8859 async fn margin_shortability_of_a_scheduled_pair_without_a_time_is_an_error() {
8860 use crate::client::ShortabilityClient;
8861 let server = wiremock::MockServer::start().await;
8862 mount_borrow_terms(
8863 &server,
8864 btcusdt_pair(true, true),
8865 btc_asset(true),
8866 serde_json::json!([]),
8867 serde_json::json!({}),
8868 )
8869 .await;
8870 mount_delist_schedule(
8871 &server,
8872 serde_json::json!([{ "crossMarginAssets": ["BTC"] }]),
8873 )
8874 .await;
8875
8876 let result = margin_client_at(&server, false)
8877 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8878 .await;
8879
8880 assert!(
8881 matches!(result, Err(UnindexedClientError::Internal(ref message))
8882 if message.contains("delistTime")),
8883 "{result:?}"
8884 );
8885 }
8886
8887 #[tokio::test]
8888 async fn margin_shortability_of_an_unlisted_pair_is_an_invalid_instrument() {
8889 use crate::client::ShortabilityClient;
8890 let server = wiremock::MockServer::start().await;
8891 mount_borrow_terms(
8892 &server,
8893 serde_json::json!([]),
8894 btc_asset(true),
8895 btc_rate("0.00000571"),
8896 serde_json::json!({}),
8897 )
8898 .await;
8899
8900 let result = margin_client_at(&server, false)
8901 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8902 .await;
8903
8904 assert!(
8905 matches!(
8906 result,
8907 Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(ref instrument, _)))
8908 if instrument.name().as_str() == "BTCUSDT"
8909 ),
8910 "{result:?}"
8911 );
8912 }
8913
8914 #[tokio::test]
8915 async fn margin_borrow_capacity_asks_for_the_asset_the_side_borrows() {
8916 use crate::client::BorrowCapacityClient;
8917 use rust_decimal_macros::dec;
8918 for (is_isolated, side, asset, isolated_symbol) in [
8919 (false, Side::Sell, "BTC", None),
8920 (false, Side::Buy, "USDT", None),
8921 (true, Side::Sell, "BTC", Some("BTCUSDT")),
8922 ] {
8923 let server = wiremock::MockServer::start().await;
8924 mount_borrow_terms(
8925 &server,
8926 btcusdt_pair(true, true),
8927 btc_asset(true),
8928 btc_rate("0.00000571"),
8929 serde_json::json!({ "amount": "1.69248805", "borrowLimit": "60" }),
8930 )
8931 .await;
8932
8933 let capacity = margin_client_at(&server, is_isolated)
8934 .fetch_borrow_capacity(&InstrumentNameExchange::new("BTCUSDT"), side)
8935 .await
8936 .unwrap();
8937
8938 assert_eq!(
8939 capacity,
8940 BorrowCapacity::new(AssetNameExchange::new(asset))
8941 .with_borrowable_now(dec!(1.69248805))
8942 .with_account_limit(dec!(60)),
8943 "{side:?} isolated: {is_isolated}"
8944 );
8945 let max_borrow = "/sapi/v1/margin/maxBorrowable";
8946 assert_eq!(
8947 query_value(&server, max_borrow, "asset").await.as_deref(),
8948 Some(asset)
8949 );
8950 assert_eq!(
8951 query_value(&server, max_borrow, "isolatedSymbol")
8952 .await
8953 .as_deref(),
8954 isolated_symbol,
8955 "only isolated margin names the pair's account"
8956 );
8957 }
8958 }
8959
8960 #[tokio::test]
8961 async fn margin_borrow_capacity_keeps_an_omitted_amount_unknown_and_fails_a_malformed_one() {
8962 use crate::client::BorrowCapacityClient;
8963 use rust_decimal_macros::dec;
8964 let instrument = InstrumentNameExchange::new("BTCUSDT");
8965
8966 let server = wiremock::MockServer::start().await;
8967 mount_borrow_terms(
8968 &server,
8969 btcusdt_pair(true, true),
8970 btc_asset(true),
8971 btc_rate("0"),
8972 serde_json::json!({ "borrowLimit": "60" }),
8973 )
8974 .await;
8975 let capacity = margin_client_at(&server, false)
8976 .fetch_borrow_capacity(&instrument, Side::Sell)
8977 .await
8978 .unwrap();
8979 assert_eq!(capacity.borrowable_now, None);
8980 assert_eq!(capacity.account_limit, Some(dec!(60)));
8981
8982 let server = wiremock::MockServer::start().await;
8983 mount_borrow_terms(
8984 &server,
8985 btcusdt_pair(true, true),
8986 btc_asset(true),
8987 btc_rate("0"),
8988 serde_json::json!({ "amount": "lots", "borrowLimit": "60" }),
8989 )
8990 .await;
8991 let result = margin_client_at(&server, false)
8992 .fetch_borrow_capacity(&instrument, Side::Sell)
8993 .await;
8994 assert!(
8995 matches!(result, Err(UnindexedClientError::Internal(ref message)) if message.contains("amount")),
8996 "{result:?}"
8997 );
8998 }
8999
9000 #[tokio::test]
9001 async fn margin_borrow_capacity_without_an_isolated_account_names_the_pair() {
9002 use crate::client::BorrowCapacityClient;
9003 use wiremock::{
9004 Mock,
9005 matchers::{method, path},
9006 };
9007 let server = wiremock::MockServer::start().await;
9008 Mock::given(method("GET"))
9009 .and(path("/sapi/v1/margin/maxBorrowable"))
9010 .respond_with(
9011 wiremock::ResponseTemplate::new(400).set_body_json(serde_json::json!({
9012 "code": -11001, "msg": "Isolated margin account does not exist."
9013 })),
9014 )
9015 .mount(&server)
9016 .await;
9017 mount_borrow_terms(
9018 &server,
9019 btcusdt_pair(true, true),
9020 btc_asset(true),
9021 btc_rate("0"),
9022 serde_json::json!({}),
9023 )
9024 .await;
9025
9026 let err = margin_client_at(&server, true)
9027 .fetch_borrow_capacity(&InstrumentNameExchange::new("BTCUSDT"), Side::Sell)
9028 .await
9029 .unwrap_err();
9030
9031 assert_eq!(
9032 err.to_string(),
9033 "API: request rejected: BTCUSDT: -11001 Isolated margin account does not exist."
9034 );
9035 assert!(!err.is_transient());
9036 }
9037}