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