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, FILL_RECOVERY_TIMEOUT_SECS,
69 HEARTBEAT_TIMEOUT_SECS, MyTradesFrom, ORDER_EXECUTIONS_BUDGET, OpenOrderListing,
70 PlacementResponse, RateLimitTracker, RequestKind, SIGNAL_RECOVERY_LOOKBACK_MS,
71 SharedDedupCache, TrailingOrderError, UnhandledEvents, UnrecoveredFills, UnsubscribeOnDrop,
72 UserDataFrame, WeightPool, binance_filled_qty, classify_order_kind_tif,
73 classify_rest_order_error, classify_rest_query_error, convert_ended_order,
74 convert_execution_report, convert_open_order_listing, convert_open_order_owned_symbol,
75 dedup_key_from_event, disconnect_error, drop_after, frame_excerpt, gap_failed, gap_time,
76 is_duplicate, is_unknown_order, log_unhandled_event, log_unrecognised_frame, new_dedup_cache,
77 parse_user_data_frame, placed_order_state, recovered_order_cids, recovered_order_totals,
78 response_decode_error, rest_call_with_retry, retry_or_end, trailing_params,
79};
80use crate::{
81 AccountEventKind, AccountSnapshot, FillRecoveryFailure, InstrumentAccountSnapshot,
82 InstrumentBalanceUpdate, IsolatedInstrumentState, IsolatedMarginRisk, UnindexedAccountEvent,
83 UnindexedAccountSnapshot,
84 balance::{AssetBalance, AssetBalanceUpdate, Balance, BalanceUpdate},
85 borrow_capacity::BorrowCapacity,
86 client::{
87 BorrowCapacityClient, ExecutionClient, OrderStatusClient, ShortabilityClient,
88 order_recovery::{
89 KnownLiveOrders, OpenListing, OrderLookup, SharedKnownLiveOrders, UncheckedOrders,
90 fetch_ended_by_key, recover_ended_orders,
91 },
92 reconnect_budget::{BackoffLimits, ReconnectBudget},
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;
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
164pub use crate::margin::MarginSideEffect;
171
172impl MarginSideEffect {
173 pub fn as_binance_str(self) -> &'static str {
175 match self {
176 MarginSideEffect::AutoBorrowRepay => "AUTO_BORROW_REPAY",
177 MarginSideEffect::NoBorrow => "NO_SIDE_EFFECT",
178 }
179 }
180
181 fn as_sdk_enum(self) -> MarginAccountNewOrderSideEffectTypeEnum {
187 match self {
188 MarginSideEffect::AutoBorrowRepay => {
189 MarginAccountNewOrderSideEffectTypeEnum::AutoBorrowRepay
190 }
191 MarginSideEffect::NoBorrow => MarginAccountNewOrderSideEffectTypeEnum::NoSideEffect,
192 }
193 }
194}
195
196#[derive(Clone, Deserialize)]
206pub struct BinanceMarginConfig {
207 api_key: String,
210 secret_key: String,
211 pub testnet: bool,
217 #[serde(default)]
224 pub is_isolated: bool,
225 #[serde(default)]
236 pub isolated_symbols: Vec<InstrumentNameExchange>,
237 #[serde(default)]
241 pub side_effect: MarginSideEffect,
242}
243
244impl std::fmt::Debug for BinanceMarginConfig {
246 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
247 f.debug_struct("BinanceMarginConfig")
248 .field("api_key", &"***")
249 .field("secret_key", &"***")
250 .field("testnet", &self.testnet)
251 .field("is_isolated", &self.is_isolated)
252 .field("isolated_symbols", &self.isolated_symbols)
253 .field("side_effect", &self.side_effect)
254 .finish()
255 }
256}
257
258impl BinanceMarginConfig {
259 pub fn new(
269 api_key: String,
270 secret_key: String,
271 testnet: bool,
272 is_isolated: bool,
273 isolated_symbols: Vec<InstrumentNameExchange>,
274 side_effect: MarginSideEffect,
275 ) -> Self {
276 Self {
277 api_key,
278 secret_key,
279 testnet,
280 is_isolated,
281 isolated_symbols,
282 side_effect,
283 }
284 }
285
286 pub fn cross_margin(api_key: String, secret_key: String) -> Self {
293 Self::new(
294 api_key,
295 secret_key,
296 false,
297 false,
298 Vec::new(),
299 MarginSideEffect::default(),
300 )
301 }
302
303 pub fn isolated(
311 api_key: String,
312 secret_key: String,
313 symbols: Vec<InstrumentNameExchange>,
314 ) -> Self {
315 Self::new(
316 api_key,
317 secret_key,
318 false,
319 true,
320 symbols,
321 MarginSideEffect::default(),
322 )
323 }
324
325 pub fn api_key(&self) -> &str {
327 &self.api_key
328 }
329}
330
331#[derive(Clone)]
453pub struct BinanceMargin {
454 config: Arc<BinanceMarginConfig>,
455 rest: Arc<RestApi>,
457 rest_config: Arc<ConfigurationRestApi>,
461 ws_config: ConfigurationWebsocketApi,
465 rate_limiter: Arc<RateLimitTracker>,
467 known_live: SharedKnownLiveOrders,
470}
471
472impl std::fmt::Debug for BinanceMargin {
473 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
474 f.debug_struct("BinanceMargin")
475 .field("testnet", &self.config.testnet)
476 .field("is_isolated", &self.config.is_isolated)
477 .field("side_effect", &self.config.side_effect)
478 .finish_non_exhaustive()
479 }
480}
481
482impl BinanceMargin {
483 fn remember_live<'a>(
485 &self,
486 orders: impl IntoIterator<Item = &'a Order<ExchangeId, InstrumentNameExchange, Open>>,
487 ) {
488 let mut known = self.known_live.lock();
489 for order in orders {
490 known.live(&order.key, order.quantity, order.kind, &order.state);
491 }
492 }
493
494 #[allow(clippy::expect_used)] fn build_rest_config(config: &BinanceMarginConfig) -> ConfigurationRestApi {
504 ConfigurationRestApi::builder()
505 .api_key(config.api_key.clone())
506 .api_secret(config.secret_key.clone())
507 .base_path(MARGIN_TRADING_REST_API_PROD_URL)
508 .build()
509 .expect("failed to build Binance margin REST configuration")
510 }
511
512 #[allow(clippy::expect_used)] fn build_ws_config(config: &BinanceMarginConfig) -> ConfigurationWebsocketApi {
523 ConfigurationWebsocketApi::builder()
524 .api_key(config.api_key.clone())
525 .api_secret(config.secret_key.clone())
526 .ws_url(SPOT_WS_API_PROD_URL)
530 .build()
531 .expect("failed to build Binance margin WebSocket configuration")
532 }
533
534 fn effective_isolated_set(
547 &self,
548 instruments: &[InstrumentNameExchange],
549 ) -> Vec<InstrumentNameExchange> {
550 if instruments.is_empty() {
551 return self.config.isolated_symbols.clone();
552 }
553 instruments
554 .iter()
555 .filter(|inst| {
556 let in_set = self.config.isolated_symbols.contains(*inst);
557 if !in_set {
558 warn!(
559 instrument = %inst.name(),
560 "BinanceMargin isolated: requested instrument not in configured isolated_symbols — skipping"
561 );
562 }
563 in_set
564 })
565 .cloned()
566 .collect()
567 }
568
569 async fn isolated_account_snapshot(
573 &self,
574 instruments: &[InstrumentNameExchange],
575 ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
576 use futures::{StreamExt as _, TryStreamExt as _};
577
578 let effective = self.effective_isolated_set(instruments);
579 if effective.is_empty() {
580 return Ok(AccountSnapshot::new(
583 ExchangeId::BinanceMargin,
584 Vec::new(),
585 Vec::new(),
586 ));
587 }
588
589 let mut balances_by_symbol = convert_isolated_margin_assets(
591 fetch_isolated_margin_account_info(
592 self.rest.clone(),
593 self.rate_limiter.clone(),
594 effective.clone(),
595 )
596 .await?,
597 );
598
599 let instrument_snapshots: Vec<_> =
602 futures::stream::iter(effective.into_iter().map(|instrument| {
603 fetch_margin_open_orders_for_instrument(
604 self.rest.clone(),
605 self.rate_limiter.clone(),
606 instrument,
607 true,
608 RequestKind::Query,
609 )
610 }))
611 .buffer_unordered(8)
612 .map(|result| {
613 let (inst, listing) = result?;
614 self.remember_live(&listing.orders);
615 let wrapped = listing
616 .orders
617 .into_iter()
618 .map(active_order_snapshot)
619 .collect();
620 let isolated = balances_by_symbol.remove(&inst);
621 if isolated.is_none() {
622 warn!(
626 instrument = %inst.name(),
627 "BinanceMargin isolated: no isolated balance entry returned for instrument — snapshot will carry isolated: None"
628 );
629 }
630 Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
631 inst,
632 wrapped,
633 listing.complete,
634 PositionReport::Unreported,
635 isolated,
636 ))
637 })
638 .try_collect()
639 .await?;
640
641 Ok(AccountSnapshot::new(
644 ExchangeId::BinanceMargin,
645 Vec::new(),
646 instrument_snapshots,
647 ))
648 }
649}
650
651impl ExecutionClient for BinanceMargin {
652 const EXCHANGE: ExchangeId = ExchangeId::BinanceMargin;
653
654 const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] =
657 &[InstrumentKindDiscriminant::Spot];
658 type Config = BinanceMarginConfig;
659 type AccountStream = BoxStream<'static, UnindexedAccountEvent>;
660
661 fn new(config: Self::Config) -> Self {
674 if config.testnet {
675 warn!(
676 "BinanceMarginConfig.testnet = true is ignored: Binance margin has no testnet; \
677 using production endpoints"
678 );
679 }
680 assert!(
681 !(config.is_isolated && config.isolated_symbols.is_empty()),
682 "BinanceMarginConfig: is_isolated = true requires a non-empty isolated_symbols"
683 );
684 let rest_config = Self::build_rest_config(&config);
689 let rest = Arc::new(MarginTradingRestApi::production(rest_config.clone()));
690 let rest_config = Arc::new(rest_config);
691 let ws_config = Self::build_ws_config(&config);
692 Self {
693 config: Arc::new(config),
694 rest,
695 rest_config,
696 ws_config,
697 rate_limiter: Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
698 known_live: KnownLiveOrders::shared(ExchangeId::BinanceMargin),
699 }
700 }
701
702 async fn open_order(
721 &self,
722 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
723 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
724 let instrument = request.key.instrument.clone();
725 let side = request.state.side;
726 let price = request.state.price;
727 let quantity = request.state.quantity;
728 let kind = request.state.kind;
729 let time_in_force = request.state.time_in_force;
730 let cid = request.key.cid.clone();
731
732 let order_key = OrderKey::new(
733 ExchangeId::BinanceMargin,
734 instrument.clone(),
735 request.key.strategy.clone(),
736 cid.clone(),
737 );
738
739 let inactive = |state: UnindexedOrderState| Order {
742 key: order_key.clone(),
743 side,
744 price,
745 quantity,
746 kind,
747 time_in_force,
748 state,
749 };
750
751 let params = match build_new_order_params(
752 instrument.name().to_string(),
753 side,
754 price,
755 quantity,
756 kind,
757 time_in_force,
758 cid.0.to_string(),
759 self.config.side_effect,
760 self.config.is_isolated,
761 ) {
762 Ok(params) => params,
763 Err(BuildOrderError::Unsupported) => {
764 return inactive(OrderState::inactive(OrderError::UnsupportedOrderType(
765 format!(
766 "BinanceMargin does not support OrderKind::{kind:?} with {time_in_force:?}"
767 ),
768 )));
769 }
770 Err(BuildOrderError::Trailing(refused)) => {
771 return inactive(OrderState::inactive(refused.into_order_error()));
772 }
773 Err(BuildOrderError::Build(msg)) => {
774 error!(%msg, "BinanceMargin failed to build new order params");
775 return inactive(OrderState::inactive(OrderError::Rejected(
776 ApiError::OrderRejected(msg),
777 )));
778 }
779 };
780
781 let response =
782 match rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Order, |rest| {
783 let params = params.clone();
784 Box::pin(async move { rest.margin_account_new_order(params).await })
785 })
786 .await
787 {
788 Ok(response) => response,
789 Err(e) => {
790 return inactive(OrderState::inactive(classify_rest_order_error(
791 &e,
792 &instrument,
793 )));
794 }
795 };
796
797 let data = match response.data().await {
798 Ok(data) => data,
799 Err(e) => {
800 return inactive(OrderState::inactive(OrderError::Rejected(
803 ApiError::OrderRejected(e.to_string()),
804 )));
805 }
806 };
807
808 let time_exchange = data
809 .transact_time
810 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
811 .unwrap_or_else(Utc::now);
812
813 let exchange_order_id = match data.order_id {
814 Some(id) => OrderId(format_smolstr!("{id}")),
815 None => {
816 error!("BinanceMargin open_order response missing orderId");
817 return inactive(OrderState::inactive(OrderError::Rejected(
818 ApiError::OrderRejected("open_order response missing orderId".into()),
819 )));
820 }
821 };
822
823 let state = placed_order_state(
826 ExchangeId::BinanceMargin,
827 &instrument,
828 quantity,
829 PlacementResponse {
830 status: data.status.as_deref(),
831 order_id: exchange_order_id,
832 time_exchange,
833 executed_qty: data.executed_qty.as_deref(),
834 cumulative_quote_qty: data.cummulative_quote_qty.as_deref(),
835 },
836 );
837 self.known_live
838 .lock()
839 .placed(&order_key, quantity, kind, &state);
840
841 Order {
842 key: order_key,
843 side,
844 price,
845 quantity,
846 kind,
847 time_in_force,
848 state,
849 }
850 }
851
852 async fn cancel_order(
862 &self,
863 request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
864 ) -> UnindexedOrderResponseCancel {
865 let instrument = request.key.instrument.clone();
866 let key = OrderKey {
867 exchange: request.key.exchange,
868 instrument: instrument.clone(),
869 strategy: request.key.strategy.clone(),
870 cid: request.key.cid.clone(),
871 };
872
873 let params = match build_cancel_order_params(
874 instrument.name().to_string(),
875 request.state.id.as_ref().and_then(VenueOrderId::assigned),
876 &request.key.cid,
877 self.config.is_isolated,
878 ) {
879 Ok(p) => p,
880 Err(e) => {
881 error!(%e, "BinanceMargin failed to build cancel order params");
882 return UnindexedOrderResponseCancel {
883 key,
884 state: Err(OrderError::Rejected(ApiError::OrderRejected(e))),
885 };
886 }
887 };
888
889 let response =
890 match rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Order, |rest| {
891 let params = params.clone();
892 Box::pin(async move { rest.margin_account_cancel_order(params).await })
893 })
894 .await
895 {
896 Ok(response) => response,
897 Err(e) => {
898 return UnindexedOrderResponseCancel {
899 key,
900 state: Err(classify_rest_order_error(&e, &instrument)),
901 };
902 }
903 };
904
905 let data = match response.data().await {
906 Ok(data) => data,
907 Err(e) => {
908 return UnindexedOrderResponseCancel {
911 key,
912 state: Err(OrderError::Rejected(ApiError::OrderRejected(e.to_string()))),
913 };
914 }
915 };
916
917 let time_exchange = Utc::now();
919
920 let exchange_order_id = match data.order_id {
921 Some(id) => OrderId(SmolStr::new(id)),
924 None => {
925 error!("BinanceMargin cancel response missing orderId");
926 return UnindexedOrderResponseCancel {
927 key,
928 state: Err(OrderError::Rejected(ApiError::OrderRejected(
929 "cancel response missing orderId".into(),
930 ))),
931 };
932 }
933 };
934
935 let filled_qty = binance_filled_qty(
936 ExchangeId::BinanceMargin,
937 &exchange_order_id,
938 data.executed_qty.as_deref(),
939 );
940
941 self.known_live.lock().ended(&key.cid);
942 UnindexedOrderResponseCancel {
943 key,
944 state: Ok(Cancelled::new(exchange_order_id, time_exchange, filled_qty)),
945 }
946 }
947
948 async fn account_snapshot(
963 &self,
964 assets: &[AssetNameExchange],
965 instruments: &[InstrumentNameExchange],
966 ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
967 if self.config.is_isolated {
968 return self.isolated_account_snapshot(instruments).await;
969 }
970 let response =
971 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
972 Box::pin(async move {
973 let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
974 rest.query_cross_margin_account_details(params).await
975 })
976 })
977 .await
978 .map_err(|e| classify_rest_query_error(&e, None))?;
979
980 let account = response.data().await.map_err(response_decode_error)?;
981
982 let balances =
983 filter_and_convert_margin_balances(account.user_assets.unwrap_or_default(), assets);
984
985 use futures::{StreamExt as _, TryStreamExt as _};
990 let instrument_snapshots: Vec<_> =
991 futures::stream::iter(instruments.iter().cloned().map(|instrument| {
992 fetch_margin_open_orders_for_instrument(
993 self.rest.clone(),
994 self.rate_limiter.clone(),
995 instrument,
996 false,
998 RequestKind::Query,
999 )
1000 }))
1001 .buffer_unordered(8)
1002 .map(|result| {
1003 let (inst, listing) = result?;
1004 self.remember_live(&listing.orders);
1005 let wrapped = listing
1006 .orders
1007 .into_iter()
1008 .map(active_order_snapshot)
1009 .collect();
1010 Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
1011 inst,
1012 wrapped,
1013 listing.complete,
1014 PositionReport::Unreported,
1015 None,
1016 ))
1017 })
1018 .try_collect()
1019 .await?;
1020
1021 Ok(AccountSnapshot::new(
1022 ExchangeId::BinanceMargin,
1023 balances,
1024 instrument_snapshots,
1025 ))
1026 }
1027
1028 async fn fetch_balances(
1038 &self,
1039 assets: &[AssetNameExchange],
1040 ) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
1041 if self.config.is_isolated {
1042 return Ok(Vec::new());
1043 }
1044 let response =
1045 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1046 Box::pin(async move {
1047 let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
1048 rest.query_cross_margin_account_details(params).await
1049 })
1050 })
1051 .await
1052 .map_err(|e| classify_rest_query_error(&e, None))?;
1053
1054 let account = response.data().await.map_err(response_decode_error)?;
1055
1056 Ok(filter_and_convert_margin_balances(
1057 account.user_assets.unwrap_or_default(),
1058 assets,
1059 ))
1060 }
1061
1062 async fn fetch_open_orders(
1081 &self,
1082 instruments: &[InstrumentNameExchange],
1083 ) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
1084 use futures::{StreamExt as _, TryStreamExt as _};
1085
1086 if self.config.is_isolated {
1090 let effective = self.effective_isolated_set(instruments);
1091 let capacity = effective.len();
1092 return futures::stream::iter(effective.into_iter().map(|instrument| {
1093 fetch_margin_open_orders_for_instrument(
1094 self.rest.clone(),
1095 self.rate_limiter.clone(),
1096 instrument,
1097 true,
1098 RequestKind::Query,
1099 )
1100 }))
1101 .buffer_unordered(8)
1102 .try_fold(
1103 Vec::with_capacity(capacity),
1104 |mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>,
1105 (_, listing)| async move {
1106 acc.extend(listing.orders);
1107 Ok(acc)
1108 },
1109 )
1110 .await
1111 .inspect(|orders| self.remember_live(orders));
1112 }
1113
1114 if instruments.is_empty() {
1117 let orders =
1118 fetch_margin_all_open_orders(self.rest.clone(), self.rate_limiter.clone(), false)
1119 .await?;
1120 self.remember_live(&orders);
1121 return Ok(orders);
1122 }
1123 futures::stream::iter(instruments.iter().cloned().map(|instrument| {
1124 fetch_margin_open_orders_for_instrument(
1125 self.rest.clone(),
1126 self.rate_limiter.clone(),
1127 instrument,
1128 false,
1129 RequestKind::Query,
1130 )
1131 }))
1132 .buffer_unordered(8)
1133 .try_fold(
1134 Vec::with_capacity(instruments.len()),
1135 |mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, (_, listing)| async move {
1136 acc.extend(listing.orders);
1137 Ok(acc)
1138 },
1139 )
1140 .await
1141 .inspect(|orders| self.remember_live(orders))
1142 }
1143
1144 async fn fetch_trades(
1173 &self,
1174 start: DateTime<Utc>,
1175 end: DateTime<Utc>,
1176 instruments: &[InstrumentNameExchange],
1177 ) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
1178 use futures::StreamExt as _;
1179
1180 if start > end {
1181 return Ok(TradesRead::complete(Vec::new()));
1182 }
1183
1184 let effective: Vec<InstrumentNameExchange> = if self.config.is_isolated {
1187 self.effective_isolated_set(instruments)
1188 } else {
1189 instruments.to_vec()
1190 };
1191 if effective.is_empty() {
1192 debug!(
1195 is_isolated = self.config.is_isolated,
1196 "BinanceMargin fetch_trades: empty effective instrument set — returning empty result"
1197 );
1198 return Ok(TradesRead::complete(Vec::new()));
1199 }
1200 let span = MyTradesFrom::Span {
1202 start: start.timestamp_millis(),
1203 end: end.timestamp_millis(),
1204 };
1205 let is_isolated = self.config.is_isolated;
1206 let mut all_trades = Vec::new();
1207
1208 let mut stream = futures::stream::iter(effective.into_iter().map(|inst| {
1211 let rest = self.rest.clone();
1212 let rate_limiter = self.rate_limiter.clone();
1213 async move {
1214 let pages = paginate_margin_my_trades(
1215 &rest,
1216 &rate_limiter,
1217 &inst,
1218 span,
1219 is_isolated,
1220 RequestKind::Query,
1221 )
1222 .await?;
1223 Ok::<_, UnindexedClientError>((inst, pages))
1224 }
1225 }))
1226 .buffer_unordered(8);
1227 while let Some(result) = stream.next().await {
1228 let (instrument, trades_data) = result?;
1229 for t in trades_data {
1230 if let Some(trade) = convert_margin_trade(&t, &instrument)
1231 && (start..=end).contains(&trade.time_exchange)
1232 {
1233 all_trades.push(trade);
1234 }
1235 }
1236 }
1237
1238 Ok(TradesRead::complete(all_trades))
1239 }
1240
1241 async fn account_stream(
1378 &self,
1379 _assets: &[AssetNameExchange],
1382 instruments: &[InstrumentNameExchange],
1383 ) -> Result<Self::AccountStream, UnindexedClientError> {
1384 let (tx, rx) = mpsc::unbounded_channel::<UnindexedAccountEvent>();
1385 let dedup = new_dedup_cache();
1386 let rest = self.rest.clone();
1387 let rest_config = self.rest_config.clone();
1388 let ws_config = self.ws_config.clone();
1389 let rate_limiter = self.rate_limiter.clone();
1390 let known_live = self.known_live.clone();
1391
1392 let cm_handle = if self.config.is_isolated {
1393 let symbols = self.config.isolated_symbols.clone();
1397 debug_assert!(
1400 symbols.iter().all(|i| i.name().len() <= 23),
1401 "instrument name exceeds SmolStr inline capacity: {:?}",
1402 symbols.iter().find(|i| i.name().len() > 23)
1403 );
1404
1405 let assets = fetch_isolated_margin_account_info(
1411 rest.clone(),
1412 rate_limiter.clone(),
1413 symbols.clone(),
1414 )
1415 .await?;
1416 let base_quote = Arc::new(build_base_quote_map(&assets));
1417 for sym in &symbols {
1418 if !base_quote.contains_key(sym) {
1419 warn!(
1420 symbol = %sym.name(),
1421 "BinanceMargin isolated: account info returned no base/quote for configured \
1422 symbol — its live per-pair balance frames will be dropped (fills/orders \
1423 unaffected; balances available via account_snapshot)"
1424 );
1425 }
1426 }
1427
1428 let tokens = acquire_all_isolated_tokens(&rest_config, &rate_limiter, &symbols).await?;
1432 rate_limiter.before_stream_open().await?;
1435 let live = isolated_connect_and_subscribe(
1436 &ws_config,
1437 &tx,
1438 &dedup,
1439 &known_live,
1440 &base_quote,
1441 &tokens,
1442 )
1443 .await
1444 .map_err(|e| {
1445 rate_limiter.stream_open_error(&e, |e| {
1446 UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
1447 })
1448 })?;
1449
1450 tokio::spawn(isolated_connection_manager(
1451 tx,
1452 dedup,
1453 ws_config,
1454 rest_config,
1455 rest,
1456 rate_limiter,
1457 known_live,
1458 symbols,
1459 base_quote,
1460 Some(live),
1461 ))
1462 } else {
1463 let instruments = instruments.to_vec();
1465 debug_assert!(
1469 instruments.iter().all(|i| i.name().len() <= 23),
1470 "instrument name exceeds SmolStr inline capacity: {:?}",
1471 instruments.iter().find(|i| i.name().len() > 23)
1472 );
1473
1474 let initial_token =
1480 acquire_user_listen_token(&rest_config, &rate_limiter, None).await?;
1481 rate_limiter.before_stream_open().await?;
1483 let initial_ws = connect_margin_ws(&ws_config).await.map_err(|e| {
1484 rate_limiter.stream_open_error(&e, |e| {
1485 UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
1486 })
1487 })?;
1488
1489 let risk = RiskStreamTask::spawn(
1493 SapiRiskListenKeys {
1494 rest: rest.clone(),
1495 rate_limiter: rate_limiter.clone(),
1496 },
1497 RISK_STREAM_URL.to_owned(),
1498 tx.clone(),
1499 dedup.clone(),
1500 );
1501 let manager = margin_connection_manager(
1502 tx,
1503 dedup,
1504 ws_config,
1505 rest_config,
1506 rest,
1507 rate_limiter,
1508 known_live,
1509 instruments,
1510 Some((initial_ws, initial_token)),
1511 );
1512 tokio::spawn(async move {
1513 let _risk = risk;
1514 manager.await;
1515 })
1516 };
1517
1518 let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
1519 let guarded_stream = AbortOnDropStream::new(rx_stream, cm_handle);
1520 Ok(futures::StreamExt::boxed(guarded_stream))
1521 }
1522}
1523
1524impl OrderStatusClient for BinanceMargin {
1541 async fn fetch_ended_orders(
1542 &self,
1543 orders: &[UnindexedOrderKey],
1544 ) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError> {
1545 fetch_ended_by_key(orders, |key| {
1546 let recorded_kind = self.known_live.lock().kind_of(&key.cid);
1547 fetch_margin_order_lookup(
1548 self.rest.clone(),
1549 self.rate_limiter.clone(),
1550 key,
1551 self.config.is_isolated,
1552 RequestKind::Query,
1553 recorded_kind,
1554 )
1555 })
1556 .await
1557 }
1558}
1559
1560const HOURS_PER_YEAR: u32 = 24 * 365;
1566
1567impl ShortabilityClient for BinanceMargin {
1568 async fn fetch_shortability(
1595 &self,
1596 instrument: &InstrumentNameExchange,
1597 ) -> Result<Shortability, UnindexedClientError> {
1598 let pair = self.fetch_margin_pair(instrument).await?;
1599 let (borrowable, hourly_rate, delisting_at) = tokio::try_join!(
1600 self.fetch_asset_borrowable(&pair.base),
1601 self.fetch_next_hourly_interest_rate(&pair.base),
1602 self.fetch_delisting_at(instrument, &pair),
1603 )?;
1604
1605 let mut shortability = Shortability::new();
1606 if let Some(delisting_at) = delisting_at {
1607 shortability = shortability.with_delisting_at(delisting_at);
1608 }
1609 let not_delisted = Some(delisting_at.is_none_or(|delisting_at| delisting_at > Utc::now()));
1613 let flags = [
1614 pair.is_margin_trade,
1615 pair.is_sell_allowed,
1616 borrowable,
1617 not_delisted,
1618 ];
1619 if flags.contains(&Some(false)) {
1620 shortability = shortability.with_shortable(false);
1621 } else if flags.iter().all(|flag| *flag == Some(true)) {
1622 shortability = shortability.with_shortable(true);
1623 }
1624 if let Some(hourly_rate) = hourly_rate {
1625 shortability = shortability.with_fee_rate(hourly_rate * Decimal::from(HOURS_PER_YEAR));
1626 }
1627 Ok(shortability)
1628 }
1629}
1630
1631impl BorrowCapacityClient for BinanceMargin {
1632 async fn fetch_borrow_capacity(
1649 &self,
1650 instrument: &InstrumentNameExchange,
1651 side: Side,
1652 ) -> Result<BorrowCapacity, UnindexedClientError> {
1653 let pair = self.fetch_margin_pair(instrument).await?;
1654 let asset = match side {
1655 Side::Buy => pair.quote,
1656 Side::Sell => pair.base,
1657 };
1658 let isolated_symbol = self
1659 .config
1660 .is_isolated
1661 .then(|| instrument.name().to_string());
1662 let response =
1663 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1664 let asset = asset.name().to_string();
1665 let isolated_symbol = isolated_symbol.clone();
1666 Box::pin(async move {
1667 let params = QueryMaxBorrowParams::builder(asset)
1668 .isolated_symbol(isolated_symbol)
1669 .build()?;
1670 rest.query_max_borrow(params).await
1671 })
1672 })
1673 .await
1674 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
1675 let max_borrow = response.data().await.map_err(response_decode_error)?;
1676
1677 let mut capacity = BorrowCapacity::new(asset);
1678 if let Some(amount) = optional_decimal("amount", max_borrow.amount.as_deref())? {
1679 capacity = capacity.with_borrowable_now(amount);
1680 }
1681 if let Some(limit) = optional_decimal("borrowLimit", max_borrow.borrow_limit.as_deref())? {
1682 capacity = capacity.with_account_limit(limit);
1683 }
1684 Ok(capacity)
1685 }
1686}
1687
1688#[derive(Debug)]
1690struct MarginPair {
1691 base: AssetNameExchange,
1692 quote: AssetNameExchange,
1693 is_margin_trade: Option<bool>,
1694 is_sell_allowed: Option<bool>,
1695}
1696
1697impl BinanceMargin {
1698 async fn fetch_margin_pair(
1702 &self,
1703 instrument: &InstrumentNameExchange,
1704 ) -> Result<MarginPair, UnindexedClientError> {
1705 let symbol = instrument.name().to_string();
1706 let listed = if self.config.is_isolated {
1707 let response =
1708 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1709 let symbol = symbol.clone();
1710 Box::pin(async move {
1711 let params = GetAllIsolatedMarginSymbolParams::builder()
1712 .symbol(symbol)
1713 .build()?;
1714 rest.get_all_isolated_margin_symbol(params).await
1715 })
1716 })
1717 .await
1718 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
1719 response
1720 .data()
1721 .await
1722 .map_err(response_decode_error)?
1723 .into_iter()
1724 .find(|pair| pair.symbol.as_deref() == Some(symbol.as_str()))
1725 .map(|pair| {
1726 (
1727 pair.base,
1728 pair.quote,
1729 pair.is_margin_trade,
1730 pair.is_sell_allowed,
1731 )
1732 })
1733 } else {
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 = GetAllCrossMarginPairsParams::builder()
1739 .symbol(symbol)
1740 .build()?;
1741 rest.get_all_cross_margin_pairs(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 };
1761 let margin = if self.config.is_isolated {
1762 "isolated"
1763 } else {
1764 "cross"
1765 };
1766 let Some((base, quote, is_margin_trade, is_sell_allowed)) = listed else {
1767 return Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(
1768 instrument.clone(),
1769 format!("Binance lists no {margin} margin pair {symbol}"),
1770 )));
1771 };
1772 let (Some(base), Some(quote)) = (base, quote) else {
1773 return Err(UnindexedClientError::Internal(format!(
1774 "Binance lists {margin} margin pair {symbol} without its base or quote asset"
1775 )));
1776 };
1777 Ok(MarginPair {
1778 base: AssetNameExchange::new(base),
1779 quote: AssetNameExchange::new(quote),
1780 is_margin_trade,
1781 is_sell_allowed,
1782 })
1783 }
1784
1785 async fn fetch_asset_borrowable(
1788 &self,
1789 asset: &AssetNameExchange,
1790 ) -> Result<Option<bool>, UnindexedClientError> {
1791 let name = asset.name().to_string();
1792 let response =
1793 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1794 let name = name.clone();
1795 Box::pin(async move {
1796 let params = GetAllMarginAssetsParams::builder().asset(name).build()?;
1797 rest.get_all_margin_assets(params).await
1798 })
1799 })
1800 .await
1801 .map_err(|e| classify_rest_query_error(&e, None))?;
1802 Ok(response
1803 .data()
1804 .await
1805 .map_err(response_decode_error)?
1806 .into_iter()
1807 .find(|listed| listed.asset_name.as_deref() == Some(name.as_str()))
1808 .and_then(|listed| listed.is_borrowable))
1809 }
1810
1811 async fn fetch_delisting_at(
1816 &self,
1817 instrument: &InstrumentNameExchange,
1818 pair: &MarginPair,
1819 ) -> Result<Option<DateTime<Utc>>, UnindexedClientError> {
1820 let response =
1821 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1822 Box::pin(async move {
1823 let params = GetDelistScheduleParams::builder().build()?;
1824 rest.get_delist_schedule(params).await
1825 })
1826 })
1827 .await
1828 .map_err(|e| classify_rest_query_error(&e, None))?;
1829 let schedule = response.data().await.map_err(response_decode_error)?;
1830
1831 let symbol = instrument.name().as_str();
1832 let (base, quote) = (pair.base.name().as_str(), pair.quote.name().as_str());
1833 let mut earliest: Option<DateTime<Utc>> = None;
1834 for entry in schedule {
1835 let scheduled = if self.config.is_isolated {
1836 entry
1837 .isolated_margin_symbols
1838 .iter()
1839 .flatten()
1840 .any(|listed| listed == symbol)
1841 } else {
1842 entry
1843 .cross_margin_assets
1844 .iter()
1845 .flatten()
1846 .any(|listed| listed == base || listed == quote)
1847 };
1848 if !scheduled {
1849 continue;
1850 }
1851 let delisting_at = entry
1853 .delist_time
1854 .and_then(DateTime::from_timestamp_millis)
1855 .ok_or_else(|| {
1856 UnindexedClientError::Internal(format!(
1857 "decoding Binance response: delist-schedule lists {symbol} with delistTime \
1858 {:?}",
1859 entry.delist_time
1860 ))
1861 })?;
1862 earliest = Some(earliest.map_or(delisting_at, |earliest| earliest.min(delisting_at)));
1863 }
1864 Ok(earliest)
1865 }
1866
1867 async fn fetch_next_hourly_interest_rate(
1871 &self,
1872 asset: &AssetNameExchange,
1873 ) -> Result<Option<Decimal>, UnindexedClientError> {
1874 let name = asset.name().to_string();
1875 let is_isolated = self.config.is_isolated;
1876 let response =
1877 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1878 let name = name.clone();
1879 Box::pin(async move {
1880 let params = GetFutureHourlyInterestRateParams::builder(
1881 name,
1882 IsolatedFlag::from_flag(is_isolated),
1883 )
1884 .build()?;
1885 rest.get_future_hourly_interest_rate(params).await
1886 })
1887 })
1888 .await
1889 .map_err(|e| classify_rest_query_error(&e, None))?;
1890 let rate = response
1891 .data()
1892 .await
1893 .map_err(response_decode_error)?
1894 .into_iter()
1895 .find(|listed| listed.asset.as_deref() == Some(name.as_str()))
1896 .and_then(|listed| listed.next_hourly_interest_rate);
1897 optional_decimal("nextHourlyInterestRate", rate.as_deref())
1898 }
1899}
1900
1901fn optional_decimal(
1904 field: &str,
1905 value: Option<&str>,
1906) -> Result<Option<Decimal>, UnindexedClientError> {
1907 value
1908 .map(|value| {
1909 Decimal::from_str(value).map_err(|error| {
1910 UnindexedClientError::Internal(format!(
1911 "decoding Binance response: {field} {value:?}: {error}"
1912 ))
1913 })
1914 })
1915 .transpose()
1916}
1917
1918const TOKEN_RENEW_MARGIN_SECS: i64 = 300; const TOKEN_MIN_LIFETIME_SECS: u64 = 60;
1926
1927struct UserListenToken {
1929 token: String,
1930 expiration_time_ms: i64,
1931}
1932
1933fn build_listen_token_query(
1937 symbol: Option<&InstrumentNameExchange>,
1938) -> BTreeMap<String, serde_json::Value> {
1939 let mut query = BTreeMap::new();
1940 if let Some(sym) = symbol {
1941 query.insert(
1942 "isIsolated".to_string(),
1943 serde_json::Value::String("TRUE".to_string()),
1944 );
1945 query.insert(
1946 "symbol".to_string(),
1947 serde_json::Value::String(sym.name().to_string()),
1948 );
1949 }
1950 query
1951}
1952
1953#[derive(Deserialize)]
1955struct UserListenTokenResponse {
1956 #[serde(alias = "listenToken")]
1958 token: String,
1959 #[serde(rename = "expirationTime")]
1960 expiration_time: i64,
1961}
1962
1963async fn acquire_user_listen_token(
1978 rest_config: &Arc<ConfigurationRestApi>,
1979 rate_limiter: &RateLimitTracker,
1980 symbol: Option<&InstrumentNameExchange>,
1981) -> Result<UserListenToken, UnindexedClientError> {
1982 let query = build_listen_token_query(symbol);
1985 let response = rest_call_with_retry(rest_config, rate_limiter, RequestKind::Essential, |cfg| {
1986 let query = query.clone();
1987 Box::pin(async move {
1988 binance_sdk::common::utils::send_request::<UserListenTokenResponse>(
1989 &cfg,
1990 "/sapi/v1/userListenToken",
1991 reqwest::Method::POST,
1992 query,
1995 BTreeMap::new(),
1996 None,
1997 true, )
1999 .await
2000 })
2001 })
2002 .await
2003 .map_err(|e| classify_rest_query_error(&e, symbol))?;
2004
2005 let data = response.data().await.map_err(response_decode_error)?;
2006 Ok(UserListenToken {
2007 token: data.token,
2008 expiration_time_ms: data.expiration_time,
2009 })
2010}
2011
2012fn token_renew_after(expiration_time_ms: i64) -> Duration {
2015 let now_ms = Utc::now().timestamp_millis();
2016 let remaining_secs = expiration_time_ms
2020 .saturating_sub(now_ms)
2021 .saturating_div(1_000)
2022 .saturating_sub(TOKEN_RENEW_MARGIN_SECS);
2023 let secs = u64::try_from(remaining_secs).unwrap_or(0);
2024 if secs < TOKEN_MIN_LIFETIME_SECS {
2027 warn!(
2028 computed_secs = secs,
2029 floor_secs = TOKEN_MIN_LIFETIME_SECS,
2030 "BinanceMargin token renewal wait clamped to floor (near-expiry or odd expirationTime)"
2031 );
2032 }
2033 Duration::from_secs(secs.max(TOKEN_MIN_LIFETIME_SECS))
2034}
2035
2036async fn connect_margin_ws(
2043 ws_config: &ConfigurationWebsocketApi,
2044) -> anyhow::Result<DisconnectOnDrop<Arc<WsApiBase>>> {
2045 let ws = WsApiBase::new(ws_config.clone(), vec![]);
2046 tokio::time::timeout(
2049 Duration::from_secs(CONNECT_TIMEOUT_SECS),
2050 ws.clone().connect(),
2051 )
2052 .await
2053 .map_err(|_| {
2054 anyhow::anyhow!("BinanceMargin WS connect timed out after {CONNECT_TIMEOUT_SECS}s")
2055 })??;
2056 Ok(DisconnectOnDrop::new(ws, "BinanceMargin account stream"))
2058}
2059
2060async fn subscribe_listen_token(ws: &Arc<WsApiBase>, token: &str) -> anyhow::Result<()> {
2067 let mut payload = BTreeMap::new();
2070 payload.insert(
2071 "listenToken".to_string(),
2072 serde_json::Value::String(token.to_string()),
2073 );
2074 ws.send_message::<serde_json::Value>(
2075 "userDataStream.subscribe.listenToken",
2076 payload,
2077 WebsocketMessageSendOptions::new(),
2078 )
2079 .await
2080 .map_err(subscribe_error)?;
2081 Ok(())
2082}
2083
2084fn subscribe_error(e: WebsocketError) -> anyhow::Error {
2087 let message = format!("userDataStream.subscribe.listenToken failed: {e}");
2088 anyhow::Error::new(e).context(message)
2089}
2090
2091async fn subscribe_listen_token_capture(
2104 ws: &Arc<WsApiBase>,
2105 token: &str,
2106) -> anyhow::Result<Option<i64>> {
2107 #[derive(Deserialize)]
2110 struct SubscribeAck {
2111 #[serde(rename = "subscriptionId", default)]
2112 subscription_id: Option<i64>,
2113 }
2114
2115 let mut payload = BTreeMap::new();
2116 payload.insert(
2117 "listenToken".to_string(),
2118 serde_json::Value::String(token.to_string()),
2119 );
2120 let result = ws
2121 .send_message::<SubscribeAck>(
2122 "userDataStream.subscribe.listenToken",
2123 payload,
2124 WebsocketMessageSendOptions::new(),
2125 )
2126 .await
2127 .map_err(subscribe_error)?;
2128 let ack = match result {
2129 SendWebsocketMessageResult::Single(resp) => resp.data()?,
2130 SendWebsocketMessageResult::Multiple(responses) => responses
2132 .into_iter()
2133 .next()
2134 .ok_or_else(|| anyhow::anyhow!("empty subscribe ack"))?
2135 .data()?,
2136 };
2137 Ok(ack.subscription_id)
2138}
2139
2140fn cross_account_position_handler(
2144 position: OutboundAccountPosition,
2145 _subscription_id: Option<i64>,
2146 buf: &mut Vec<UnindexedAccountEvent>,
2147) {
2148 convert_margin_account_position(position, buf);
2149}
2150
2151fn cross_notice_instrument(_subscription_id: Option<i64>) -> Option<InstrumentNameExchange> {
2155 None
2156}
2157
2158fn isolated_notice_instrument(
2164 subscription_id: Option<i64>,
2165 sub_map: &Mutex<HashMap<i64, InstrumentNameExchange>>,
2166) -> Option<InstrumentNameExchange> {
2167 let Some(sub_id) = subscription_id else {
2168 warn!(
2169 "BinanceMargin isolated: MARGIN_LEVEL_STATUS_CHANGE without subscriptionId — \
2170 delivering the notice without an instrument"
2171 );
2172 return None;
2173 };
2174 let instrument = sub_map
2175 .lock()
2176 .unwrap_or_else(|p| p.into_inner())
2177 .get(&sub_id)
2178 .cloned();
2179 if instrument.is_none() {
2180 warn!(
2181 subscription_id = sub_id,
2182 "BinanceMargin isolated: MARGIN_LEVEL_STATUS_CHANGE for unmapped subscriptionId — \
2183 delivering the notice without an instrument"
2184 );
2185 }
2186 instrument
2187}
2188
2189#[cfg(test)]
2195fn convert_margin_user_data_events(frame: &str, buf: &mut Vec<UnindexedAccountEvent>) -> bool {
2196 matches!(
2197 convert_margin_user_data_events_with(
2198 frame,
2199 buf,
2200 &mut cross_account_position_handler,
2201 &mut cross_notice_instrument,
2202 ),
2203 FrameOutcome::Pushed { terminated: true }
2204 )
2205}
2206
2207#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2209enum FrameOutcome {
2210 Response,
2212 Pushed { terminated: bool },
2214}
2215
2216fn convert_margin_user_data_events_with(
2234 frame: &str,
2235 buf: &mut Vec<UnindexedAccountEvent>,
2236 handle_position: &mut impl FnMut(
2237 OutboundAccountPosition,
2238 Option<i64>,
2239 &mut Vec<UnindexedAccountEvent>,
2240 ),
2241 notice_instrument: &mut impl FnMut(Option<i64>) -> Option<InstrumentNameExchange>,
2242) -> FrameOutcome {
2243 let (subscription_id, event_type, event_raw) = match parse_user_data_frame(frame) {
2244 UserDataFrame::Response => return FrameOutcome::Response,
2246 UserDataFrame::Event {
2247 subscription_id,
2248 event_type,
2249 event,
2250 } => (subscription_id, event_type, event),
2251 UserDataFrame::Unrecognised => {
2252 static SEEN: AtomicU64 = AtomicU64::new(0);
2253 log_unrecognised_frame("BinanceMargin", &SEEN, frame);
2254 return FrameOutcome::Pushed { terminated: false };
2255 }
2256 };
2257 let terminated = match event_type {
2258 "executionReport" => {
2259 match serde_json::from_str::<ExecutionReport>(event_raw) {
2262 Ok(report) => {
2263 debug!(
2267 symbol = report.symbol.as_deref().unwrap_or("?"),
2268 order_id = ?report.order_id,
2269 client_order_id = report.client_order_id.as_deref().unwrap_or("?"),
2270 order_type = report.order_type.as_deref().unwrap_or("?"),
2271 execution_type = report.execution_type.as_deref().unwrap_or("?"),
2272 order_status = report.order_status.as_deref().unwrap_or("?"),
2273 "BinanceMargin executionReport"
2274 );
2275 convert_execution_report(&report, ExchangeId::BinanceMargin, buf);
2276 }
2277 Err(e) => {
2278 warn!(error = %e, frame = frame_excerpt(event_raw), "BinanceMargin: undeserializable executionReport, dropping")
2279 }
2280 }
2281 false
2282 }
2283 "outboundAccountPosition" => {
2284 match serde_json::from_str::<OutboundAccountPosition>(event_raw) {
2285 Ok(position) => handle_position(position, subscription_id, buf),
2287 Err(e) => {
2288 warn!(error = %e, frame = frame_excerpt(event_raw), "BinanceMargin: undeserializable outboundAccountPosition, dropping")
2289 }
2290 }
2291 false
2292 }
2293 "balanceUpdate" => {
2294 false
2298 }
2299 "USER_LIABILITY_CHANGE" | "userLiabilityChange" => {
2303 match serde_json::from_str::<UserLiabilityChange>(event_raw) {
2310 Ok(c) => log_liability_change(&c, "user data"),
2311 Err(e) => warn!(
2312 error = %e,
2313 frame = frame_excerpt(event_raw),
2314 "BinanceMargin: undeserializable USER_LIABILITY_CHANGE, dropping"
2315 ),
2316 }
2317 false
2318 }
2319 "MARGIN_LEVEL_STATUS_CHANGE" | "marginLevelStatusChange" => {
2320 match serde_json::from_str::<MarginLevelStatusChange>(event_raw) {
2323 Ok(change) => {
2324 log_margin_level_change(&change, "user data");
2325 buf.extend(margin_level_notice(
2326 change,
2327 notice_instrument(subscription_id),
2328 ));
2329 }
2330 Err(e) => warn!(
2331 error = %e,
2332 frame = frame_excerpt(event_raw),
2333 "BinanceMargin: undeserializable MARGIN_LEVEL_STATUS_CHANGE, dropping"
2334 ),
2335 }
2336 false
2337 }
2338 "eventStreamTerminated" => {
2339 warn!("BinanceMargin user data stream terminated by exchange, signalling reconnect");
2340 true
2341 }
2342 "listStatus" => {
2343 trace!("BinanceMargin ignoring listStatus user data event");
2346 false
2347 }
2348 other => {
2349 static SEEN: UnhandledEvents = UnhandledEvents::new();
2350 log_unhandled_event("BinanceMargin", &SEEN, other, event_raw);
2351 false
2352 }
2353 };
2354 FrameOutcome::Pushed { terminated }
2355}
2356
2357fn convert_margin_account_position(
2362 position: OutboundAccountPosition,
2363 buf: &mut Vec<UnindexedAccountEvent>,
2364) {
2365 let time_exchange = position
2366 .u
2367 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
2368 .unwrap_or_else(Utc::now);
2369
2370 for b in position.b_uppercase.unwrap_or_default() {
2371 let asset = match b.a {
2372 Some(a) => AssetNameExchange::new(a),
2373 None => {
2374 warn!("BinanceMargin account position entry missing asset name");
2375 continue;
2376 }
2377 };
2378 let free = match b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2379 Some(v) => v,
2380 None => {
2381 warn!(%asset, "BinanceMargin account position missing/unparseable 'free' field");
2382 continue;
2383 }
2384 };
2385 let locked = match b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2386 Some(v) => v,
2387 None => {
2388 warn!(%asset, "BinanceMargin account position missing/unparseable 'locked' field");
2389 continue;
2390 }
2391 };
2392 let update =
2393 AssetBalanceUpdate::new(asset, BalanceUpdate::new(free, locked), time_exchange);
2394 buf.push(UnindexedAccountEvent::new(
2395 ExchangeId::BinanceMargin,
2396 AccountEventKind::BalanceStreamUpdate(
2397 rustrade_integration::collection::snapshot::Snapshot::new(update),
2398 ),
2399 ));
2400 }
2401}
2402
2403fn route_isolated_account_position(
2419 position: OutboundAccountPosition,
2420 subscription_id: Option<i64>,
2421 sub_map: &Mutex<HashMap<i64, InstrumentNameExchange>>,
2422 base_quote: &HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>,
2423 buf: &mut Vec<UnindexedAccountEvent>,
2424) {
2425 let Some(sub_id) = subscription_id else {
2426 warn!(
2427 "BinanceMargin isolated: outboundAccountPosition without subscriptionId — dropping \
2428 (per-instrument balance routing requires it)"
2429 );
2430 return;
2431 };
2432 let instrument = {
2433 let map = sub_map.lock().unwrap_or_else(|p| p.into_inner());
2436 match map.get(&sub_id) {
2437 Some(inst) => inst.clone(),
2438 None => {
2439 warn!(
2440 subscription_id = sub_id,
2441 "BinanceMargin isolated: outboundAccountPosition for unmapped subscriptionId — dropping"
2442 );
2443 return;
2444 }
2445 }
2446 };
2447 let Some((base_asset, quote_asset)) = base_quote.get(&instrument) else {
2448 warn!(
2449 instrument = %instrument.name(),
2450 "BinanceMargin isolated: no base/quote known for instrument — dropping balance frame"
2451 );
2452 return;
2453 };
2454
2455 let time_exchange = position
2456 .u
2457 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
2458 .unwrap_or_else(Utc::now);
2459
2460 for b in position.b_uppercase.unwrap_or_default() {
2461 let Some(asset) = b.a.as_deref() else {
2462 warn!(instrument = %instrument.name(), "BinanceMargin isolated account position entry missing asset name");
2463 continue;
2464 };
2465 if asset != base_asset.name().as_str() && asset != quote_asset.name().as_str() {
2467 debug!(%asset, instrument = %instrument.name(), "BinanceMargin isolated account position: skipping asset outside the pair");
2468 continue;
2469 }
2470 let (Some(free), Some(locked)) = (
2471 b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()),
2472 b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()),
2473 ) else {
2474 warn!(%asset, instrument = %instrument.name(), "BinanceMargin isolated account position missing/unparseable free/locked");
2475 continue;
2476 };
2477 buf.push(UnindexedAccountEvent::new(
2478 ExchangeId::BinanceMargin,
2479 AccountEventKind::InstrumentBalanceUpdate(InstrumentBalanceUpdate::new(
2480 instrument.clone(),
2481 AssetBalanceUpdate::new(
2482 AssetNameExchange::new(asset),
2483 BalanceUpdate::new(free, locked),
2484 time_exchange,
2485 ),
2486 )),
2487 ));
2488 }
2489}
2490
2491fn register_user_data_listener(
2504 ws: &Arc<WsApiBase>,
2505 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
2506 dedup: SharedDedupCache,
2507 known: SharedKnownLiveOrders,
2508 heartbeat_flag: Arc<AtomicBool>,
2509 signal_tx: oneshot::Sender<()>,
2510 mut handle_position: impl FnMut(
2511 OutboundAccountPosition,
2512 Option<i64>,
2513 &mut Vec<UnindexedAccountEvent>,
2514 ) + Send
2515 + 'static,
2516 mut notice_instrument: impl FnMut(Option<i64>) -> Option<InstrumentNameExchange> + Send + 'static,
2517) -> UnsubscribeOnDrop {
2518 let mut signal_tx_opt = Some(signal_tx);
2519 let mut event_tx = Some(tx);
2520 let mut event_buf = Vec::with_capacity(32);
2524
2525 UnsubscribeOnDrop::new(ws.common.events.subscribe(move |event| {
2529 let Some(ref sender) = event_tx else { return };
2530 match event {
2531 WebsocketEvent::Message(json_str) => {
2532 let terminated = match convert_margin_user_data_events_with(
2535 &json_str,
2536 &mut event_buf,
2537 &mut handle_position,
2538 &mut notice_instrument,
2539 ) {
2540 FrameOutcome::Response => return,
2543 FrameOutcome::Pushed { terminated } => {
2544 heartbeat_flag.store(true, Ordering::Release);
2545 terminated
2546 }
2547 };
2548 for ev in event_buf.drain(..) {
2549 if let Some(key) = dedup_key_from_event(&ev)
2550 && is_duplicate(&dedup, key)
2551 {
2552 trace!("BinanceMargin dedup: skipping duplicate event");
2553 continue;
2554 }
2555 let held = KnownLiveOrders::observes(&ev.kind).then(|| {
2559 let mut held = known.lock();
2560 held.observe(&ev);
2561 held
2562 });
2563 if sender.send(ev).is_err() {
2564 drop(held);
2565 warn!("BinanceMargin account_stream receiver dropped, suppressing sends");
2566 event_tx.take();
2567 if let Some(s) = signal_tx_opt.take() {
2568 let _ = s.send(());
2569 }
2570 return;
2571 }
2572 }
2573 if terminated {
2574 event_tx.take();
2575 if let Some(s) = signal_tx_opt.take() {
2576 let _ = s.send(());
2577 }
2578 }
2579 }
2580 WebsocketEvent::Ping | WebsocketEvent::Pong => {
2581 heartbeat_flag.store(true, Ordering::Release);
2582 }
2583 WebsocketEvent::Error(e) => {
2584 warn!(%e, "BinanceMargin WebSocket error, will attempt reconnect");
2585 event_tx.take();
2586 if let Some(s) = signal_tx_opt.take() {
2587 let _ = s.send(());
2588 }
2589 }
2590 WebsocketEvent::Close(code, reason) => {
2591 warn!(code, %reason, "BinanceMargin WebSocket closed");
2592 event_tx.take();
2593 if let Some(s) = signal_tx_opt.take() {
2594 let _ = s.send(());
2595 }
2596 }
2597 _ => {
2598 trace!("BinanceMargin ignoring unhandled WebsocketEvent variant");
2599 }
2600 }
2601 }))
2602}
2603
2604async fn recover_margin_fills(
2626 rest: &Arc<RestApi>,
2627 rate_limiter: &Arc<RateLimitTracker>,
2628 unrecovered: &mut UnrecoveredFills,
2629 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
2630 dedup: &SharedDedupCache,
2631 known: &SharedKnownLiveOrders,
2632 is_isolated: bool,
2633) {
2634 use futures::StreamExt as _;
2635
2636 let due = unrecovered.due(tokio::time::Instant::now());
2637 let Some(oldest) = due.iter().map(|(_, gap)| gap.start_ms).min() else {
2638 return;
2639 };
2640 info!(
2641 gaps = due.len(),
2642 oldest = %gap_time(oldest),
2643 is_isolated,
2644 "BinanceMargin recovering fills missed while disconnected"
2645 );
2646
2647 let mut settled = vec![false; due.len()];
2650 let started = AtomicUsize::new(0);
2651 let recovery = async {
2652 let order_executions_deadline = tokio::time::Instant::now() + ORDER_EXECUTIONS_BUDGET;
2653 let mut recovered = 0u32;
2654 let mut duplicates = 0u32;
2655
2656 let mut stream =
2657 futures::stream::iter(due.iter().cloned().enumerate().map(|(index, (inst, gap))| {
2658 started.store(index + 1, Ordering::Relaxed);
2659 let rest = rest.clone();
2660 let rl = rate_limiter.clone();
2661 async move {
2662 let span = MyTradesFrom::Span {
2663 start: gap.start_ms,
2664 end: gap.end_ms,
2665 };
2666 let raw = match paginate_margin_my_trades(
2667 &rest,
2668 &rl,
2669 &inst,
2670 span,
2671 is_isolated,
2672 RequestKind::Essential,
2673 )
2674 .await
2675 {
2676 Ok(pages) => pages,
2677 Err(e) => return (index, Err(e)),
2678 };
2679 let totals = recovered_order_totals(
2683 ExchangeId::BinanceMargin,
2684 &inst,
2685 &raw,
2686 order_executions_deadline,
2687 |order_id| {
2688 paginate_margin_my_trades(
2689 &rest,
2690 &rl,
2691 &inst,
2692 MyTradesFrom::Order(order_id),
2693 is_isolated,
2694 RequestKind::Essential,
2695 )
2696 },
2697 )
2698 .await;
2699 let cids = recovered_order_cids(
2702 ExchangeId::BinanceMargin,
2703 &inst,
2704 &raw,
2705 order_executions_deadline,
2706 |order_id| fetch_margin_order_cid(&rest, &rl, &inst, order_id, is_isolated),
2707 )
2708 .await;
2709 let trades = raw
2710 .iter()
2711 .filter_map(|t| {
2712 let mut trade = convert_margin_trade(t, &inst)?;
2713 trade.order_filled_quantity =
2714 t.id.and_then(|id| totals.get(&id).copied());
2715 trade.client_order_id =
2716 t.order_id.and_then(|id| cids.get(&id).cloned());
2717 Some(trade)
2718 })
2719 .collect::<Vec<_>>();
2720 (index, Ok(trades))
2721 }
2722 }))
2723 .buffer_unordered(8);
2724 while let Some((index, result)) = stream.next().await {
2725 let (inst, gap) = &due[index];
2726 settled[index] = true;
2727 let trades = match result {
2728 Ok(trades) => trades,
2729 Err(e) => {
2730 gap_failed(
2731 unrecovered,
2732 tx,
2733 ExchangeId::BinanceMargin,
2734 inst,
2735 gap,
2736 FillRecoveryFailure::Request(e.to_string()),
2737 );
2738 continue;
2739 }
2740 };
2741 for trade in trades {
2742 let event = UnindexedAccountEvent::new(
2743 ExchangeId::BinanceMargin,
2744 AccountEventKind::Trade(trade),
2745 );
2746 if let Some(key) = dedup_key_from_event(&event)
2747 && is_duplicate(dedup, key)
2748 {
2749 duplicates += 1;
2750 continue;
2751 }
2752 let sent = {
2755 let mut known = known.lock();
2756 known.observe(&event);
2757 tx.send(event)
2758 };
2759 if sent.is_err() {
2760 debug!("BinanceMargin fill recovery: consumer dropped during recovery");
2761 return;
2762 }
2763 recovered += 1;
2764 }
2765 unrecovered.recovered(inst, gap);
2766 }
2767 info!(
2768 recovered,
2769 duplicates, is_isolated, "BinanceMargin fill recovery complete"
2770 );
2771 };
2772 if tokio::time::timeout(Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS), recovery)
2776 .await
2777 .is_err()
2778 {
2779 let started = started.load(Ordering::Relaxed);
2780 for ((inst, gap), _) in due[..started]
2781 .iter()
2782 .zip(&settled)
2783 .filter(|(_, settled)| !**settled)
2784 {
2785 gap_failed(
2786 unrecovered,
2787 tx,
2788 ExchangeId::BinanceMargin,
2789 inst,
2790 gap,
2791 FillRecoveryFailure::TimedOut {
2792 timeout_secs: FILL_RECOVERY_TIMEOUT_SECS,
2793 },
2794 );
2795 }
2796 }
2797}
2798
2799fn open_reconnect_recovery(
2804 unrecovered: &mut UnrecoveredFills,
2805 unchecked: &mut UncheckedOrders,
2806 known: &SharedKnownLiveOrders,
2807 instruments: &[InstrumentNameExchange],
2808 disconnected: DateTime<Utc>,
2809) {
2810 unrecovered.open(instruments, disconnected, Utc::now());
2811 let held = known.lock().instruments_among(instruments);
2812 unchecked.open(held);
2813}
2814
2815#[allow(
2823 clippy::cognitive_complexity,
2824 reason = "inherent reconnect-loop complexity (token + connect + subscribe + callback + recovery \
2825 + monitor + cleanup + backoff); mirrors spot's `connection_manager`, not worth splitting further"
2826)]
2827async fn margin_connection_manager(
2828 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
2829 dedup: SharedDedupCache,
2830 ws_config: ConfigurationWebsocketApi,
2831 rest_config: Arc<ConfigurationRestApi>,
2832 rest: Arc<RestApi>,
2833 rate_limiter: Arc<RateLimitTracker>,
2834 known: SharedKnownLiveOrders,
2835 instruments: Vec<InstrumentNameExchange>,
2836 initial: Option<(DisconnectOnDrop<Arc<WsApiBase>>, UserListenToken)>,
2837) {
2838 enum DisconnectReason {
2839 Signal,
2840 HeartbeatTimeout,
2841 TokenRefresh,
2842 ConsumerDropped,
2843 }
2844
2845 let mut budget = ReconnectBudget::new("Binance", BackoffLimits::ACCOUNT_STREAM);
2846 let mut disconnect_time: Option<DateTime<Utc>> = None;
2847 let mut unrecovered = UnrecoveredFills::default();
2849 let mut unchecked = UncheckedOrders::default();
2851 let (mut current_ws, mut current_token) = match initial {
2852 Some((ws, token)) => (Some(ws), Some(token)),
2853 None => (None, None),
2854 };
2855
2856 loop {
2857 let token = match current_token.take() {
2859 Some(t) => t,
2860 None => {
2861 if let Err(reason) = rate_limiter.before_stream_connect().await {
2864 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
2865 break;
2866 }
2867 match acquire_user_listen_token(&rest_config, &rate_limiter, None).await {
2869 Ok(t) => t,
2870 Err(e) => {
2871 error!(%e, "BinanceMargin userListenToken acquisition failed");
2872 if let Err(reason) =
2873 retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await
2874 {
2875 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
2876 break;
2877 }
2878 continue;
2879 }
2880 }
2881 }
2882 };
2883
2884 let ws = match current_ws.take() {
2886 Some(ws) => ws,
2887 None => match connect_margin_ws(&ws_config).await {
2888 Ok(ws) => ws,
2889 Err(e) => {
2890 error!(%e, "BinanceMargin WS connect failed");
2891 rate_limiter.on_stream_refused(&e);
2892 if let Err(reason) =
2893 retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await
2894 {
2895 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
2896 break;
2897 }
2898 continue;
2899 }
2900 },
2901 };
2902
2903 let (signal_tx, signal_rx) = oneshot::channel::<()>();
2905 let heartbeat_flag = Arc::new(AtomicBool::new(false));
2908 let subscription = register_user_data_listener(
2911 &ws,
2912 tx.clone(),
2913 dedup.clone(),
2914 known.clone(),
2915 heartbeat_flag.clone(),
2916 signal_tx,
2917 cross_account_position_handler,
2918 cross_notice_instrument,
2919 );
2920
2921 if let Err(e) = subscribe_listen_token(&ws, &token.token).await {
2923 warn!(%e, "BinanceMargin user-data subscribe failed, cleaning up and retrying");
2924 subscription.unsubscribe();
2925 if let Err(de) = ws.disconnect().await {
2926 warn!(%de, "BinanceMargin failed to disconnect after subscribe failure");
2927 }
2928 rate_limiter.on_stream_refused(&e);
2929 if let Err(reason) = retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await {
2931 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
2932 break;
2933 }
2934 continue;
2935 }
2936 info!("BinanceMargin account_stream connected and subscribed");
2937 let token_deadline =
2944 tokio::time::Instant::now() + token_renew_after(token.expiration_time_ms);
2945
2946 if let Some(dt) = disconnect_time.take() {
2951 open_reconnect_recovery(&mut unrecovered, &mut unchecked, &known, &instruments, dt);
2952 }
2953 recover_margin_fills(
2956 &rest,
2957 &rate_limiter,
2958 &mut unrecovered,
2959 &tx,
2960 &dedup,
2961 &known,
2962 false,
2963 )
2964 .await;
2965 recover_margin_ended_orders(
2966 &rest,
2967 &rate_limiter,
2968 &known,
2969 &mut unchecked,
2970 &unrecovered,
2971 &tx,
2972 false,
2973 )
2974 .await;
2975
2976 let reason = {
2978 let mut signal_rx = signal_rx;
2979 let retry_gaps = async {
2985 loop {
2986 let next_check = unchecked.next_due(&unrecovered);
2987 match unrecovered.next_due().into_iter().chain(next_check).min() {
2988 Some(due) => tokio::time::sleep_until(due).await,
2989 None => std::future::pending::<()>().await,
2990 }
2991 recover_margin_fills(
2992 &rest,
2993 &rate_limiter,
2994 &mut unrecovered,
2995 &tx,
2996 &dedup,
2997 &known,
2998 false,
2999 )
3000 .await;
3001 recover_margin_ended_orders(
3002 &rest,
3003 &rate_limiter,
3004 &known,
3005 &mut unchecked,
3006 &unrecovered,
3007 &tx,
3008 false,
3009 )
3010 .await;
3011 }
3012 };
3013 tokio::pin!(retry_gaps);
3014 let mut grace = true;
3016 loop {
3017 tokio::select! {
3018 biased;
3021 _ = tx.closed() => {
3022 debug!("BinanceMargin account_stream consumer dropped, terminating");
3023 break DisconnectReason::ConsumerDropped;
3024 }
3025 _ = &mut signal_rx => {
3026 warn!("BinanceMargin WS disconnected, will attempt reconnect");
3027 break DisconnectReason::Signal;
3028 }
3029 () = tokio::time::sleep_until(token_deadline) => {
3030 info!("BinanceMargin userListenToken nearing expiry, renewing");
3031 break DisconnectReason::TokenRefresh;
3032 }
3033 () = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
3034 let in_grace = std::mem::take(&mut grace);
3035 if heartbeat_flag.swap(false, Ordering::AcqRel) {
3036 budget.reset();
3039 continue;
3040 }
3041 if in_grace {
3042 continue;
3045 }
3046 warn!("BinanceMargin heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
3047 break DisconnectReason::HeartbeatTimeout;
3048 }
3049 () = &mut retry_gaps => {}
3051 }
3052 }
3053 };
3054 let should_reconnect = !matches!(reason, DisconnectReason::ConsumerDropped);
3055
3056 if should_reconnect {
3057 disconnect_time = Some(match reason {
3058 DisconnectReason::HeartbeatTimeout => {
3059 Utc::now()
3060 - chrono::Duration::seconds(HEARTBEAT_TIMEOUT_SECS as i64)
3061 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3062 }
3063 DisconnectReason::TokenRefresh => {
3064 Utc::now()
3069 - chrono::Duration::seconds(CONNECT_TIMEOUT_SECS as i64)
3070 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3071 }
3072 _ => Utc::now() - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS),
3073 });
3074 }
3075
3076 subscription.unsubscribe();
3078 if let Err(e) = ws.disconnect().await {
3079 warn!(%e, "BinanceMargin failed to disconnect WebSocket");
3080 }
3081
3082 if !should_reconnect || tx.is_closed() {
3083 debug!("BinanceMargin connection manager exiting");
3085 break;
3086 }
3087
3088 if !matches!(reason, DisconnectReason::TokenRefresh) {
3091 let heartbeat_timed_out = matches!(reason, DisconnectReason::HeartbeatTimeout);
3094 let last_error = || disconnect_error(heartbeat_timed_out);
3095 if let Err(reason) = retry_or_end(&mut budget, &rate_limiter, last_error).await {
3096 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3097 break;
3098 }
3099 }
3100 }
3101}
3102
3103fn build_base_quote_map(
3115 assets: &[QueryIsolatedMarginAccountInfoResponseAssetsInner],
3116) -> HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)> {
3117 let mut map = HashMap::with_capacity(assets.len());
3118 for entry in assets {
3119 let (Some(symbol), Some(base), Some(quote)) = (
3120 entry.symbol.as_deref(),
3121 entry.base_asset.as_deref().and_then(|b| b.asset.as_deref()),
3122 entry
3123 .quote_asset
3124 .as_deref()
3125 .and_then(|q| q.asset.as_deref()),
3126 ) else {
3127 warn!(
3128 "BinanceMargin isolated: account-info entry missing symbol/base/quote asset — \
3129 skipping base/quote map entry"
3130 );
3131 continue;
3132 };
3133 map.insert(
3134 InstrumentNameExchange::new(symbol),
3135 (AssetNameExchange::new(base), AssetNameExchange::new(quote)),
3136 );
3137 }
3138 map
3139}
3140
3141async fn acquire_all_isolated_tokens(
3154 rest_config: &Arc<ConfigurationRestApi>,
3155 rate_limiter: &Arc<RateLimitTracker>,
3156 symbols: &[InstrumentNameExchange],
3157) -> Result<Vec<(InstrumentNameExchange, UserListenToken)>, UnindexedClientError> {
3158 use futures::{StreamExt as _, TryStreamExt as _};
3159
3160 futures::stream::iter(symbols.iter().cloned().map(|sym| {
3161 let rest_config = rest_config.clone();
3162 let rate_limiter = rate_limiter.clone();
3163 async move {
3164 let token = acquire_user_listen_token(&rest_config, &rate_limiter, Some(&sym)).await?;
3165 Ok::<_, UnindexedClientError>((sym, token))
3166 }
3167 }))
3168 .buffer_unordered(8)
3169 .try_collect()
3170 .await
3171}
3172
3173fn earliest_token_expiry_ms(tokens: &[(InstrumentNameExchange, UserListenToken)]) -> i64 {
3178 tokens
3179 .iter()
3180 .map(|(_, t)| t.expiration_time_ms)
3181 .min()
3182 .unwrap_or(i64::MAX)
3183}
3184
3185struct IsolatedLiveConn {
3188 ws: DisconnectOnDrop<Arc<WsApiBase>>,
3189 subscription: UnsubscribeOnDrop,
3190 signal_rx: oneshot::Receiver<()>,
3191 heartbeat_flag: Arc<AtomicBool>,
3192 earliest_expiry_ms: i64,
3195}
3196
3197async fn isolated_connect_and_subscribe(
3207 ws_config: &ConfigurationWebsocketApi,
3208 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3209 dedup: &SharedDedupCache,
3210 known: &SharedKnownLiveOrders,
3211 base_quote: &Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
3212 tokens: &[(InstrumentNameExchange, UserListenToken)],
3213) -> anyhow::Result<IsolatedLiveConn> {
3214 let ws = connect_margin_ws(ws_config).await?;
3215 let sub_map = Arc::new(Mutex::new(
3221 HashMap::<i64, InstrumentNameExchange>::with_capacity(tokens.len()),
3222 ));
3223 let (signal_tx, signal_rx) = oneshot::channel::<()>();
3224 let heartbeat_flag = Arc::new(AtomicBool::new(false));
3226
3227 let handle_position = {
3228 let sub_map = sub_map.clone();
3229 let base_quote = base_quote.clone();
3230 move |position, subscription_id, buf: &mut Vec<UnindexedAccountEvent>| {
3231 route_isolated_account_position(position, subscription_id, &sub_map, &base_quote, buf);
3232 }
3233 };
3234 let notice_instrument = {
3235 let sub_map = sub_map.clone();
3236 move |subscription_id| isolated_notice_instrument(subscription_id, &sub_map)
3237 };
3238 let subscription = register_user_data_listener(
3240 &ws,
3241 tx.clone(),
3242 dedup.clone(),
3243 known.clone(),
3244 heartbeat_flag.clone(),
3245 signal_tx,
3246 handle_position,
3247 notice_instrument,
3248 );
3249
3250 let earliest_expiry_ms = earliest_token_expiry_ms(tokens);
3251 for (sym, token) in tokens {
3252 match subscribe_listen_token_capture(&ws, &token.token).await {
3253 Ok(Some(id)) => {
3254 sub_map
3255 .lock()
3256 .unwrap_or_else(|p| p.into_inner())
3257 .insert(id, sym.clone());
3258 }
3259 Ok(None) => warn!(
3260 symbol = %sym.name(),
3261 "BinanceMargin isolated: subscribe ack carried no subscriptionId — live per-pair \
3262 balances unavailable for this symbol (fills/orders unaffected; balances via snapshot)"
3263 ),
3264 Err(e) => {
3265 warn!(%e, symbol = %sym.name(), "BinanceMargin isolated subscribe failed");
3268 subscription.unsubscribe();
3269 if let Err(de) = ws.disconnect().await {
3270 warn!(%de, "BinanceMargin isolated: disconnect after subscribe failure failed");
3271 }
3272 return Err(e);
3273 }
3274 }
3275 }
3276
3277 Ok(IsolatedLiveConn {
3278 ws,
3279 subscription,
3280 signal_rx,
3281 heartbeat_flag,
3282 earliest_expiry_ms,
3283 })
3284}
3285
3286#[allow(
3295 clippy::cognitive_complexity,
3296 reason = "inherent reconnect-loop complexity (tokens + connect + subscribe + monitor + cleanup \
3297 + backoff); mirrors the cross manager, not worth splitting further"
3298)]
3299async fn isolated_connection_manager(
3300 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
3301 dedup: SharedDedupCache,
3302 ws_config: ConfigurationWebsocketApi,
3303 rest_config: Arc<ConfigurationRestApi>,
3304 rest: Arc<RestApi>,
3305 rate_limiter: Arc<RateLimitTracker>,
3306 known: SharedKnownLiveOrders,
3307 symbols: Vec<InstrumentNameExchange>,
3308 base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
3309 initial: Option<IsolatedLiveConn>,
3310) {
3311 enum DisconnectReason {
3312 Signal,
3313 HeartbeatTimeout,
3314 TokenRefresh,
3315 ConsumerDropped,
3316 }
3317
3318 let mut budget = ReconnectBudget::new("Binance", BackoffLimits::ACCOUNT_STREAM);
3319 let mut disconnect_time: Option<DateTime<Utc>> = None;
3320 let mut unrecovered = UnrecoveredFills::default();
3322 let mut unchecked = UncheckedOrders::default();
3324 let mut current = initial;
3325
3326 loop {
3327 let IsolatedLiveConn {
3329 ws,
3330 subscription,
3331 signal_rx,
3332 heartbeat_flag,
3333 earliest_expiry_ms,
3334 } = match current.take() {
3335 Some(live) => live,
3337 None => {
3338 if let Err(reason) = rate_limiter.before_stream_connect().await {
3340 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3341 break;
3342 }
3343 let tokens = match acquire_all_isolated_tokens(
3346 &rest_config,
3347 &rate_limiter,
3348 &symbols,
3349 )
3350 .await
3351 {
3352 Ok(t) => t,
3353 Err(e) => {
3354 error!(%e, "BinanceMargin isolated userListenToken acquisition failed");
3355 if let Err(reason) =
3356 retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await
3357 {
3358 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3359 break;
3360 }
3361 continue;
3362 }
3363 };
3364 match isolated_connect_and_subscribe(
3365 &ws_config,
3366 &tx,
3367 &dedup,
3368 &known,
3369 &base_quote,
3370 &tokens,
3371 )
3372 .await
3373 {
3374 Ok(live) => live,
3375 Err(e) => {
3376 error!(%e, "BinanceMargin isolated connect/subscribe failed");
3377 rate_limiter.on_stream_refused(&e);
3378 if let Err(reason) =
3379 retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await
3380 {
3381 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3382 break;
3383 }
3384 continue;
3385 }
3386 }
3387 }
3388 };
3389
3390 info!(
3391 symbols = symbols.len(),
3392 "BinanceMargin isolated account_stream connected and subscribed"
3393 );
3394 let token_deadline = tokio::time::Instant::now() + token_renew_after(earliest_expiry_ms);
3400
3401 if let Some(dt) = disconnect_time.take() {
3404 open_reconnect_recovery(&mut unrecovered, &mut unchecked, &known, &symbols, dt);
3405 }
3406 recover_margin_fills(
3408 &rest,
3409 &rate_limiter,
3410 &mut unrecovered,
3411 &tx,
3412 &dedup,
3413 &known,
3414 true,
3415 )
3416 .await;
3417 recover_margin_ended_orders(
3418 &rest,
3419 &rate_limiter,
3420 &known,
3421 &mut unchecked,
3422 &unrecovered,
3423 &tx,
3424 true,
3425 )
3426 .await;
3427
3428 let reason = {
3431 let mut signal_rx = signal_rx;
3432 let retry_gaps = async {
3435 loop {
3436 let next_check = unchecked.next_due(&unrecovered);
3437 match unrecovered.next_due().into_iter().chain(next_check).min() {
3438 Some(due) => tokio::time::sleep_until(due).await,
3439 None => std::future::pending::<()>().await,
3440 }
3441 recover_margin_fills(
3442 &rest,
3443 &rate_limiter,
3444 &mut unrecovered,
3445 &tx,
3446 &dedup,
3447 &known,
3448 true,
3449 )
3450 .await;
3451 recover_margin_ended_orders(
3452 &rest,
3453 &rate_limiter,
3454 &known,
3455 &mut unchecked,
3456 &unrecovered,
3457 &tx,
3458 true,
3459 )
3460 .await;
3461 }
3462 };
3463 tokio::pin!(retry_gaps);
3464 let mut grace = true;
3466 loop {
3467 tokio::select! {
3468 biased;
3469 _ = tx.closed() => {
3470 debug!("BinanceMargin isolated consumer dropped, terminating");
3471 break DisconnectReason::ConsumerDropped;
3472 }
3473 _ = &mut signal_rx => {
3474 warn!("BinanceMargin isolated WS disconnected, will attempt reconnect");
3475 break DisconnectReason::Signal;
3476 }
3477 () = tokio::time::sleep_until(token_deadline) => {
3478 info!("BinanceMargin isolated userListenToken nearing expiry, renewing all");
3479 break DisconnectReason::TokenRefresh;
3480 }
3481 () = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
3482 let in_grace = std::mem::take(&mut grace);
3483 if heartbeat_flag.swap(false, Ordering::AcqRel) {
3484 budget.reset();
3486 continue;
3487 }
3488 if in_grace {
3489 continue;
3491 }
3492 warn!("BinanceMargin isolated heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
3493 break DisconnectReason::HeartbeatTimeout;
3494 }
3495 () = &mut retry_gaps => {}
3497 }
3498 }
3499 };
3500 let should_reconnect = !matches!(reason, DisconnectReason::ConsumerDropped);
3501
3502 if should_reconnect {
3503 disconnect_time = Some(match reason {
3504 DisconnectReason::HeartbeatTimeout => {
3505 Utc::now()
3506 - chrono::Duration::seconds(HEARTBEAT_TIMEOUT_SECS as i64)
3507 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3508 }
3509 DisconnectReason::TokenRefresh => {
3510 Utc::now()
3514 - chrono::Duration::seconds(CONNECT_TIMEOUT_SECS as i64)
3515 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3516 }
3517 _ => Utc::now() - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS),
3518 });
3519 }
3520
3521 subscription.unsubscribe();
3523 if let Err(e) = ws.disconnect().await {
3524 warn!(%e, "BinanceMargin isolated failed to disconnect WebSocket");
3525 }
3526
3527 if !should_reconnect || tx.is_closed() {
3528 debug!("BinanceMargin isolated connection manager exiting");
3530 break;
3531 }
3532
3533 if !matches!(reason, DisconnectReason::TokenRefresh) {
3536 let heartbeat_timed_out = matches!(reason, DisconnectReason::HeartbeatTimeout);
3539 let last_error = || disconnect_error(heartbeat_timed_out);
3540 if let Err(reason) = retry_or_end(&mut budget, &rate_limiter, last_error).await {
3541 emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
3542 break;
3543 }
3544 }
3545 }
3546}
3547
3548async fn fetch_margin_open_orders_for_instrument(
3555 rest: Arc<RestApi>,
3556 rate_limiter: Arc<RateLimitTracker>,
3557 instrument: InstrumentNameExchange,
3558 is_isolated: bool,
3559 kind: RequestKind,
3560) -> Result<(InstrumentNameExchange, OpenOrderListing), UnindexedClientError> {
3561 let symbol_str = instrument.name().to_string();
3563 let isolated = QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(is_isolated);
3564 let response = rest_call_with_retry(&rest, &rate_limiter, kind, |rest| {
3565 let sym = symbol_str.clone();
3566 let isolated = isolated.clone();
3567 Box::pin(async move {
3568 let params = QueryMarginAccountsOpenOrdersParams::builder()
3569 .symbol(sym)
3570 .is_isolated(isolated)
3571 .build()?;
3572 rest.query_margin_accounts_open_orders(params).await
3573 })
3574 })
3575 .await
3576 .map_err(|e| classify_rest_query_error(&e, Some(&instrument)))?;
3577
3578 let orders_data = response.data().await.map_err(response_decode_error)?;
3579
3580 let listing = convert_open_order_listing(&orders_data, ExchangeId::BinanceMargin, &instrument);
3581
3582 Ok((instrument, listing))
3583}
3584
3585async fn fetch_margin_all_open_orders(
3592 rest: Arc<RestApi>,
3593 rate_limiter: Arc<RateLimitTracker>,
3594 is_isolated: bool,
3595) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
3596 let isolated = QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(is_isolated);
3597 let response = rest_call_with_retry(&rest, &rate_limiter, RequestKind::Query, |rest| {
3598 let isolated = isolated.clone();
3599 Box::pin(async move {
3600 let params = QueryMarginAccountsOpenOrdersParams::builder()
3601 .is_isolated(isolated)
3602 .build()?;
3603 rest.query_margin_accounts_open_orders(params).await
3604 })
3605 })
3606 .await
3607 .map_err(|e| classify_rest_query_error(&e, None))?;
3608
3609 let orders_data = response.data().await.map_err(response_decode_error)?;
3610
3611 let orders = orders_data
3612 .into_iter()
3613 .filter_map(|o| convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin))
3614 .collect();
3615
3616 Ok(orders)
3617}
3618
3619async fn fetch_margin_order_lookup(
3626 rest: Arc<RestApi>,
3627 rate_limiter: Arc<RateLimitTracker>,
3628 key: UnindexedOrderKey,
3629 is_isolated: bool,
3630 kind: RequestKind,
3631 recorded_kind: Option<OrderKind>,
3632) -> Result<OrderLookup, UnindexedClientError> {
3633 let symbol = key.instrument.name().to_string();
3635 let cid = key.cid.0.to_string();
3636 let isolated = QueryMarginAccountsOrderIsIsolatedEnum::from_flag(is_isolated);
3637 let response = match rest_call_with_retry(&rest, &rate_limiter, kind, |rest| {
3638 let (symbol, cid, isolated) = (symbol.clone(), cid.clone(), isolated.clone());
3639 Box::pin(async move {
3640 let params = QueryMarginAccountsOrderParams::builder(symbol)
3641 .orig_client_order_id(cid)
3642 .is_isolated(isolated)
3643 .build()?;
3644 rest.query_margin_accounts_order(params).await
3645 })
3646 })
3647 .await
3648 {
3649 Ok(response) => response,
3650 Err(e) if is_unknown_order(&e) => {
3651 debug!(instrument = %key.instrument, cid = %key.cid, error = %e, "BinanceMargin does not know this order");
3652 return Ok(OrderLookup::Unknown);
3653 }
3654 Err(e) => return Err(classify_rest_query_error(&e, Some(&key.instrument))),
3655 };
3656
3657 let row = response.data().await.map_err(response_decode_error)?;
3658
3659 Ok(
3660 convert_ended_order(&row, ExchangeId::BinanceMargin, &key, recorded_kind)
3661 .map_or(OrderLookup::NotEnded, |order| {
3662 OrderLookup::Ended(Box::new(order))
3663 }),
3664 )
3665}
3666
3667async fn fetch_margin_order_cid(
3672 rest: &Arc<RestApi>,
3673 rate_limiter: &RateLimitTracker,
3674 instrument: &InstrumentNameExchange,
3675 order_id: i64,
3676 is_isolated: bool,
3677) -> Result<Option<ClientOrderId>, UnindexedClientError> {
3678 let symbol = instrument.name().to_string();
3679 let isolated = QueryMarginAccountsOrderIsIsolatedEnum::from_flag(is_isolated);
3680 let response = match rest_call_with_retry(rest, rate_limiter, RequestKind::Essential, |rest| {
3681 let (symbol, isolated) = (symbol.clone(), isolated.clone());
3682 Box::pin(async move {
3683 let params = QueryMarginAccountsOrderParams::builder(symbol)
3684 .order_id(order_id)
3685 .is_isolated(isolated)
3686 .build()?;
3687 rest.query_margin_accounts_order(params).await
3688 })
3689 })
3690 .await
3691 {
3692 Ok(response) => response,
3693 Err(e) if is_unknown_order(&e) => {
3694 debug!(%instrument, order_id, error = %e, "BinanceMargin does not know this order");
3695 return Ok(None);
3696 }
3697 Err(e) => return Err(classify_rest_query_error(&e, Some(instrument))),
3698 };
3699 let row = response.data().await.map_err(response_decode_error)?;
3700 Ok(row
3701 .client_order_id
3702 .filter(|cid| !cid.is_empty())
3703 .map(ClientOrderId::new))
3704}
3705
3706async fn listed_margin_open_cids(
3709 rest: Arc<RestApi>,
3710 rate_limiter: Arc<RateLimitTracker>,
3711 instruments: Vec<InstrumentNameExchange>,
3712 is_isolated: bool,
3713) -> Result<FnvHashSet<ClientOrderId>, UnindexedClientError> {
3714 let mut listed = FnvHashSet::default();
3715 for instrument in instruments {
3716 let (_, listing) = fetch_margin_open_orders_for_instrument(
3717 rest.clone(),
3718 rate_limiter.clone(),
3719 instrument,
3720 is_isolated,
3721 RequestKind::Essential,
3722 )
3723 .await?;
3724 listed.extend(listing.orders.into_iter().map(|order| order.key.cid));
3725 }
3726 Ok(listed)
3727}
3728
3729async fn recover_margin_ended_orders(
3732 rest: &Arc<RestApi>,
3733 rate_limiter: &Arc<RateLimitTracker>,
3734 known: &SharedKnownLiveOrders,
3735 unchecked: &mut UncheckedOrders,
3736 unrecovered: &UnrecoveredFills,
3737 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3738 is_isolated: bool,
3739) {
3740 recover_ended_orders(
3741 ExchangeId::BinanceMargin,
3742 known,
3743 unchecked,
3744 unrecovered,
3745 tx,
3746 OpenListing::PerInstrument,
3747 |instruments| {
3748 listed_margin_open_cids(rest.clone(), rate_limiter.clone(), instruments, is_isolated)
3749 },
3750 |key| {
3751 let recorded_kind = known.lock().kind_of(&key.cid);
3753 fetch_margin_order_lookup(
3754 rest.clone(),
3755 rate_limiter.clone(),
3756 key,
3757 is_isolated,
3758 RequestKind::Essential,
3759 recorded_kind,
3760 )
3761 },
3762 )
3763 .await;
3764}
3765
3766const MARGIN_MY_TRADES_WINDOW_MS: i64 = 23 * 60 * 60 * 1000;
3772
3773#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3776enum MarginTradesQuery {
3777 Window { start: i64, end: i64 },
3779 FromId(i64),
3781 Order(i64),
3783 OrderFromId { order_id: i64, from_id: i64 },
3785}
3786
3787impl MarginTradesQuery {
3788 fn first(from: MyTradesFrom, now_ms: i64) -> Option<Self> {
3792 match from {
3793 MyTradesFrom::Span { start, .. } => {
3794 (start <= now_ms).then(|| Self::window(start, now_ms))
3795 }
3796 MyTradesFrom::Order(order_id) => Some(Self::Order(order_id)),
3797 }
3798 }
3799
3800 fn window(start: i64, now_ms: i64) -> Self {
3802 Self::Window {
3803 start,
3804 end: start.saturating_add(MARGIN_MY_TRADES_WINDOW_MS).min(now_ms),
3805 }
3806 }
3807
3808 fn next(self, page_len: usize, last_id: Option<i64>, now_ms: i64) -> Option<Self> {
3811 let full = page_len >= BINANCE_MAX_TRADES;
3812 let after_last = || last_id?.checked_add(1);
3813 match self {
3814 Self::Window { end, .. } if page_len == 0 => {
3819 (end < now_ms).then(|| Self::window(end, now_ms))
3820 }
3821 Self::Window { end, .. } if end >= now_ms && !full => None,
3823 Self::Window { .. } => after_last().map(Self::FromId),
3826 Self::FromId(_) if full => after_last().map(Self::FromId),
3827 Self::Order(order_id) | Self::OrderFromId { order_id, .. } if full => {
3828 after_last().map(|from_id| Self::OrderFromId { order_id, from_id })
3829 }
3830 Self::FromId(_) | Self::Order(_) | Self::OrderFromId { .. } => None,
3831 }
3832 }
3833}
3834
3835async fn paginate_margin_my_trades(
3857 rest: &Arc<RestApi>,
3858 rate_limiter: &Arc<RateLimitTracker>,
3859 instrument: &InstrumentNameExchange,
3860 from: MyTradesFrom,
3861 is_isolated: bool,
3862 kind: RequestKind,
3863) -> Result<Vec<QueryMarginAccountsTradeListResponseInner>, UnindexedClientError> {
3864 let symbol_str = instrument.name().to_string();
3866 let isolated = QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(is_isolated);
3867 #[allow(clippy::cast_possible_wrap)]
3872 let limit = BINANCE_MAX_TRADES as i64;
3873 let now_ms = match from {
3876 MyTradesFrom::Span { end, .. } => end.min(Utc::now().timestamp_millis()),
3877 MyTradesFrom::Order(_) => Utc::now().timestamp_millis(),
3878 };
3879 let mut all_pages = Vec::new();
3880 let mut requests = 0u32;
3881 let mut next = MarginTradesQuery::first(from, now_ms);
3882 while let Some(query) = next {
3883 requests += 1;
3884 let response = rest_call_with_retry(rest, rate_limiter, kind, |rest| {
3885 let sym = symbol_str.clone();
3886 let isolated = isolated.clone();
3887 Box::pin(async move {
3888 let builder = QueryMarginAccountsTradeListParams::builder(sym)
3889 .is_isolated(isolated)
3890 .limit(limit);
3891 let params = match query {
3892 MarginTradesQuery::Window { start, end } => {
3893 builder.start_time(start).end_time(end)
3894 }
3895 MarginTradesQuery::FromId(id) => builder.from_id(id),
3896 MarginTradesQuery::Order(order_id) => builder.order_id(order_id),
3897 MarginTradesQuery::OrderFromId { order_id, from_id } => {
3898 builder.order_id(order_id).from_id(from_id)
3899 }
3900 }
3901 .build()?;
3902 rest.query_margin_accounts_trade_list(params).await
3903 })
3904 })
3905 .await
3906 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
3907
3908 let mut page = response.data().await.map_err(response_decode_error)?;
3909
3910 let page_len = page.len();
3911 let last_id = page.last().and_then(|t| t.id);
3912 let past_end = match from {
3914 MyTradesFrom::Span { end, .. } => drop_after(&mut page, end),
3915 MyTradesFrom::Order(_) => false,
3916 };
3917 all_pages.extend(page);
3918 if past_end {
3919 break;
3920 }
3921
3922 if page_len > 0 && last_id.is_none() {
3923 warn!(%instrument, "BinanceMargin paginate_my_trades: trade missing ID, cannot page past it");
3924 break;
3925 }
3926 next = query.next(page_len, last_id, now_ms);
3927 }
3928 debug!(
3929 %instrument,
3930 requests,
3931 trades = all_pages.len(),
3932 "BinanceMargin paginate_my_trades: walk complete"
3933 );
3934 Ok(all_pages)
3935}
3936
3937fn convert_margin_balance_entry(
3947 b: QueryCrossMarginAccountDetailsResponseUserAssetsInner,
3948 now: DateTime<Utc>,
3949) -> Option<AssetBalance<AssetNameExchange>> {
3950 let asset_name = AssetNameExchange::new(b.asset.as_deref()?);
3951 let free = match b.free.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
3952 Some(v) => v,
3953 None => {
3954 warn!(%asset_name, "BinanceMargin balance missing/unparseable 'free' field");
3955 return None;
3956 }
3957 };
3958 let locked = match b.locked.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
3959 Some(v) => v,
3960 None => {
3961 warn!(%asset_name, "BinanceMargin balance missing/unparseable 'locked' field");
3962 return None;
3963 }
3964 };
3965 let borrowed = b
3970 .borrowed
3971 .as_deref()
3972 .and_then(|s| Decimal::from_str(s).ok())
3973 .unwrap_or(Decimal::ZERO);
3974 let interest = b
3975 .interest
3976 .as_deref()
3977 .and_then(|s| Decimal::from_str(s).ok())
3978 .unwrap_or(Decimal::ZERO);
3979 Some(AssetBalance::new(
3980 asset_name,
3981 Balance::new_margin(free + locked, free, borrowed, interest),
3982 now,
3983 ))
3984}
3985
3986fn filter_and_convert_margin_balances(
3989 user_assets: Vec<QueryCrossMarginAccountDetailsResponseUserAssetsInner>,
3990 assets: &[AssetNameExchange],
3991) -> Vec<AssetBalance<AssetNameExchange>> {
3992 let now = Utc::now();
3993 if assets.is_empty() {
3995 return user_assets
3996 .into_iter()
3997 .filter_map(|b| convert_margin_balance_entry(b, now))
3998 .collect();
3999 }
4000 if assets.len() <= 16 {
4002 return user_assets
4003 .into_iter()
4004 .filter_map(|b| {
4005 let asset_name_str = b.asset.as_deref()?;
4006 if !assets.iter().any(|a| a.name().as_str() == asset_name_str) {
4007 return None;
4008 }
4009 convert_margin_balance_entry(b, now)
4010 })
4011 .collect();
4012 }
4013 use std::collections::HashSet;
4014 let asset_set: HashSet<&str> = assets.iter().map(|a| a.name().as_str()).collect();
4015 user_assets
4016 .into_iter()
4017 .filter_map(|b| {
4018 let asset_name_str = b.asset.as_deref()?;
4019 if !asset_set.contains(asset_name_str) {
4020 return None;
4021 }
4022 convert_margin_balance_entry(b, now)
4023 })
4024 .collect()
4025}
4026
4027fn active_order_snapshot(
4034 o: Order<ExchangeId, InstrumentNameExchange, Open>,
4035) -> Order<ExchangeId, InstrumentNameExchange, OrderState<AssetNameExchange, InstrumentNameExchange>>
4036{
4037 Order {
4038 key: o.key,
4039 side: o.side,
4040 price: o.price,
4041 quantity: o.quantity,
4042 kind: o.kind,
4043 time_in_force: o.time_in_force,
4044 state: OrderState::active(o.state),
4045 }
4046}
4047
4048fn convert_isolated_asset_balance(
4058 asset: Option<&str>,
4059 free: Option<&str>,
4060 locked: Option<&str>,
4061 borrowed: Option<&str>,
4062 interest: Option<&str>,
4063 now: DateTime<Utc>,
4064) -> Option<AssetBalance<AssetNameExchange>> {
4065 let asset_name = AssetNameExchange::new(asset?);
4066 let free = match free.and_then(|s| Decimal::from_str(s).ok()) {
4067 Some(v) => v,
4068 None => {
4069 warn!(%asset_name, "BinanceMargin isolated balance missing/unparseable 'free' field");
4070 return None;
4071 }
4072 };
4073 let locked = match locked.and_then(|s| Decimal::from_str(s).ok()) {
4074 Some(v) => v,
4075 None => {
4076 warn!(%asset_name, "BinanceMargin isolated balance missing/unparseable 'locked' field");
4077 return None;
4078 }
4079 };
4080 let borrowed = borrowed
4081 .and_then(|s| Decimal::from_str(s).ok())
4082 .unwrap_or(Decimal::ZERO);
4083 let interest = interest
4084 .and_then(|s| Decimal::from_str(s).ok())
4085 .unwrap_or(Decimal::ZERO);
4086 Some(AssetBalance::new(
4087 asset_name,
4088 Balance::new_margin(free + locked, free, borrowed, interest),
4089 now,
4090 ))
4091}
4092
4093fn convert_isolated_margin_assets(
4101 assets: Vec<QueryIsolatedMarginAccountInfoResponseAssetsInner>,
4102) -> HashMap<InstrumentNameExchange, IsolatedInstrumentState<AssetNameExchange>> {
4103 let now = Utc::now();
4106 let mut map = HashMap::with_capacity(assets.len());
4107 for entry in assets {
4108 let Some(symbol) = entry.symbol.as_deref() else {
4109 warn!("BinanceMargin isolated asset entry missing 'symbol' — skipping");
4110 continue;
4111 };
4112 let instrument = InstrumentNameExchange::new(symbol);
4113
4114 let Some(base_raw) = entry.base_asset.as_deref() else {
4115 warn!(%instrument, "BinanceMargin isolated entry missing baseAsset — skipping");
4116 continue;
4117 };
4118 let Some(quote_raw) = entry.quote_asset.as_deref() else {
4119 warn!(%instrument, "BinanceMargin isolated entry missing quoteAsset — skipping");
4120 continue;
4121 };
4122
4123 let Some(base) = convert_isolated_asset_balance(
4124 base_raw.asset.as_deref(),
4125 base_raw.free.as_deref(),
4126 base_raw.locked.as_deref(),
4127 base_raw.borrowed.as_deref(),
4128 base_raw.interest.as_deref(),
4129 now,
4130 ) else {
4131 warn!(%instrument, "BinanceMargin isolated baseAsset missing required field — skipping");
4132 continue;
4133 };
4134 let Some(quote) = convert_isolated_asset_balance(
4135 quote_raw.asset.as_deref(),
4136 quote_raw.free.as_deref(),
4137 quote_raw.locked.as_deref(),
4138 quote_raw.borrowed.as_deref(),
4139 quote_raw.interest.as_deref(),
4140 now,
4141 ) else {
4142 warn!(%instrument, "BinanceMargin isolated quoteAsset missing required field — skipping");
4143 continue;
4144 };
4145
4146 let risk = IsolatedMarginRisk {
4147 margin_level: entry
4148 .margin_level
4149 .as_deref()
4150 .and_then(|s| Decimal::from_str(s).ok()),
4151 margin_ratio: entry
4152 .margin_ratio
4153 .as_deref()
4154 .and_then(|s| Decimal::from_str(s).ok()),
4155 liquidation_price: entry
4156 .liquidate_price
4157 .as_deref()
4158 .and_then(|s| Decimal::from_str(s).ok()),
4159 };
4160
4161 map.insert(instrument, IsolatedInstrumentState { base, quote, risk });
4162 }
4163 map
4164}
4165
4166fn chunk_symbols(symbols: &[InstrumentNameExchange]) -> Vec<String> {
4169 symbols
4170 .chunks(5)
4171 .map(|chunk| {
4172 chunk
4173 .iter()
4174 .map(|s| s.name().as_str())
4175 .collect::<Vec<_>>()
4176 .join(",")
4177 })
4178 .collect()
4179}
4180
4181async fn fetch_isolated_margin_account_info(
4187 rest: Arc<RestApi>,
4188 rate_limiter: Arc<RateLimitTracker>,
4189 symbols: Vec<InstrumentNameExchange>,
4190) -> Result<Vec<QueryIsolatedMarginAccountInfoResponseAssetsInner>, UnindexedClientError> {
4191 use futures::{StreamExt as _, TryStreamExt as _};
4192
4193 let chunks = chunk_symbols(&symbols);
4194
4195 futures::stream::iter(chunks.into_iter().map(|symbols_param| {
4199 let rest = rest.clone();
4200 let rate_limiter = rate_limiter.clone();
4201 async move {
4202 let response = rest_call_with_retry(&rest, &rate_limiter, RequestKind::Query, |rest| {
4203 let symbols_param = symbols_param.clone();
4204 Box::pin(async move {
4205 let params = QueryIsolatedMarginAccountInfoParams::builder()
4206 .symbols(symbols_param)
4207 .build()?;
4208 rest.query_isolated_margin_account_info(params).await
4209 })
4210 })
4211 .await
4212 .map_err(|e| classify_rest_query_error(&e, None))?;
4213
4214 let info = response.data().await.map_err(response_decode_error)?;
4215 Ok::<_, UnindexedClientError>(info.assets.unwrap_or_default())
4216 }
4217 }))
4218 .buffer_unordered(8)
4219 .try_fold(
4220 Vec::with_capacity(symbols.len()),
4221 |mut acc, assets| async move {
4222 acc.extend(assets);
4223 Ok(acc)
4224 },
4225 )
4226 .await
4227}
4228
4229fn convert_margin_trade(
4236 t: &QueryMarginAccountsTradeListResponseInner,
4237 instrument: &InstrumentNameExchange,
4238) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
4239 let trade_id_raw = match t.id {
4240 Some(id) => id,
4241 None => {
4242 warn!(%instrument, "BinanceMargin trade missing id");
4243 return None;
4244 }
4245 };
4246 let trade_id = TradeId(format_smolstr!("{trade_id_raw}"));
4247 let order_id = match t.order_id {
4248 Some(id) => OrderId(format_smolstr!("{id}")),
4249 None => {
4250 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing orderId");
4251 return None;
4252 }
4253 };
4254 let side = match t.is_buyer {
4255 Some(true) => Side::Buy,
4256 Some(false) => Side::Sell,
4257 None => {
4258 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing isBuyer");
4259 return None;
4260 }
4261 };
4262 let price = match t.price.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
4263 Some(v) => v,
4264 None => {
4265 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable price");
4266 return None;
4267 }
4268 };
4269 let quantity = match t.qty.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
4270 Some(v) => v,
4271 None => {
4272 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable qty");
4273 return None;
4274 }
4275 };
4276 let commission = t
4277 .commission
4278 .as_deref()
4279 .and_then(|s| Decimal::from_str(s).ok())
4280 .unwrap_or(Decimal::ZERO);
4281 let time_exchange = match t.time.and_then(|ms| Utc.timestamp_millis_opt(ms).single()) {
4282 Some(ts) => ts,
4283 None => {
4284 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable time, using now");
4285 Utc::now()
4286 }
4287 };
4288
4289 let fee_asset = t
4294 .commission_asset
4295 .as_deref()
4296 .map(AssetNameExchange::from)
4297 .unwrap_or_else(|| AssetNameExchange::from("UNKNOWN"));
4298
4299 Some(Trade::new(
4300 trade_id,
4301 order_id,
4302 instrument.clone(),
4303 StrategyId::unknown(), time_exchange,
4305 side,
4306 price,
4307 quantity,
4308 None,
4310 AssetFees::new(fee_asset, commission, None),
4311 ))
4312}
4313
4314#[derive(Debug)]
4320enum BuildOrderError {
4321 Unsupported,
4323 Build(String),
4325 Trailing(TrailingOrderError),
4327}
4328
4329trait IsolatedFlag {
4340 fn from_flag(is_isolated: bool) -> Self;
4342}
4343
4344macro_rules! impl_isolated_flag {
4349 ($($t:ty),* $(,)?) => {
4350 $(
4351 impl IsolatedFlag for $t {
4352 fn from_flag(is_isolated: bool) -> Self {
4353 if is_isolated { Self::True } else { Self::False }
4354 }
4355 }
4356 )*
4357 };
4358}
4359
4360impl_isolated_flag!(
4361 MarginAccountNewOrderIsIsolatedEnum,
4362 MarginAccountCancelOrderIsIsolatedEnum,
4363 QueryMarginAccountsOpenOrdersIsIsolatedEnum,
4364 QueryMarginAccountsOrderIsIsolatedEnum,
4365 QueryMarginAccountsTradeListIsIsolatedEnum,
4366 GetFutureHourlyInterestRateIsIsolatedEnum,
4367);
4368
4369fn build_cancel_order_params(
4377 symbol: String,
4378 id: Option<&OrderId>,
4379 cid: &ClientOrderId,
4380 is_isolated: bool,
4381) -> Result<MarginAccountCancelOrderParams, String> {
4382 let mut builder = MarginAccountCancelOrderParams::builder(symbol).is_isolated(
4383 MarginAccountCancelOrderIsIsolatedEnum::from_flag(is_isolated),
4384 );
4385
4386 match id {
4387 Some(order_id) => match order_id.0.parse::<i64>() {
4388 Ok(id) => builder = builder.order_id(id),
4389 Err(_) => {
4390 error!(
4392 order_id = %order_id.0,
4393 "BinanceMargin cancel: exchange orderId not parseable as i64, falling back to clientOrderId"
4394 );
4395 builder = builder.orig_client_order_id(cid.0.to_string());
4396 }
4397 },
4398 None => builder = builder.orig_client_order_id(cid.0.to_string()),
4399 }
4400
4401 builder.build().map_err(|e| e.to_string())
4402}
4403
4404#[allow(clippy::too_many_arguments)] fn build_new_order_params(
4413 symbol: String,
4414 side: Side,
4415 price: Option<Decimal>,
4416 quantity: Decimal,
4417 kind: OrderKind,
4418 time_in_force: TimeInForce,
4419 new_client_order_id: String,
4420 side_effect: MarginSideEffect,
4421 is_isolated: bool,
4422) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
4423 let binance_side = match side {
4424 Side::Buy => MarginAccountNewOrderSideEnum::Buy,
4425 Side::Sell => MarginAccountNewOrderSideEnum::Sell,
4426 };
4427
4428 let (binance_type, binance_tif) =
4429 convert_order_kind_tif_margin(kind, time_in_force).ok_or(BuildOrderError::Unsupported)?;
4430
4431 let mut builder = MarginAccountNewOrderParams::builder(symbol, binance_side, binance_type)
4433 .quantity(quantity)
4434 .is_isolated(MarginAccountNewOrderIsIsolatedEnum::from_flag(is_isolated))
4435 .side_effect_type(side_effect.as_sdk_enum())
4436 .new_client_order_id(new_client_order_id)
4437 .new_order_resp_type(MarginAccountNewOrderNewOrderRespTypeEnum::Full);
4438
4439 if side_effect == MarginSideEffect::AutoBorrowRepay {
4442 builder = builder.auto_repay_at_cancel(true);
4443 }
4444
4445 if let Some(tif) = binance_tif {
4446 builder = builder.time_in_force(tif);
4447 }
4448
4449 if let Some(trailing) = trailing_params(&kind, price).map_err(BuildOrderError::Trailing)? {
4452 builder = builder.trailing_delta(i64::from(trailing.trailing_delta));
4453 if let Some(limit) = trailing.price {
4454 builder = builder.price(limit);
4455 }
4456 if let Some(activation_price) = trailing.stop_price {
4457 builder = builder.stop_price(activation_price);
4458 }
4459 }
4460
4461 match kind {
4465 OrderKind::Limit => {
4466 builder = builder.price(price);
4467 }
4468 OrderKind::Stop { trigger_price } | OrderKind::TakeProfit { trigger_price } => {
4469 builder = builder.stop_price(trigger_price);
4470 }
4471 OrderKind::StopLimit { trigger_price } | OrderKind::TakeProfitLimit { trigger_price } => {
4472 builder = builder.price(price).stop_price(trigger_price);
4473 }
4474 OrderKind::TrailingStop { .. }
4477 | OrderKind::TrailingStopFixedLimit { .. }
4478 | OrderKind::Market
4479 | OrderKind::TrailingStopLimit { .. } => {}
4480 }
4481
4482 builder
4483 .build()
4484 .map_err(|e| BuildOrderError::Build(e.to_string()))
4485}
4486
4487fn convert_order_kind_tif_margin(
4498 kind: OrderKind,
4499 tif: TimeInForce,
4500) -> Option<(
4501 MarginAccountNewOrderTypeEnum,
4502 Option<MarginAccountNewOrderTimeInForceEnum>,
4503)> {
4504 let (binance_type, binance_tif) = classify_order_kind_tif(kind, tif)?;
4505 let margin_type = match binance_type {
4506 BinanceOrderType::Market => MarginAccountNewOrderTypeEnum::Market,
4507 BinanceOrderType::Limit => MarginAccountNewOrderTypeEnum::Limit,
4508 BinanceOrderType::LimitMaker => MarginAccountNewOrderTypeEnum::LimitMaker,
4509 BinanceOrderType::StopLoss => MarginAccountNewOrderTypeEnum::StopLoss,
4510 BinanceOrderType::StopLossLimit => MarginAccountNewOrderTypeEnum::StopLossLimit,
4511 BinanceOrderType::TakeProfit => MarginAccountNewOrderTypeEnum::TakeProfit,
4512 BinanceOrderType::TakeProfitLimit => MarginAccountNewOrderTypeEnum::TakeProfitLimit,
4513 };
4514 let margin_tif = binance_tif.map(|t| match t {
4515 BinanceTimeInForce::Gtc => MarginAccountNewOrderTimeInForceEnum::Gtc,
4516 BinanceTimeInForce::Ioc => MarginAccountNewOrderTimeInForceEnum::Ioc,
4517 BinanceTimeInForce::Fok => MarginAccountNewOrderTimeInForceEnum::Fok,
4518 });
4519 Some((margin_type, margin_tif))
4520}
4521
4522#[cfg(test)]
4523#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
4525 use super::*;
4526 use crate::client::binance::shared::{
4527 convert_open_order,
4528 test_ws_api::{self, Lifetime},
4529 unix_ms,
4530 };
4531 use crate::error::StreamTerminationReason;
4532 use crate::notice::{AccountNotice, NoticeKind};
4533 use crate::order::state::ActiveOrderState;
4534 use binance_sdk::margin_trading::rest_api::QueryMarginAccountsOpenOrdersResponseInner;
4535
4536 #[tokio::test]
4539 async fn a_banned_listen_token_subscribe_starts_the_cooldown() {
4540 let until = i64::try_from(unix_ms()).unwrap() + 120_000;
4541 let error = subscribe_error(WebsocketError::ResponseError {
4542 code: -1003,
4543 message: "Way too much request weight used".to_string(),
4544 retry_after: Some(until),
4545 });
4546 assert!(
4547 error
4548 .to_string()
4549 .starts_with("userDataStream.subscribe.listenToken failed: "),
4550 "{error}"
4551 );
4552 let tracker = RateLimitTracker::new(WeightPool::Sapi);
4553
4554 assert!(tracker.on_stream_refused(&error));
4555 assert!(
4556 tracker
4557 .cooldown_remaining()
4558 .is_some_and(|left| left > Duration::from_secs(115)),
4559 "{:?}",
4560 tracker.cooldown_remaining()
4561 );
4562 }
4563
4564 struct ManagerRig {
4568 ws_config: ConfigurationWebsocketApi,
4569 rest_config: Arc<ConfigurationRestApi>,
4570 rest: Arc<RestApi>,
4571 rate_limiter: Arc<RateLimitTracker>,
4572 requests: Arc<parking_lot::Mutex<Vec<tokio::time::Instant>>>,
4574 _rest_server: wiremock::MockServer,
4575 }
4576
4577 impl ManagerRig {
4578 async fn new(
4579 answer: fn(&serde_json::Value) -> serde_json::Value,
4580 lifetime: fn(usize) -> Lifetime,
4581 ) -> Self {
4582 let (ws_url, requests) = test_ws_api::serve(answer, lifetime).await;
4583
4584 let rest_server = wiremock::MockServer::start().await;
4585 wiremock::Mock::given(wiremock::matchers::method("POST"))
4586 .and(wiremock::matchers::path("/sapi/v1/userListenToken"))
4587 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
4588 serde_json::json!({
4589 "token": "token",
4590 "expirationTime": Utc::now().timestamp_millis() + 86_400_000,
4591 }),
4592 ))
4593 .mount(&rest_server)
4594 .await;
4595 let rest_config = Arc::new(
4596 ConfigurationRestApi::builder()
4597 .api_key("key")
4598 .api_secret("secret")
4599 .base_path(rest_server.uri())
4600 .build()
4601 .unwrap(),
4602 );
4603 Self {
4604 ws_config: ConfigurationWebsocketApi::builder()
4605 .api_key("key")
4606 .api_secret("secret")
4607 .ws_url(ws_url)
4608 .build()
4609 .unwrap(),
4610 rest: Arc::new(MarginTradingRestApi::from_config((*rest_config).clone())),
4611 rest_config,
4612 rate_limiter: Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
4613 requests,
4614 _rest_server: rest_server,
4615 }
4616 }
4617
4618 fn token() -> UserListenToken {
4620 UserListenToken {
4621 token: "token".to_string(),
4622 expiration_time_ms: Utc::now().timestamp_millis() + 86_400_000,
4623 }
4624 }
4625
4626 async fn cross(&self) -> mpsc::UnboundedReceiver<UnindexedAccountEvent> {
4628 let (tx, rx) = mpsc::unbounded_channel();
4629 let ws = connect_margin_ws(&self.ws_config).await.unwrap();
4630 tokio::spawn(margin_connection_manager(
4631 tx,
4632 new_dedup_cache(),
4633 self.ws_config.clone(),
4634 Arc::clone(&self.rest_config),
4635 Arc::clone(&self.rest),
4636 Arc::clone(&self.rate_limiter),
4637 KnownLiveOrders::shared(ExchangeId::BinanceMargin),
4638 Vec::new(),
4639 Some((ws, Self::token())),
4640 ));
4641 rx
4642 }
4643
4644 async fn isolated(&self) -> mpsc::UnboundedReceiver<UnindexedAccountEvent> {
4647 let (tx, rx) = mpsc::unbounded_channel();
4648 let (dedup, known) = (
4649 new_dedup_cache(),
4650 KnownLiveOrders::shared(ExchangeId::BinanceMargin),
4651 );
4652 let symbol = InstrumentNameExchange::new("BTCUSDT");
4653 let base_quote = Arc::new(HashMap::from([(
4654 symbol.clone(),
4655 (
4656 AssetNameExchange::new("BTC"),
4657 AssetNameExchange::new("USDT"),
4658 ),
4659 )]));
4660 let tokens = [(symbol.clone(), Self::token())];
4661 let live = isolated_connect_and_subscribe(
4662 &self.ws_config,
4663 &tx,
4664 &dedup,
4665 &known,
4666 &base_quote,
4667 &tokens,
4668 )
4669 .await
4670 .unwrap();
4671 tokio::spawn(isolated_connection_manager(
4672 tx,
4673 dedup,
4674 self.ws_config.clone(),
4675 Arc::clone(&self.rest_config),
4676 Arc::clone(&self.rest),
4677 Arc::clone(&self.rate_limiter),
4678 known,
4679 vec![symbol],
4680 base_quote,
4681 Some(live),
4682 ));
4683 rx
4684 }
4685 }
4686
4687 fn subscribed(request: &serde_json::Value) -> serde_json::Value {
4689 serde_json::json!({
4690 "id": request["id"],
4691 "status": 200,
4692 "result": { "subscriptionId": 0 },
4693 })
4694 }
4695
4696 fn banned_for_two_minutes(request: &serde_json::Value) -> serde_json::Value {
4698 let until = unix_ms() + 120_000;
4699 serde_json::json!({
4700 "id": request["id"],
4701 "status": 418,
4702 "error": {
4703 "code": -1003,
4704 "msg": format!("Way too much request weight used; IP banned until {until}."),
4705 "data": { "serverTime": unix_ms(), "retryAfter": until },
4706 },
4707 })
4708 }
4709
4710 async fn rate_limited_end(rx: &mut mpsc::UnboundedReceiver<UnindexedAccountEvent>) -> Duration {
4713 let event = tokio::time::timeout(Duration::from_secs(5), rx.recv())
4714 .await
4715 .expect("the stream ends rate-limited within the timeout")
4716 .expect("it reports why before it ends");
4717 let AccountEventKind::StreamTerminated(StreamTerminationReason::RateLimited {
4718 retry_after,
4719 }) = event.kind
4720 else {
4721 panic!("expected a rate-limited termination, got {:?}", event.kind);
4722 };
4723 assert!(rx.recv().await.is_none(), "the stream has ended");
4724 retry_after
4725 }
4726
4727 #[tokio::test]
4730 async fn a_banned_subscribe_ends_the_cross_margin_stream_rate_limited() {
4731 let rig = ManagerRig::new(banned_for_two_minutes, |_| Lifetime::Open).await;
4732 let mut events = rig.cross().await;
4733
4734 let wait = rate_limited_end(&mut events).await;
4735
4736 assert!(wait > Duration::from_secs(115), "{wait:?}");
4737 assert_eq!(rig.requests.lock().len(), 1, "no second subscribe");
4738 }
4739
4740 #[tokio::test]
4743 async fn a_banned_isolated_subscribe_is_recognised_as_a_refusal() {
4744 let rig = ManagerRig::new(banned_for_two_minutes, |_| Lifetime::Open).await;
4745 let (tx, _rx) = mpsc::unbounded_channel();
4746 let symbol = InstrumentNameExchange::new("BTCUSDT");
4747
4748 let Err(error) = isolated_connect_and_subscribe(
4749 &rig.ws_config,
4750 &tx,
4751 &new_dedup_cache(),
4752 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
4753 &Arc::new(HashMap::new()),
4754 &[(symbol, ManagerRig::token())],
4755 )
4756 .await
4757 else {
4758 panic!("a banned subscribe fails");
4759 };
4760
4761 assert!(rig.rate_limiter.on_stream_refused(&error), "{error:#}");
4762 assert!(
4763 rig.rate_limiter
4764 .cooldown_remaining()
4765 .is_some_and(|left| left > Duration::from_secs(115))
4766 );
4767 }
4768
4769 #[tokio::test]
4772 async fn a_drop_during_a_long_cooldown_ends_the_isolated_margin_stream_rate_limited() {
4773 let rig = ManagerRig::new(subscribed, |_| Lifetime::Silent(Duration::from_secs(1))).await;
4775 let mut events = rig.isolated().await;
4776 rig.rate_limiter
4777 .on_rate_limited(Some(Duration::from_secs(60)));
4778
4779 let wait = rate_limited_end(&mut events).await;
4780
4781 assert!(wait > Duration::from_secs(55), "{wait:?}");
4782 assert_eq!(rig.requests.lock().len(), 1, "no second subscribe");
4783 }
4784
4785 #[tokio::test]
4788 async fn a_dropped_margin_connection_backs_off_before_reconnecting() {
4789 for isolated in [false, true] {
4790 let rig =
4791 ManagerRig::new(subscribed, |_| Lifetime::Silent(Duration::from_millis(100))).await;
4792 let _events = if isolated {
4793 rig.isolated().await
4794 } else {
4795 rig.cross().await
4796 };
4797
4798 let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
4799 while rig.requests.lock().len() < 2 {
4800 assert!(
4801 tokio::time::Instant::now() < deadline,
4802 "isolated {isolated}: no reconnect"
4803 );
4804 tokio::time::sleep(Duration::from_millis(20)).await;
4805 }
4806
4807 let requests = rig.requests.lock();
4808 let between = requests[1] - requests[0];
4809 assert!(
4811 between >= BackoffLimits::ACCOUNT_STREAM.initial,
4812 "isolated {isolated}: reconnected {between:?} after the first subscribe"
4813 );
4814 }
4815 }
4816
4817 #[tokio::test]
4822 async fn a_margin_reconnect_budget_resets_only_once_a_connection_proves_stable() {
4823 async fn run(isolated: bool, pinging: bool) {
4824 let lifetime = if pinging {
4825 test_ws_api::dropped_then_pinging
4826 } else {
4827 test_ws_api::dropped_then_silent
4828 };
4829 let rig = ManagerRig::new(subscribed, lifetime).await;
4830 let _events = if isolated {
4831 rig.isolated().await
4832 } else {
4833 rig.cross().await
4834 };
4835 let label = format!("isolated {isolated}");
4836 if pinging {
4837 test_ws_api::assert_reset_once_pinged(&rig.requests, &label).await;
4838 } else {
4839 test_ws_api::assert_no_reset_when_silent(&rig.requests, &label).await;
4840 }
4841 }
4842
4843 tokio::join!(
4844 run(false, true),
4845 run(true, true),
4846 run(false, false),
4847 run(true, false)
4848 );
4849 }
4850
4851 #[tokio::test]
4854 async fn a_dropped_margin_connection_is_closed() {
4855 use futures::StreamExt as _;
4856 use tokio_tungstenite::tungstenite::Message;
4857
4858 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4859 let url = format!("ws://{}", listener.local_addr().unwrap());
4860 let (closed_tx, mut closed) = mpsc::unbounded_channel();
4861 tokio::spawn(async move {
4862 let (stream, _) = listener.accept().await.unwrap();
4863 let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
4864 while let Some(Ok(message)) = ws.next().await {
4865 if let Message::Close(_) = message {
4866 let _ = closed_tx.send(());
4867 return;
4868 }
4869 }
4870 });
4871 let config = ConfigurationWebsocketApi::builder()
4872 .api_key("key")
4873 .api_secret("secret")
4874 .ws_url(url)
4875 .build()
4876 .unwrap();
4877
4878 drop(connect_margin_ws(&config).await.unwrap());
4879
4880 tokio::time::timeout(Duration::from_secs(5), closed.recv())
4881 .await
4882 .expect("the dropped connection is closed")
4883 .unwrap();
4884 }
4885
4886 #[test]
4887 fn margin_side_effect_default_is_auto_borrow_repay() {
4888 assert_eq!(
4889 MarginSideEffect::default(),
4890 MarginSideEffect::AutoBorrowRepay
4891 );
4892 }
4893
4894 #[test]
4895 fn margin_side_effect_wire_strings() {
4896 assert_eq!(
4897 MarginSideEffect::AutoBorrowRepay.as_binance_str(),
4898 "AUTO_BORROW_REPAY"
4899 );
4900 assert_eq!(
4901 MarginSideEffect::NoBorrow.as_binance_str(),
4902 "NO_SIDE_EFFECT"
4903 );
4904 }
4905
4906 #[test]
4907 fn margin_side_effect_serde_round_trip() {
4908 assert_eq!(
4910 serde_json::to_string(&MarginSideEffect::AutoBorrowRepay).unwrap(),
4911 r#""auto_borrow_repay""#
4912 );
4913 assert_eq!(
4914 serde_json::to_string(&MarginSideEffect::NoBorrow).unwrap(),
4915 r#""no_borrow""#
4916 );
4917 assert_eq!(
4918 serde_json::from_str::<MarginSideEffect>(r#""auto_borrow_repay""#).unwrap(),
4919 MarginSideEffect::AutoBorrowRepay
4920 );
4921 assert_eq!(
4922 serde_json::from_str::<MarginSideEffect>(r#""no_borrow""#).unwrap(),
4923 MarginSideEffect::NoBorrow
4924 );
4925 }
4926
4927 #[test]
4928 fn config_debug_redacts_secrets() {
4929 let config = BinanceMarginConfig::new(
4930 "my_api_key".to_string(),
4931 "my_secret_key".to_string(),
4932 false,
4933 false,
4934 Vec::new(),
4935 MarginSideEffect::default(),
4936 );
4937 let debug = format!("{config:?}");
4938 assert!(!debug.contains("my_api_key"));
4939 assert!(!debug.contains("my_secret_key"));
4940 assert!(debug.contains("***"));
4941 }
4942
4943 #[test]
4944 fn cross_margin_uses_common_case_defaults() {
4945 let config = BinanceMarginConfig::cross_margin("k".to_string(), "s".to_string());
4946 assert!(!config.testnet);
4947 assert!(!config.is_isolated);
4948 assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
4949 assert_eq!(config.api_key(), "k");
4950 }
4951
4952 #[test]
4953 fn config_deserializes_with_defaults() {
4954 let config: BinanceMarginConfig =
4956 serde_json::from_str(r#"{"api_key":"k","secret_key":"s","testnet":false}"#)
4957 .expect("deserialize");
4958 assert!(!config.is_isolated);
4959 assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
4960 }
4961
4962 #[test]
4963 fn config_deserializes_explicit_side_effect() {
4964 let config: BinanceMarginConfig = serde_json::from_str(
4965 r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true,"side_effect":"no_borrow"}"#,
4966 )
4967 .expect("deserialize");
4968 assert!(config.is_isolated);
4969 assert_eq!(config.side_effect, MarginSideEffect::NoBorrow);
4970 }
4971
4972 #[test]
4973 fn isolated_ctor_sets_symbols_and_flag() {
4974 let symbols = vec![
4975 InstrumentNameExchange::new("BTCUSDT"),
4976 InstrumentNameExchange::new("ETHUSDT"),
4977 ];
4978 let config =
4979 BinanceMarginConfig::isolated("k".to_string(), "s".to_string(), symbols.clone());
4980 assert!(config.is_isolated);
4981 assert_eq!(config.isolated_symbols, symbols);
4982 assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
4983 }
4984
4985 #[test]
4986 fn isolated_config_with_symbols_constructs() {
4987 let config = BinanceMarginConfig::isolated(
4989 "k".to_string(),
4990 "s".to_string(),
4991 vec![InstrumentNameExchange::new("BTCUSDT")],
4992 );
4993 let _client = BinanceMargin::new(config); }
4995
4996 #[test]
4997 #[should_panic(expected = "non-empty isolated_symbols")]
4998 fn isolated_empty_symbols_panics_at_new() {
4999 let config = BinanceMarginConfig::new(
5001 "k".to_string(),
5002 "s".to_string(),
5003 false,
5004 true,
5005 Vec::new(),
5006 MarginSideEffect::default(),
5007 );
5008 let _ = BinanceMargin::new(config);
5009 }
5010
5011 #[test]
5012 #[should_panic(expected = "non-empty isolated_symbols")]
5013 fn isolated_empty_symbols_panics_via_deserialize_path() {
5014 let config: BinanceMarginConfig = serde_json::from_str(
5017 r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true}"#,
5018 )
5019 .expect("deserialize");
5020 assert!(config.isolated_symbols.is_empty());
5021 let _ = BinanceMargin::new(config);
5022 }
5023
5024 #[test]
5025 fn isolated_flag_maps_mode() {
5026 assert_eq!(
5029 MarginAccountNewOrderIsIsolatedEnum::from_flag(true).as_str(),
5030 "TRUE"
5031 );
5032 assert_eq!(
5033 MarginAccountNewOrderIsIsolatedEnum::from_flag(false).as_str(),
5034 "FALSE"
5035 );
5036 assert_eq!(
5037 MarginAccountCancelOrderIsIsolatedEnum::from_flag(true).as_str(),
5038 "TRUE"
5039 );
5040 assert_eq!(
5041 MarginAccountCancelOrderIsIsolatedEnum::from_flag(false).as_str(),
5042 "FALSE"
5043 );
5044 assert_eq!(
5045 QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(true).as_str(),
5046 "TRUE"
5047 );
5048 assert_eq!(
5049 QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(false).as_str(),
5050 "FALSE"
5051 );
5052 assert_eq!(
5053 QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(true).as_str(),
5054 "TRUE"
5055 );
5056 assert_eq!(
5057 QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(false).as_str(),
5058 "FALSE"
5059 );
5060 }
5061
5062 #[test]
5067 fn cancel_params_cross_with_parseable_order_id() {
5068 let p = build_cancel_order_params(
5069 "BTCUSDT".to_string(),
5070 Some(&OrderId::new("12345")),
5071 &ClientOrderId::new("cid-1"),
5072 false,
5073 )
5074 .expect("build");
5075 assert_eq!(p.symbol, "BTCUSDT");
5076 assert_eq!(p.is_isolated.as_ref().map(|i| i.as_str()), Some("FALSE"));
5077 assert_eq!(p.order_id, Some(12345));
5078 assert_eq!(p.orig_client_order_id, None);
5079 }
5080
5081 #[test]
5082 fn cancel_params_isolated_is_config_driven() {
5083 let p = build_cancel_order_params(
5084 "BTCUSDT".to_string(),
5085 Some(&OrderId::new("12345")),
5086 &ClientOrderId::new("cid-1"),
5087 true,
5088 )
5089 .expect("build");
5090 assert_eq!(p.is_isolated.as_ref().map(|i| i.as_str()), Some("TRUE"));
5091 assert_eq!(p.order_id, Some(12345));
5092 }
5093
5094 #[test]
5095 fn cancel_params_non_i64_order_id_falls_back_to_cid() {
5096 let p = build_cancel_order_params(
5097 "BTCUSDT".to_string(),
5098 Some(&OrderId::new("not-an-i64")),
5099 &ClientOrderId::new("cid-1"),
5100 false,
5101 )
5102 .expect("build");
5103 assert_eq!(p.order_id, None);
5104 assert_eq!(p.orig_client_order_id.as_deref(), Some("cid-1"));
5105 }
5106
5107 #[test]
5108 fn cancel_params_absent_order_id_uses_cid() {
5109 let p = build_cancel_order_params(
5110 "BTCUSDT".to_string(),
5111 None,
5112 &ClientOrderId::new("cid-1"),
5113 false,
5114 )
5115 .expect("build");
5116 assert_eq!(p.order_id, None);
5117 assert_eq!(p.orig_client_order_id.as_deref(), Some("cid-1"));
5118 }
5119
5120 use crate::order::TrailingOffsetType;
5125
5126 fn params(
5128 side: Side,
5129 price: Option<Decimal>,
5130 kind: OrderKind,
5131 tif: TimeInForce,
5132 side_effect: MarginSideEffect,
5133 ) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
5134 build_new_order_params(
5135 "BTCUSDT".to_string(),
5136 side,
5137 price,
5138 Decimal::from(2),
5139 kind,
5140 tif,
5141 "cid-1".to_string(),
5142 side_effect,
5143 false, )
5145 }
5146
5147 fn gtc() -> TimeInForce {
5148 TimeInForce::GoodUntilCancelled { post_only: false }
5149 }
5150
5151 #[test]
5152 fn new_order_limit_maps_core_fields() {
5153 let p = params(
5154 Side::Buy,
5155 Some(Decimal::from(50_000)),
5156 OrderKind::Limit,
5157 gtc(),
5158 MarginSideEffect::AutoBorrowRepay,
5159 )
5160 .expect("build");
5161
5162 assert_eq!(p.symbol, "BTCUSDT");
5163 assert_eq!(p.side.as_str(), "BUY");
5164 assert_eq!(p.r#type.as_str(), "LIMIT");
5165 assert_eq!(p.quantity, Some(Decimal::from(2)));
5166 assert_eq!(p.price, Some(Decimal::from(50_000)));
5167 assert_eq!(p.stop_price, None);
5168 assert_eq!(p.time_in_force.as_ref().map(|t| t.as_str()), Some("GTC"));
5169 assert_eq!(p.new_client_order_id.as_deref(), Some("cid-1"));
5170 assert_eq!(
5172 p.new_order_resp_type.as_ref().map(|r| r.as_str()),
5173 Some("FULL")
5174 );
5175 }
5176
5177 #[test]
5178 fn new_order_is_isolated_is_config_driven() {
5179 let cross = build_new_order_params(
5181 "BTCUSDT".to_string(),
5182 Side::Sell,
5183 Some(Decimal::from(10)),
5184 Decimal::from(2),
5185 OrderKind::Limit,
5186 gtc(),
5187 "cid-1".to_string(),
5188 MarginSideEffect::AutoBorrowRepay,
5189 false,
5190 )
5191 .expect("build");
5192 assert_eq!(
5193 cross.is_isolated.as_ref().map(|i| i.as_str()),
5194 Some("FALSE")
5195 );
5196 assert_eq!(cross.side.as_str(), "SELL");
5197
5198 let isolated = build_new_order_params(
5200 "BTCUSDT".to_string(),
5201 Side::Sell,
5202 Some(Decimal::from(10)),
5203 Decimal::from(2),
5204 OrderKind::Limit,
5205 gtc(),
5206 "cid-1".to_string(),
5207 MarginSideEffect::AutoBorrowRepay,
5208 true,
5209 )
5210 .expect("build");
5211 assert_eq!(
5212 isolated.is_isolated.as_ref().map(|i| i.as_str()),
5213 Some("TRUE")
5214 );
5215 }
5216
5217 #[test]
5218 fn side_effect_and_auto_repay_gating() {
5219 let auto = params(
5221 Side::Buy,
5222 Some(Decimal::from(1)),
5223 OrderKind::Limit,
5224 gtc(),
5225 MarginSideEffect::AutoBorrowRepay,
5226 )
5227 .expect("build");
5228 assert_eq!(
5229 auto.side_effect_type.as_ref().map(|s| s.as_str()),
5230 Some("AUTO_BORROW_REPAY")
5231 );
5232 assert_eq!(auto.auto_repay_at_cancel, Some(true));
5233
5234 let no_borrow = params(
5236 Side::Buy,
5237 Some(Decimal::from(1)),
5238 OrderKind::Limit,
5239 gtc(),
5240 MarginSideEffect::NoBorrow,
5241 )
5242 .expect("build");
5243 assert_eq!(
5244 no_borrow.side_effect_type.as_ref().map(|s| s.as_str()),
5245 Some("NO_SIDE_EFFECT")
5246 );
5247 assert_eq!(no_borrow.auto_repay_at_cancel, None);
5248 }
5249
5250 #[test]
5251 fn new_order_market_has_no_price_or_tif() {
5252 let p = params(
5253 Side::Buy,
5254 None,
5255 OrderKind::Market,
5256 gtc(),
5257 MarginSideEffect::AutoBorrowRepay,
5258 )
5259 .expect("build");
5260 assert_eq!(p.r#type.as_str(), "MARKET");
5261 assert_eq!(p.price, None);
5262 assert_eq!(p.stop_price, None);
5263 assert!(p.time_in_force.is_none());
5264 }
5265
5266 #[test]
5267 fn post_only_limit_maps_to_limit_maker() {
5268 let p = params(
5269 Side::Buy,
5270 Some(Decimal::from(10)),
5271 OrderKind::Limit,
5272 TimeInForce::GoodUntilCancelled { post_only: true },
5273 MarginSideEffect::AutoBorrowRepay,
5274 )
5275 .expect("build");
5276 assert_eq!(p.r#type.as_str(), "LIMIT_MAKER");
5277 assert!(p.time_in_force.is_none());
5279 }
5280
5281 #[test]
5282 fn conditional_kinds_set_stop_price() {
5283 let trigger = Decimal::from(48_000);
5284
5285 let stop = params(
5286 Side::Sell,
5287 None,
5288 OrderKind::Stop {
5289 trigger_price: trigger,
5290 },
5291 gtc(),
5292 MarginSideEffect::AutoBorrowRepay,
5293 )
5294 .expect("build");
5295 assert_eq!(stop.r#type.as_str(), "STOP_LOSS");
5296 assert_eq!(stop.stop_price, Some(trigger));
5297 assert_eq!(stop.price, None);
5298
5299 let stop_limit = params(
5300 Side::Sell,
5301 Some(Decimal::from(47_900)),
5302 OrderKind::StopLimit {
5303 trigger_price: trigger,
5304 },
5305 gtc(),
5306 MarginSideEffect::AutoBorrowRepay,
5307 )
5308 .expect("build");
5309 assert_eq!(stop_limit.r#type.as_str(), "STOP_LOSS_LIMIT");
5310 assert_eq!(stop_limit.stop_price, Some(trigger));
5311 assert_eq!(stop_limit.price, Some(Decimal::from(47_900)));
5312 assert_eq!(
5313 stop_limit.time_in_force.as_ref().map(|t| t.as_str()),
5314 Some("GTC")
5315 );
5316
5317 let take_profit = params(
5318 Side::Sell,
5319 None,
5320 OrderKind::TakeProfit {
5321 trigger_price: trigger,
5322 },
5323 gtc(),
5324 MarginSideEffect::AutoBorrowRepay,
5325 )
5326 .expect("build");
5327 assert_eq!(take_profit.r#type.as_str(), "TAKE_PROFIT");
5328 assert_eq!(take_profit.stop_price, Some(trigger));
5329 }
5330
5331 #[test]
5334 fn a_trailing_stop_maps_to_stop_loss_with_trailing_delta() {
5335 let cases = [
5336 (Decimal::from(25), TrailingOffsetType::BasisPoints, 25),
5337 (Decimal::new(15, 1), TrailingOffsetType::Percentage, 150),
5338 ];
5339 for (offset, offset_type, trailing_delta) in cases {
5340 let p = params(
5341 Side::Sell,
5342 None,
5343 OrderKind::TrailingStop {
5344 offset,
5345 offset_type,
5346 activation_price: None,
5347 },
5348 gtc(),
5349 MarginSideEffect::NoBorrow,
5350 )
5351 .expect("build");
5352 assert_eq!(p.r#type.as_str(), "STOP_LOSS");
5353 assert_eq!(p.trailing_delta, Some(trailing_delta), "{offset_type:?}");
5354 assert_eq!(p.stop_price, None);
5355 assert_eq!(p.price, None);
5356 assert!(p.time_in_force.is_none());
5357 }
5358 }
5359
5360 #[test]
5364 fn trailing_orders_carry_their_activation_price_and_fixed_limit() {
5365 let activation = Some(Decimal::from(50_000));
5366 let trailing_stop = params(
5367 Side::Sell,
5368 None,
5369 OrderKind::TrailingStop {
5370 offset: Decimal::from(25),
5371 offset_type: TrailingOffsetType::BasisPoints,
5372 activation_price: activation,
5373 },
5374 gtc(),
5375 MarginSideEffect::NoBorrow,
5376 )
5377 .expect("build");
5378 assert_eq!(trailing_stop.r#type.as_str(), "STOP_LOSS");
5379 assert_eq!(trailing_stop.trailing_delta, Some(25));
5380 assert_eq!(trailing_stop.stop_price, activation);
5381 assert_eq!(trailing_stop.price, None);
5382
5383 for activation_price in [None, activation] {
5384 let fixed_limit = params(
5385 Side::Sell,
5386 Some(Decimal::from(49_000)),
5387 OrderKind::TrailingStopFixedLimit {
5388 offset: Decimal::new(15, 1),
5389 offset_type: TrailingOffsetType::Percentage,
5390 activation_price,
5391 },
5392 TimeInForce::ImmediateOrCancel,
5393 MarginSideEffect::NoBorrow,
5394 )
5395 .expect("build");
5396 assert_eq!(fixed_limit.r#type.as_str(), "STOP_LOSS_LIMIT");
5397 assert_eq!(fixed_limit.trailing_delta, Some(150));
5398 assert_eq!(fixed_limit.price, Some(Decimal::from(49_000)));
5399 assert_eq!(fixed_limit.stop_price, activation_price);
5400 assert_eq!(
5401 fixed_limit.time_in_force.as_ref().map(|t| t.as_str()),
5402 Some("IOC")
5403 );
5404 }
5405
5406 let absolute = params(
5407 Side::Sell,
5408 Some(Decimal::from(49_000)),
5409 OrderKind::TrailingStopFixedLimit {
5410 offset: Decimal::from(100),
5411 offset_type: TrailingOffsetType::Absolute,
5412 activation_price: None,
5413 },
5414 gtc(),
5415 MarginSideEffect::NoBorrow,
5416 );
5417 assert!(matches!(absolute, Err(BuildOrderError::Unsupported)));
5418
5419 let no_limit = params(
5421 Side::Sell,
5422 None,
5423 OrderKind::TrailingStopFixedLimit {
5424 offset: Decimal::from(25),
5425 offset_type: TrailingOffsetType::BasisPoints,
5426 activation_price: None,
5427 },
5428 gtc(),
5429 MarginSideEffect::NoBorrow,
5430 );
5431 assert!(matches!(
5432 no_limit,
5433 Err(BuildOrderError::Trailing(TrailingOrderError::LimitPrice(
5434 None
5435 )))
5436 ));
5437 let zero_activation = params(
5438 Side::Sell,
5439 None,
5440 OrderKind::TrailingStop {
5441 offset: Decimal::from(25),
5442 offset_type: TrailingOffsetType::BasisPoints,
5443 activation_price: Some(Decimal::ZERO),
5444 },
5445 gtc(),
5446 MarginSideEffect::NoBorrow,
5447 );
5448 assert!(matches!(
5449 zero_activation,
5450 Err(BuildOrderError::Trailing(
5451 TrailingOrderError::NonPositiveActivation(_)
5452 ))
5453 ));
5454 }
5455
5456 #[test]
5457 fn absolute_trailing_offsets_and_trailing_stop_limits_are_unsupported() {
5458 let absolute = params(
5459 Side::Sell,
5460 None,
5461 OrderKind::TrailingStop {
5462 offset: Decimal::from(100),
5463 offset_type: TrailingOffsetType::Absolute,
5464 activation_price: None,
5465 },
5466 gtc(),
5467 MarginSideEffect::AutoBorrowRepay,
5468 );
5469 assert!(matches!(absolute, Err(BuildOrderError::Unsupported)));
5470
5471 let trailing_stop_limit = params(
5472 Side::Sell,
5473 Some(Decimal::from(100)),
5474 OrderKind::TrailingStopLimit {
5475 offset: Decimal::from(100),
5476 offset_type: TrailingOffsetType::BasisPoints,
5477 stop_price: Decimal::from(95),
5478 limit_offset: Decimal::from(10),
5479 },
5480 gtc(),
5481 MarginSideEffect::AutoBorrowRepay,
5482 );
5483 assert!(matches!(
5484 trailing_stop_limit,
5485 Err(BuildOrderError::Unsupported)
5486 ));
5487 }
5488
5489 #[test]
5491 fn a_fractional_trailing_offset_is_refused() {
5492 let fractional = params(
5493 Side::Sell,
5494 None,
5495 OrderKind::TrailingStop {
5496 offset: Decimal::new(155, 3),
5497 offset_type: TrailingOffsetType::Percentage,
5498 activation_price: None,
5499 },
5500 gtc(),
5501 MarginSideEffect::AutoBorrowRepay,
5502 );
5503 assert!(matches!(
5504 fractional,
5505 Err(BuildOrderError::Trailing(TrailingOrderError::Delta(
5506 crate::client::binance::shared::TrailingDeltaError::Invalid(_)
5507 )))
5508 ));
5509 }
5510
5511 #[test]
5512 fn tif_variants_map_to_margin_enum() {
5513 for (tif, expected) in [
5514 (TimeInForce::ImmediateOrCancel, "IOC"),
5515 (TimeInForce::FillOrKill, "FOK"),
5516 (gtc(), "GTC"),
5517 ] {
5518 let p = params(
5519 Side::Buy,
5520 Some(Decimal::from(1)),
5521 OrderKind::Limit,
5522 tif,
5523 MarginSideEffect::AutoBorrowRepay,
5524 )
5525 .expect("build");
5526 assert_eq!(p.time_in_force.as_ref().map(|t| t.as_str()), Some(expected));
5527 }
5528 }
5529
5530 fn user_asset(
5535 asset: &str,
5536 free: &str,
5537 locked: &str,
5538 borrowed: &str,
5539 interest: &str,
5540 ) -> QueryCrossMarginAccountDetailsResponseUserAssetsInner {
5541 let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
5542 a.asset = Some(asset.to_string());
5543 a.free = Some(free.to_string());
5544 a.locked = Some(locked.to_string());
5545 a.borrowed = Some(borrowed.to_string());
5546 a.interest = Some(interest.to_string());
5547 a
5548 }
5549
5550 #[test]
5551 fn margin_balance_entry_maps_debt() {
5552 let ab =
5554 convert_margin_balance_entry(user_asset("USDT", "10", "2", "3", "0.5"), Utc::now())
5555 .expect("convert");
5556 assert_eq!(ab.asset.name().as_str(), "USDT");
5557 assert_eq!(ab.balance.total, Decimal::from(12));
5558 assert_eq!(ab.balance.free, Decimal::from(10));
5559 assert!(
5560 ab.balance.margin.is_some(),
5561 "REST snapshot must populate margin"
5562 );
5563 assert_eq!(ab.balance.net_asset(), Decimal::from(9));
5565 }
5566
5567 #[test]
5568 fn margin_balance_short_is_negative_net() {
5569 let ab = convert_margin_balance_entry(user_asset("BTC", "1", "0", "5", "0"), Utc::now())
5571 .expect("convert");
5572 assert_eq!(ab.balance.net_asset(), Decimal::from(-4));
5573 }
5574
5575 #[test]
5576 fn margin_balance_missing_debt_defaults_to_zero() {
5577 let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
5580 a.asset = Some("ETH".to_string());
5581 a.free = Some("4".to_string());
5582 a.locked = Some("0".to_string());
5583 let ab = convert_margin_balance_entry(a, Utc::now()).expect("convert");
5584 assert!(ab.balance.margin.is_some());
5585 assert_eq!(ab.balance.net_asset(), Decimal::from(4));
5586 }
5587
5588 #[test]
5589 fn margin_balance_missing_free_is_dropped() {
5590 let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
5592 a.asset = Some("USDT".to_string());
5593 a.locked = Some("0".to_string());
5594 assert!(convert_margin_balance_entry(a, Utc::now()).is_none());
5595 }
5596
5597 #[test]
5598 fn margin_balance_filtering() {
5599 let assets = vec![
5600 user_asset("USDT", "10", "0", "0", "0"),
5601 user_asset("BTC", "1", "0", "0", "0"),
5602 user_asset("ETH", "5", "0", "0", "0"),
5603 ];
5604 assert_eq!(
5606 filter_and_convert_margin_balances(assets.clone(), &[]).len(),
5607 3
5608 );
5609 let filtered = filter_and_convert_margin_balances(
5611 assets,
5612 &[AssetNameExchange::new("BTC"), AssetNameExchange::new("ETH")],
5613 );
5614 assert_eq!(filtered.len(), 2);
5615 assert!(filtered.iter().all(|b| b.asset.name().as_str() != "USDT"));
5616 }
5617
5618 #[test]
5619 fn margin_balance_filtering_large_slice_uses_hashset_branch() {
5620 let assets = vec![
5622 user_asset("BTC", "1", "0", "0", "0"),
5623 user_asset("ETH", "5", "0", "0", "0"),
5624 user_asset("DOGE", "9", "0", "0", "0"), ];
5626 let mut requested: Vec<AssetNameExchange> = (0..17)
5628 .map(|i| AssetNameExchange::new(format!("A{i:02}")))
5629 .collect();
5630 requested.push(AssetNameExchange::new("BTC"));
5631 requested.push(AssetNameExchange::new("ETH"));
5632 assert!(
5633 requested.len() > 16,
5634 "must exceed the linear-scan threshold"
5635 );
5636
5637 let filtered = filter_and_convert_margin_balances(assets, &requested);
5638 assert_eq!(filtered.len(), 2);
5639 assert!(filtered.iter().all(|b| b.asset.name().as_str() != "DOGE"));
5640 }
5641
5642 use binance_sdk::margin_trading::rest_api::{
5647 QueryIsolatedMarginAccountInfoResponseAssetsInnerBaseAsset as IsoBase,
5648 QueryIsolatedMarginAccountInfoResponseAssetsInnerQuoteAsset as IsoQuote,
5649 };
5650
5651 fn d(s: &str) -> Decimal {
5652 Decimal::from_str(s).unwrap()
5653 }
5654
5655 fn iso_base(asset: &str, free: &str, locked: &str, borrowed: &str, interest: &str) -> IsoBase {
5656 let mut b = IsoBase::new();
5657 b.asset = Some(asset.to_string());
5658 b.free = Some(free.to_string());
5659 b.locked = Some(locked.to_string());
5660 b.borrowed = Some(borrowed.to_string());
5661 b.interest = Some(interest.to_string());
5662 b
5663 }
5664
5665 fn iso_quote(
5666 asset: &str,
5667 free: &str,
5668 locked: &str,
5669 borrowed: &str,
5670 interest: &str,
5671 ) -> IsoQuote {
5672 let mut q = IsoQuote::new();
5673 q.asset = Some(asset.to_string());
5674 q.free = Some(free.to_string());
5675 q.locked = Some(locked.to_string());
5676 q.borrowed = Some(borrowed.to_string());
5677 q.interest = Some(interest.to_string());
5678 q
5679 }
5680
5681 fn iso_entry(
5682 symbol: &str,
5683 base: IsoBase,
5684 quote: IsoQuote,
5685 margin_level: Option<&str>,
5686 margin_ratio: Option<&str>,
5687 liquidate_price: Option<&str>,
5688 ) -> QueryIsolatedMarginAccountInfoResponseAssetsInner {
5689 let mut e = QueryIsolatedMarginAccountInfoResponseAssetsInner::new();
5690 e.symbol = Some(symbol.to_string());
5691 e.base_asset = Some(Box::new(base));
5692 e.quote_asset = Some(Box::new(quote));
5693 e.margin_level = margin_level.map(str::to_string);
5694 e.margin_ratio = margin_ratio.map(str::to_string);
5695 e.liquidate_price = liquidate_price.map(str::to_string);
5696 e
5697 }
5698
5699 fn isolated_client(symbols: &[&str]) -> BinanceMargin {
5700 let config = BinanceMarginConfig::isolated(
5701 "k".to_string(),
5702 "s".to_string(),
5703 symbols
5704 .iter()
5705 .map(|s| InstrumentNameExchange::new(*s))
5706 .collect(),
5707 );
5708 BinanceMargin::new(config)
5709 }
5710
5711 #[test]
5712 fn isolated_assets_map_base_quote_and_risk() {
5713 let entry = iso_entry(
5716 "BTCUSDT",
5717 iso_base("BTC", "0.5", "0.1", "0.2", "0.001"),
5718 iso_quote("USDT", "1000", "50", "300", "1.5"),
5719 Some("3.5"),
5720 Some("0.12"),
5721 Some("48000"),
5722 );
5723 let map = convert_isolated_margin_assets(vec![entry]);
5724 let state = map
5725 .get(&InstrumentNameExchange::new("BTCUSDT"))
5726 .expect("entry present");
5727
5728 assert_eq!(state.base.asset.name().as_str(), "BTC");
5729 assert_eq!(state.base.balance.total, d("0.6"));
5730 assert_eq!(state.base.balance.free, d("0.5"));
5731 assert_eq!(state.base.balance.net_asset(), d("0.4"));
5732
5733 assert_eq!(state.quote.asset.name().as_str(), "USDT");
5734 assert_eq!(state.quote.balance.total, d("1050"));
5735 assert_eq!(state.quote.balance.net_asset(), d("750"));
5736
5737 assert_eq!(state.risk.margin_level, Some(d("3.5")));
5738 assert_eq!(state.risk.margin_ratio, Some(d("0.12")));
5739 assert_eq!(state.risk.liquidation_price, Some(d("48000")));
5740 }
5741
5742 #[test]
5743 fn isolated_base_short_is_negative_net() {
5744 let entry = iso_entry(
5746 "BTCUSDT",
5747 iso_base("BTC", "0", "0", "1.5", "0.001"),
5748 iso_quote("USDT", "100", "0", "0", "0"),
5749 None,
5750 None,
5751 None,
5752 );
5753 let map = convert_isolated_margin_assets(vec![entry]);
5754 let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
5755 assert_eq!(state.base.balance.net_asset(), d("-1.5"));
5756 }
5757
5758 #[test]
5759 fn isolated_missing_debt_defaults_zero_and_risk_optional() {
5760 let mut base = IsoBase::new();
5763 base.asset = Some("BTC".to_string());
5764 base.free = Some("0.5".to_string());
5765 base.locked = Some("0".to_string());
5766 let mut quote = IsoQuote::new();
5767 quote.asset = Some("USDT".to_string());
5768 quote.free = Some("100".to_string());
5769 quote.locked = Some("0".to_string());
5770
5771 let map = convert_isolated_margin_assets(vec![iso_entry(
5772 "BTCUSDT", base, quote, None, None, None,
5773 )]);
5774 let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
5775 assert!(state.base.balance.margin.is_some());
5776 assert_eq!(state.base.balance.net_asset(), d("0.5"));
5777 assert_eq!(state.risk.margin_level, None);
5778 assert_eq!(state.risk.liquidation_price, None);
5779 }
5780
5781 #[test]
5782 fn isolated_unparseable_risk_is_none_but_entry_kept() {
5783 let entry = iso_entry(
5784 "BTCUSDT",
5785 iso_base("BTC", "1", "0", "0", "0"),
5786 iso_quote("USDT", "1", "0", "0", "0"),
5787 Some("not_a_number"),
5788 None,
5789 None,
5790 );
5791 let map = convert_isolated_margin_assets(vec![entry]);
5792 let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
5793 assert_eq!(state.risk.margin_level, None);
5794 }
5795
5796 #[test]
5797 fn isolated_missing_base_free_drops_entry() {
5798 let mut base = IsoBase::new();
5800 base.asset = Some("BTC".to_string());
5801 base.locked = Some("0".to_string()); let entry = iso_entry(
5803 "BTCUSDT",
5804 base,
5805 iso_quote("USDT", "100", "0", "0", "0"),
5806 None,
5807 None,
5808 None,
5809 );
5810 assert!(convert_isolated_margin_assets(vec![entry]).is_empty());
5811 }
5812
5813 #[test]
5814 fn isolated_missing_symbol_drops_entry() {
5815 let mut e = QueryIsolatedMarginAccountInfoResponseAssetsInner::new();
5816 e.base_asset = Some(Box::new(iso_base("BTC", "1", "0", "0", "0")));
5817 e.quote_asset = Some(Box::new(iso_quote("USDT", "1", "0", "0", "0")));
5818 assert!(convert_isolated_margin_assets(vec![e]).is_empty());
5820 }
5821
5822 #[test]
5823 fn chunk_symbols_batches_by_five() {
5824 let syms: Vec<InstrumentNameExchange> = (0..12)
5825 .map(|i| InstrumentNameExchange::new(format!("S{i:02}")))
5826 .collect();
5827 let chunks = chunk_symbols(&syms);
5828 assert_eq!(chunks.len(), 3, "12 symbols → 5 + 5 + 2");
5829 assert_eq!(chunks[0].split(',').count(), 5);
5830 assert_eq!(chunks[1].split(',').count(), 5);
5831 assert_eq!(chunks[2].split(',').count(), 2);
5832
5833 let three: Vec<_> = (0..3)
5835 .map(|i| InstrumentNameExchange::new(format!("S{i}")))
5836 .collect();
5837 assert_eq!(chunk_symbols(&three).len(), 1);
5838
5839 let five: Vec<_> = (0..5)
5841 .map(|i| InstrumentNameExchange::new(format!("S{i}")))
5842 .collect();
5843 let c = chunk_symbols(&five);
5844 assert_eq!(c.len(), 1);
5845 assert_eq!(c[0].split(',').count(), 5);
5846
5847 assert!(chunk_symbols(&[]).is_empty());
5849 }
5850
5851 #[test]
5852 fn effective_isolated_set_empty_returns_all_configured() {
5853 let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
5854 assert_eq!(
5855 client.effective_isolated_set(&[]),
5856 vec![
5857 InstrumentNameExchange::new("BTCUSDT"),
5858 InstrumentNameExchange::new("ETHUSDT"),
5859 ]
5860 );
5861 }
5862
5863 #[test]
5864 fn effective_isolated_set_intersects_requested() {
5865 let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
5866 assert_eq!(
5867 client.effective_isolated_set(&[InstrumentNameExchange::new("ETHUSDT")]),
5868 vec![InstrumentNameExchange::new("ETHUSDT")]
5869 );
5870 }
5871
5872 #[test]
5873 fn effective_isolated_set_skips_out_of_set() {
5874 let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
5875 assert_eq!(
5877 client.effective_isolated_set(&[
5878 InstrumentNameExchange::new("BTCUSDT"),
5879 InstrumentNameExchange::new("DOGEUSDT"),
5880 ]),
5881 vec![InstrumentNameExchange::new("BTCUSDT")]
5882 );
5883 }
5884
5885 #[tokio::test]
5886 async fn fetch_balances_isolated_returns_empty() {
5887 let client = isolated_client(&["BTCUSDT"]);
5890 assert!(client.fetch_balances(&[]).await.expect("ok").is_empty());
5891 }
5892
5893 fn open_order(order_id: Option<i64>, side: &str) -> QueryMarginAccountsOpenOrdersResponseInner {
5894 let mut o = QueryMarginAccountsOpenOrdersResponseInner::new();
5895 o.order_id = order_id;
5896 o.client_order_id = Some("cid-9".to_string());
5897 o.side = Some(side.to_string());
5898 o.price = Some("50000".to_string());
5899 o.orig_qty = Some("2".to_string());
5900 o.executed_qty = Some("0.5".to_string());
5901 o.r#type = Some("LIMIT".to_string());
5902 o.time_in_force = Some("GTC".to_string());
5903 o.time = Some(1_700_000_000_000);
5904 o.status = Some("PARTIALLY_FILLED".to_string());
5905 o
5906 }
5907
5908 #[test]
5909 fn margin_open_order_converts() {
5910 let inst = InstrumentNameExchange::new("BTCUSDT");
5911 let order = convert_open_order(
5912 &open_order(Some(42), "BUY"),
5913 ExchangeId::BinanceMargin,
5914 &inst,
5915 )
5916 .expect("convert");
5917 assert_eq!(order.key.exchange, ExchangeId::BinanceMargin);
5918 assert_eq!(
5919 order.state.id.assigned().map(|id| id.0.as_str()),
5920 Some("42")
5921 );
5922 assert_eq!(order.key.cid.0.as_str(), "cid-9");
5923 assert_eq!(order.side, Side::Buy);
5924 assert_eq!(order.price, Some(Decimal::from(50_000)));
5925 assert_eq!(order.quantity, Decimal::from(2));
5926 assert_eq!(order.state.filled_quantity, Decimal::new(5, 1));
5927 assert_eq!(order.kind, OrderKind::Limit);
5928 }
5929
5930 #[test]
5933 fn margin_open_order_that_is_not_live_is_dropped() {
5934 let inst = InstrumentNameExchange::new("BTCUSDT");
5935 let mut o = open_order(Some(42), "BUY");
5936 o.status = Some("CANCELED".to_string());
5937 assert!(convert_open_order(&o, ExchangeId::BinanceMargin, &inst).is_none());
5938 }
5939
5940 #[test]
5941 fn margin_open_order_missing_order_id_is_dropped() {
5942 let inst = InstrumentNameExchange::new("BTCUSDT");
5943 assert!(
5944 convert_open_order(&open_order(None, "BUY"), ExchangeId::BinanceMargin, &inst)
5945 .is_none()
5946 );
5947 }
5948
5949 #[test]
5955 fn margin_open_order_prefers_update_time_over_creation_time() {
5956 let inst = InstrumentNameExchange::new("BTCUSDT");
5957
5958 let mut updated = open_order(Some(42), "BUY");
5959 updated.update_time = Some(1_700_000_005_000);
5960 let order =
5961 convert_open_order(&updated, ExchangeId::BinanceMargin, &inst).expect("convert");
5962 assert_eq!(
5963 order.state.time_exchange,
5964 Utc.timestamp_millis_opt(1_700_000_005_000)
5965 .single()
5966 .unwrap(),
5967 "update_time must win over time"
5968 );
5969 }
5970
5971 #[test]
5974 fn margin_open_order_falls_back_to_creation_time_when_update_time_absent() {
5975 let inst = InstrumentNameExchange::new("BTCUSDT");
5976
5977 let base = open_order(Some(42), "BUY");
5978 assert!(base.update_time.is_none(), "fixture must omit update_time");
5979 let order = convert_open_order(&base, ExchangeId::BinanceMargin, &inst).expect("convert");
5980 assert_eq!(
5981 order.state.time_exchange,
5982 Utc.timestamp_millis_opt(1_700_000_000_000)
5983 .single()
5984 .unwrap(),
5985 "absent update_time falls back to time"
5986 );
5987 }
5988
5989 #[test]
5990 fn margin_open_order_owned_symbol_recovers_instrument() {
5991 let mut o = open_order(Some(42), "SELL");
5993 o.symbol = Some("ETHUSDT".to_string());
5994 let order =
5995 convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin).expect("convert");
5996 assert_eq!(order.key.instrument.name().as_str(), "ETHUSDT");
5997 assert_eq!(order.key.exchange, ExchangeId::BinanceMargin);
5998 assert_eq!(order.side, Side::Sell);
5999 }
6000
6001 #[test]
6002 fn margin_open_order_owned_symbol_missing_symbol_is_dropped() {
6003 let o = open_order(Some(42), "BUY");
6005 assert!(o.symbol.is_none());
6006 assert!(convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin).is_none());
6007 }
6008
6009 fn trade(id: Option<i64>, is_buyer: Option<bool>) -> QueryMarginAccountsTradeListResponseInner {
6010 let mut t = QueryMarginAccountsTradeListResponseInner::new();
6011 t.id = id;
6012 t.order_id = Some(7);
6013 t.is_buyer = is_buyer;
6014 t.price = Some("48000".to_string());
6015 t.qty = Some("0.25".to_string());
6016 t.commission = Some("0.001".to_string());
6017 t.commission_asset = Some("BNB".to_string());
6018 t.time = Some(1_700_000_000_000);
6019 t
6020 }
6021
6022 #[test]
6023 fn margin_trade_converts() {
6024 let inst = InstrumentNameExchange::new("BTCUSDT");
6025 let tr = convert_margin_trade(&trade(Some(11), Some(false)), &inst).expect("convert");
6026 assert_eq!(tr.id.0.as_str(), "11");
6027 assert_eq!(tr.order_id.0.as_str(), "7");
6028 assert_eq!(tr.side, Side::Sell);
6029 assert_eq!(tr.price, Decimal::from(48_000));
6030 assert_eq!(tr.quantity, Decimal::new(25, 2));
6031 assert_eq!(tr.fees.asset.name().as_str(), "BNB");
6033 assert_eq!(tr.fees.fees, Decimal::new(1, 3));
6034 }
6035
6036 #[test]
6037 fn margin_trade_missing_fields_are_dropped() {
6038 let inst = InstrumentNameExchange::new("BTCUSDT");
6039 assert!(convert_margin_trade(&trade(None, Some(true)), &inst).is_none());
6041 assert!(convert_margin_trade(&trade(Some(11), None), &inst).is_none());
6042 }
6043
6044 #[test]
6047 fn token_renew_after_applies_margin_and_floor() {
6048 let now_ms = Utc::now().timestamp_millis();
6049 let far = token_renew_after(now_ms + 86_400_000).as_secs();
6051 assert!(
6052 far > 80_000 && far <= 86_400,
6053 "expected ~24h-minus-margin, got {far}"
6054 );
6055 assert_eq!(
6057 token_renew_after(now_ms - 10_000).as_secs(),
6058 TOKEN_MIN_LIFETIME_SECS
6059 );
6060 assert_eq!(
6062 token_renew_after(now_ms + 60_000).as_secs(),
6063 TOKEN_MIN_LIFETIME_SECS
6064 );
6065 }
6066
6067 fn push(event: serde_json::Value) -> String {
6072 serde_json::json!({ "subscriptionId": 1, "event": event }).to_string()
6073 }
6074
6075 fn cross_outcome(frame: &str, buf: &mut Vec<UnindexedAccountEvent>) -> FrameOutcome {
6076 convert_margin_user_data_events_with(
6077 frame,
6078 buf,
6079 &mut cross_account_position_handler,
6080 &mut cross_notice_instrument,
6081 )
6082 }
6083
6084 #[test]
6087 fn margin_ws_rpc_ack_yields_no_events() {
6088 let frame = serde_json::json!({ "id": "abc", "status": 200, "result": {} }).to_string();
6089 let mut buf = Vec::new();
6090 assert_eq!(cross_outcome(&frame, &mut buf), FrameOutcome::Response);
6091 assert!(buf.is_empty());
6092 }
6093
6094 #[test]
6097 fn margin_ws_unrecognised_frames_yield_no_events() {
6098 for frame in [
6099 r#"{"e":"eventStreamTerminated","E":1}"#,
6100 r#"{"subscriptionId":1,"event":{"E":1}}"#,
6101 ] {
6102 let mut buf = Vec::new();
6103 assert_eq!(
6104 cross_outcome(frame, &mut buf),
6105 FrameOutcome::Pushed { terminated: false },
6106 "{frame}"
6107 );
6108 assert!(buf.is_empty(), "{frame}");
6109 }
6110 }
6111
6112 #[test]
6115 fn margin_ws_trailing_stop_report_decodes_as_a_trailing_stop() {
6116 for trailing_delta in [serde_json::json!(25), serde_json::json!("25")] {
6117 let frame = push(serde_json::json!({
6118 "e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "STOP_LOSS",
6119 "x": "NEW", "X": "NEW", "i": 12_345_i64, "c": "cid-1", "t": -1_i64,
6120 "p": "0.00000000", "P": "0.00000000", "d": trailing_delta, "D": 1_700_000_000_000_i64,
6121 "q": "2", "f": "GTC", "z": "0", "T": 1_700_000_000_000_i64,
6122 }));
6123 let mut buf = Vec::new();
6124 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6125 let [event] = buf.as_slice() else {
6126 panic!("one snapshot expected for {frame}, got {buf:?}");
6127 };
6128 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
6129 panic!("expected an order snapshot, got {event:?}");
6130 };
6131 assert_eq!(
6132 snapshot.0.kind,
6133 OrderKind::TrailingStop {
6134 offset: Decimal::from(25),
6135 offset_type: TrailingOffsetType::BasisPoints,
6136 activation_price: None,
6137 }
6138 );
6139 assert_eq!(snapshot.0.price, None);
6140 }
6141 }
6142
6143 #[test]
6151 fn margin_ws_execution_report_trade_maps_to_trade_and_order_snapshot() {
6152 let frame = push(serde_json::json!({
6153 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6154 "x": "TRADE", "X": "PARTIALLY_FILLED", "i": 12_345_i64, "c": "cid-1",
6155 "t": 99_i64, "l": "0.5", "L": "48000", "z": "0.5",
6156 "p": "48000", "q": "2", "f": "GTC",
6157 "n": "0.001", "N": "BNB", "T": 1_700_000_000_000_i64,
6158 }));
6159 let mut buf = Vec::new();
6160 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6161 assert_eq!(
6162 buf.len(),
6163 2,
6164 "a fill must produce the execution and the order snapshot, got: {buf:?}"
6165 );
6166 assert_eq!(buf[0].exchange, ExchangeId::BinanceMargin);
6167 match &buf[0].kind {
6168 AccountEventKind::Trade(t) => {
6169 assert_eq!(t.instrument.name().as_str(), "BTCUSDT");
6170 assert_eq!(t.side, Side::Buy);
6171 assert_eq!(t.price, Decimal::from(48_000));
6172 assert_eq!(t.quantity, Decimal::new(5, 1));
6173 assert_eq!(t.fees.asset.name().as_str(), "BNB"); assert_eq!(t.fees.fees, Decimal::new(1, 3));
6175 assert_eq!(t.client_order_id, Some(ClientOrderId::new("cid-1")));
6176 }
6177 other => panic!("expected Trade, got {other:?}"),
6178 }
6179 assert_eq!(buf[1].exchange, ExchangeId::BinanceMargin);
6180 match &buf[1].kind {
6181 AccountEventKind::OrderSnapshot(snap) => {
6182 assert_eq!(
6183 snap.0.quantity,
6184 Decimal::from(2),
6185 "the order's total quantity"
6186 );
6187 match &snap.0.state {
6188 OrderState::Active(ActiveOrderState::Open(open)) => {
6189 assert_eq!(open.id.assigned().map(|id| id.0.as_str()), Some("12345"));
6190 assert_eq!(
6191 open.filled_quantity,
6192 Decimal::new(5, 1),
6193 "the snapshot must carry the order's cumulative filled quantity (z)"
6194 );
6195 assert_eq!(
6196 open.time_exchange.timestamp_millis(),
6197 1_700_000_000_000,
6198 "stamped with the execution's transaction time (T), which is what \
6199 gives the engine's staleness gate an ordering key"
6200 );
6201 }
6202 other => panic!("expected an Open order, got {other:?}"),
6203 }
6204 }
6205 other => panic!("expected OrderSnapshot, got {other:?}"),
6206 }
6207 }
6208
6209 #[test]
6213 fn margin_trade_for_an_order_no_longer_working_emits_no_snapshot() {
6214 for status in ["CANCELED", "EXPIRED", "REJECTED", "PENDING_CANCEL", "NEW"] {
6215 let frame = push(serde_json::json!({
6216 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6217 "x": "TRADE", "X": status, "i": 12_345_i64, "c": "cid-1",
6218 "t": 99_i64, "l": "0.5", "L": "48000", "z": "0.5",
6219 "p": "48000", "q": "2", "f": "GTC", "T": 1_700_000_000_000_i64,
6220 }));
6221 let mut buf = Vec::new();
6222 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6223 assert_eq!(
6224 buf.len(),
6225 1,
6226 "status {status} must produce the execution alone, got: {buf:?}"
6227 );
6228 assert!(
6229 matches!(buf[0].kind, AccountEventKind::Trade(_)),
6230 "the execution must still be reported for status {status}"
6231 );
6232 }
6233 }
6234
6235 #[test]
6236 fn margin_ws_new_report_maps_to_active_order_snapshot() {
6237 let frame = push(serde_json::json!({
6238 "e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "LIMIT",
6239 "x": "NEW", "X": "NEW", "i": 7_i64, "c": "cid-2",
6240 "p": "48000", "q": "1", "f": "GTC", "z": "0", "T": 1_700_000_000_000_i64,
6241 }));
6242 let mut buf = Vec::new();
6243 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6244 assert_eq!(buf.len(), 1);
6245 match &buf[0].kind {
6246 AccountEventKind::OrderSnapshot(snap) => {
6247 assert_eq!(snap.0.side, Side::Sell);
6248 assert_eq!(snap.0.price, Some(Decimal::from(48_000)));
6249 assert_eq!(snap.0.quantity, Decimal::from(1));
6250 assert!(
6251 matches!(snap.0.state, OrderState::Active(_)),
6252 "NEW should be an active (resting) order"
6253 );
6254 }
6255 other => panic!("expected OrderSnapshot, got {other:?}"),
6256 }
6257 }
6258
6259 #[test]
6260 fn margin_ws_canceled_report_maps_to_order_cancelled() {
6261 let frame = push(serde_json::json!({
6262 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6263 "x": "CANCELED", "X": "CANCELED", "i": 8_i64, "c": "cid-3",
6264 "z": "0", "T": 1_700_000_000_000_i64,
6265 }));
6266 let mut buf = Vec::new();
6267 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6268 assert_eq!(buf.len(), 1);
6269 match &buf[0].kind {
6270 AccountEventKind::OrderCancelled(resp) => assert!(resp.state.is_ok()),
6271 other => panic!("expected OrderCancelled, got {other:?}"),
6272 }
6273 }
6274
6275 #[test]
6278 fn margin_ws_trade_without_a_client_order_id_reports_none() {
6279 let frame = push(serde_json::json!({
6280 "e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "MARKET",
6281 "x": "TRADE", "X": "FILLED", "i": 12_345_i64,
6282 "t": 99_i64, "l": "1", "L": "48000", "z": "1",
6283 "q": "1", "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
6284 }));
6285 let mut buf = Vec::new();
6286 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6287 let AccountEventKind::Trade(trade) = &buf[0].kind else {
6288 panic!("expected Trade, got {:?}", buf[0].kind);
6289 };
6290 assert_eq!(trade.order_id.0.as_str(), "12345");
6291 assert_eq!(trade.client_order_id, None);
6292 }
6293
6294 #[test]
6295 fn margin_ws_execution_report_missing_symbol_is_dropped() {
6296 let frame = push(serde_json::json!({
6298 "e": "executionReport", "S": "BUY", "x": "TRADE", "i": 1_i64, "t": 1_i64,
6299 "l": "1", "L": "100", "T": 1_700_000_000_000_i64,
6300 }));
6301 let mut buf = Vec::new();
6302 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6303 assert!(buf.is_empty());
6304 }
6305
6306 #[test]
6307 fn margin_ws_outbound_account_position_maps_to_balance_stream_updates() {
6308 let frame = push(serde_json::json!({
6309 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6310 "B": [
6311 { "a": "USDT", "f": "100.0", "l": "5.0" },
6312 { "a": "BTC", "f": "0.5", "l": "0" },
6313 ],
6314 }));
6315 let mut buf = Vec::new();
6316 assert!(!convert_margin_user_data_events(&frame, &mut buf));
6317 assert_eq!(buf.len(), 2);
6318 for ev in &buf {
6320 assert!(matches!(ev.kind, AccountEventKind::BalanceStreamUpdate(_)));
6321 }
6322 }
6323
6324 #[test]
6325 fn margin_ws_stream_terminated_signals_reconnect() {
6326 let frame = push(serde_json::json!({ "e": "eventStreamTerminated" }));
6327 let mut buf = Vec::new();
6328 assert!(
6329 convert_margin_user_data_events(&frame, &mut buf),
6330 "eventStreamTerminated must signal reconnect"
6331 );
6332 assert!(buf.is_empty());
6333 }
6334
6335 #[test]
6336 fn margin_ws_liability_change_is_observable_only() {
6337 let mut buf = Vec::new();
6340 assert!(!convert_margin_user_data_events(
6341 &push(documented_liability_change("USER_LIABILITY_CHANGE")),
6342 &mut buf
6343 ));
6344 assert!(buf.is_empty());
6345 }
6346
6347 fn only_notice(buf: &[UnindexedAccountEvent]) -> &AccountNotice<InstrumentNameExchange> {
6349 match buf {
6350 [
6351 UnindexedAccountEvent {
6352 exchange: ExchangeId::BinanceMargin,
6353 kind: AccountEventKind::Notice(notice),
6354 },
6355 ] => notice,
6356 other => panic!("expected one BinanceMargin notice, got {other:?}"),
6357 }
6358 }
6359
6360 #[test]
6361 fn margin_ws_margin_level_change_is_forwarded_as_a_notice() {
6362 let mut buf = Vec::new();
6363 assert!(!convert_margin_user_data_events(
6364 &push(documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE")),
6365 &mut buf
6366 ));
6367 assert_eq!(
6368 only_notice(&buf),
6369 &AccountNotice::new(
6370 NoticeKind::MarginCall,
6371 None,
6372 Utc.timestamp_millis_opt(1_573_200_697_110).unwrap(),
6373 Some(Decimal::new(125, 2)),
6374 SmolStr::new("MARGIN_CALL"),
6375 ),
6376 "cross names no instrument"
6377 );
6378 }
6379
6380 #[test]
6381 fn margin_level_statuses_map_to_notice_kinds() {
6382 for (status, kind) in [
6383 ("MARGIN_CALL", NoticeKind::MarginCall),
6384 ("PRE_LIQUIDATION", NoticeKind::LiquidationWarning),
6385 ("FORCE_LIQUIDATION", NoticeKind::Liquidation),
6386 ("NORMAL", NoticeKind::MarginRestored),
6387 ("EXCESSIVE", NoticeKind::MarginRestored),
6388 ("SOME_FUTURE_STATUS", NoticeKind::Other),
6389 ] {
6390 let mut buf = Vec::new();
6391 let frame = push(serde_json::json!({
6392 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "1.5",
6393 "s": status,
6394 }));
6395 convert_margin_user_data_events(&frame, &mut buf);
6396 let notice = only_notice(&buf);
6397 assert_eq!(notice.kind, kind, "{status}");
6398 assert_eq!(notice.status, status, "the raw status is kept");
6399 }
6400 }
6401
6402 #[test]
6403 fn a_margin_level_change_missing_parts_is_still_delivered_where_it_can_be() {
6404 let mut buf = Vec::new();
6406 convert_margin_user_data_events(
6407 &push(serde_json::json!({
6408 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "n/a",
6409 "s": "FORCE_LIQUIDATION",
6410 })),
6411 &mut buf,
6412 );
6413 let notice = only_notice(&buf);
6414 assert_eq!(notice.kind, NoticeKind::Liquidation);
6415 assert_eq!(notice.margin_level, None);
6416
6417 let before = Utc::now();
6419 let mut buf = Vec::new();
6420 convert_margin_user_data_events(
6421 &push(serde_json::json!({ "e": "MARGIN_LEVEL_STATUS_CHANGE", "s": "MARGIN_CALL" })),
6422 &mut buf,
6423 );
6424 assert!(only_notice(&buf).time_exchange >= before);
6425
6426 let mut buf = Vec::new();
6428 convert_margin_user_data_events(
6429 &push(serde_json::json!({
6430 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "1.5",
6431 })),
6432 &mut buf,
6433 );
6434 assert!(buf.is_empty());
6435 }
6436
6437 fn documented_liability_change(e: &str) -> serde_json::Value {
6439 serde_json::json!({
6440 "e": e, "E": 1_573_200_697_110_i64, "a": "BTC", "t": "BORROW",
6441 "p": "1.03453430", "i": "0",
6442 })
6443 }
6444
6445 fn documented_margin_level_change(e: &str) -> serde_json::Value {
6447 serde_json::json!({
6448 "e": e, "E": 1_573_200_697_110_i64, "l": "1.25", "s": "MARGIN_CALL",
6449 })
6450 }
6451
6452 #[test]
6453 fn margin_ws_margin_specific_events_match_their_documented_names() {
6454 for (e, frame, expected) in [
6459 (
6460 "USER_LIABILITY_CHANGE",
6461 documented_liability_change("USER_LIABILITY_CHANGE"),
6462 "USER_LIABILITY_CHANGE (observable",
6463 ),
6464 (
6465 "MARGIN_LEVEL_STATUS_CHANGE",
6466 documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE"),
6467 "MARGIN_LEVEL_STATUS_CHANGE received",
6468 ),
6469 (
6471 "userLiabilityChange",
6472 documented_liability_change("userLiabilityChange"),
6473 "USER_LIABILITY_CHANGE (observable",
6474 ),
6475 (
6476 "marginLevelStatusChange",
6477 documented_margin_level_change("marginLevelStatusChange"),
6478 "MARGIN_LEVEL_STATUS_CHANGE received",
6479 ),
6480 ] {
6481 let mut buf = Vec::new();
6482 let (terminated, messages) =
6483 log_messages(|| convert_margin_user_data_events(&push(frame), &mut buf));
6484 assert!(!terminated, "{e}");
6485 assert_eq!(
6487 buf.len(),
6488 usize::from(expected.starts_with("MARGIN")),
6489 "{e}"
6490 );
6491 assert!(
6492 messages.iter().any(|m| m.contains(expected)),
6493 "{e}: expected its own arm's log, got {messages:?}"
6494 );
6495 assert!(
6496 !messages.iter().any(|m| m.contains("undeserializable")
6497 || m.contains("unhandled")
6498 || m.contains("does not handle")),
6499 "{e}: {messages:?}"
6500 );
6501 }
6502 }
6503
6504 fn log_messages<T>(f: impl FnOnce() -> T) -> (T, Vec<String>) {
6508 use std::sync::{Arc, Mutex};
6509 use tracing_subscriber::layer::SubscriberExt;
6510
6511 #[derive(Default)]
6512 struct Messages(Arc<Mutex<Vec<String>>>);
6513
6514 struct Message<'a>(&'a mut String);
6515
6516 impl tracing::field::Visit for Message<'_> {
6517 fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
6518 if field.name() == "message" {
6519 *self.0 = format!("{value:?}");
6520 }
6521 }
6522 }
6523
6524 impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Messages {
6525 fn on_event(
6526 &self,
6527 event: &tracing::Event<'_>,
6528 _: tracing_subscriber::layer::Context<'_, S>,
6529 ) {
6530 let mut message = String::new();
6531 event.record(&mut Message(&mut message));
6532 self.0.lock().unwrap().push(message);
6533 }
6534 }
6535
6536 let layer = Messages::default();
6537 let messages = Arc::clone(&layer.0);
6538 let value =
6540 tracing::subscriber::with_default(tracing_subscriber::registry().with(layer), f);
6541 let messages = messages.lock().unwrap().clone();
6542 (value, messages)
6543 }
6544
6545 #[test]
6546 fn margin_ws_unhandled_event_types_are_ignored_without_teardown() {
6547 for (e, expected) in [
6550 ("listStatus", "ignoring listStatus"),
6551 ("someFutureEvent", "of a type this client does not handle"),
6553 ] {
6554 let mut buf = Vec::new();
6555 let frame = push(serde_json::json!({ "e": e, "E": 1_700_000_000_000_i64 }));
6556 let (terminated, messages) =
6557 log_messages(|| convert_margin_user_data_events(&frame, &mut buf));
6558 assert!(!terminated, "{e}");
6559 assert!(buf.is_empty(), "{e}");
6560 assert!(
6561 messages.iter().any(|m| m.contains(expected)),
6562 "{e}: {messages:?}"
6563 );
6564 }
6565 }
6566
6567 #[test]
6570 fn margin_trade_event_dedup_by_key() {
6571 let frame = push(serde_json::json!({
6574 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6575 "x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 4_242_i64,
6576 "l": "1", "L": "100", "z": "1", "p": "100", "q": "1", "f": "GTC",
6577 "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
6578 }));
6579 let mut buf = Vec::new();
6580 convert_margin_user_data_events(&frame, &mut buf);
6581 let event = buf.first().expect("a Trade event");
6584 assert!(
6585 matches!(event.kind, AccountEventKind::Trade(_)),
6586 "the execution is the first of the two events a fill produces"
6587 );
6588
6589 let cache = new_dedup_cache();
6590 let first = dedup_key_from_event(event).expect("Trade events have a dedup key");
6592 assert!(!is_duplicate(&cache, first), "first sighting is fresh");
6593 let second = dedup_key_from_event(event).expect("Trade events have a dedup key");
6594 assert!(
6595 is_duplicate(&cache, second),
6596 "second sighting is a duplicate"
6597 );
6598 }
6599
6600 fn drive_ws_pipeline(frames: &[&str]) -> Vec<UnindexedAccountEvent> {
6608 let cache = new_dedup_cache();
6609 let mut delivered = Vec::new();
6610 let mut buf = Vec::new();
6611 for frame in frames {
6612 let _ = convert_margin_user_data_events(frame, &mut buf);
6613 for ev in buf.drain(..) {
6614 if let Some(key) = dedup_key_from_event(&ev)
6615 && is_duplicate(&cache, key)
6616 {
6617 continue;
6618 }
6619 delivered.push(ev);
6620 }
6621 }
6622 delivered
6623 }
6624
6625 fn ws_new(cumulative_filled: &str) -> String {
6626 push(serde_json::json!({
6627 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6628 "x": "NEW", "X": "NEW", "i": 12_345_i64, "c": "client-1",
6629 "q": "2", "p": "100", "f": "GTC", "z": cumulative_filled,
6630 "T": 1_700_000_000_000_i64,
6631 }))
6632 }
6633
6634 fn ws_partial_fill(cumulative_filled: &str) -> String {
6635 push(serde_json::json!({
6636 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6637 "x": "TRADE", "X": "PARTIALLY_FILLED", "i": 12_345_i64, "c": "client-1",
6638 "q": "2", "p": "100", "f": "GTC", "z": cumulative_filled,
6639 "l": "1", "L": "100", "t": 555_i64, "n": "0.1", "N": "USDT",
6640 "T": 1_700_000_001_000_i64,
6641 }))
6642 }
6643
6644 fn open_filled_quantities(events: &[UnindexedAccountEvent]) -> Vec<Decimal> {
6645 events
6646 .iter()
6647 .filter_map(|ev| match &ev.kind {
6648 AccountEventKind::OrderSnapshot(snap) => match &snap.0.state {
6649 OrderState::Active(ActiveOrderState::Open(open)) => Some(open.filled_quantity),
6650 _ => None,
6651 },
6652 _ => None,
6653 })
6654 .collect()
6655 }
6656
6657 #[test]
6665 fn margin_ws_fill_snapshot_survives_the_dedup_gate() {
6666 let delivered = drive_ws_pipeline(&[&ws_new("0"), &ws_partial_fill("1")]);
6667
6668 assert_eq!(
6669 delivered.len(),
6670 3,
6671 "expected the ack snapshot, then the fill's execution and its snapshot, got: {delivered:?}"
6672 );
6673 assert_eq!(
6674 open_filled_quantities(&delivered),
6675 vec![Decimal::ZERO, Decimal::ONE],
6676 "the fill's snapshot must reach the consumer carrying the advanced filled quantity"
6677 );
6678 }
6679
6680 #[test]
6683 fn margin_ws_replayed_frame_is_still_deduplicated() {
6684 let delivered = drive_ws_pipeline(&[
6685 &ws_new("0"),
6686 &ws_partial_fill("1"),
6687 &ws_new("0"),
6688 &ws_partial_fill("1"),
6689 ]);
6690
6691 assert_eq!(
6692 delivered.len(),
6693 3,
6694 "re-delivering both frames must add nothing, got: {delivered:?}"
6695 );
6696 }
6697
6698 #[test]
6704 fn margin_ws_an_ack_and_its_fill_reporting_one_state_deliver_that_state_once() {
6705 let delivered = drive_ws_pipeline(&[&ws_new("1"), &ws_partial_fill("1")]);
6706
6707 assert_eq!(
6708 delivered.len(),
6709 2,
6710 "the ack snapshot and the execution, with the fill's duplicate snapshot collapsed, \
6711 got: {delivered:?}"
6712 );
6713 assert!(
6714 delivered
6715 .iter()
6716 .any(|ev| matches!(ev.kind, AccountEventKind::Trade(_))),
6717 "the execution must reach the consumer regardless"
6718 );
6719 assert_eq!(
6720 open_filled_quantities(&delivered),
6721 vec![Decimal::ONE],
6722 "one snapshot, carrying the state both reports agree on"
6723 );
6724 }
6725
6726 fn push_with_sub(subscription_id: i64, event: serde_json::Value) -> String {
6731 serde_json::json!({ "subscriptionId": subscription_id, "event": event }).to_string()
6732 }
6733
6734 fn isolated_handler(
6737 sub_map: Arc<Mutex<HashMap<i64, InstrumentNameExchange>>>,
6738 base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
6739 ) -> impl FnMut(OutboundAccountPosition, Option<i64>, &mut Vec<UnindexedAccountEvent>) {
6740 move |position, subscription_id, buf| {
6741 route_isolated_account_position(position, subscription_id, &sub_map, &base_quote, buf);
6742 }
6743 }
6744
6745 fn btcusdt() -> InstrumentNameExchange {
6746 InstrumentNameExchange::new("BTCUSDT")
6747 }
6748
6749 fn isolated_balances(buf: &[UnindexedAccountEvent]) -> Vec<(&str, &str, Decimal, Decimal)> {
6752 buf.iter()
6753 .map(|event| match &event.kind {
6754 AccountEventKind::InstrumentBalanceUpdate(ibu) => (
6755 ibu.instrument.name().as_str(),
6756 ibu.balance.asset.name().as_str(),
6757 ibu.balance.update.free,
6758 ibu.balance.update.locked,
6759 ),
6760 other => panic!("expected InstrumentBalanceUpdate, got {other:?}"),
6761 })
6762 .collect()
6763 }
6764
6765 fn token(expiration_time_ms: i64) -> UserListenToken {
6766 UserListenToken {
6767 token: "t".to_string(),
6768 expiration_time_ms,
6769 }
6770 }
6771
6772 #[test]
6773 fn listen_token_query_cross_vs_isolated() {
6774 assert!(build_listen_token_query(None).is_empty());
6776
6777 let q = build_listen_token_query(Some(&btcusdt()));
6779 assert_eq!(q.get("isIsolated").and_then(|v| v.as_str()), Some("TRUE"));
6780 assert_eq!(q.get("symbol").and_then(|v| v.as_str()), Some("BTCUSDT"));
6781 assert_eq!(q.len(), 2);
6782 }
6783
6784 #[test]
6785 fn earliest_token_expiry_picks_min() {
6786 let tokens = vec![
6787 (btcusdt(), token(3_000)),
6788 (InstrumentNameExchange::new("ETHUSDT"), token(1_000)),
6789 (InstrumentNameExchange::new("BNBUSDT"), token(2_000)),
6790 ];
6791 assert_eq!(earliest_token_expiry_ms(&tokens), 1_000);
6792 assert_eq!(earliest_token_expiry_ms(&[]), i64::MAX);
6794 }
6795
6796 #[test]
6797 fn base_quote_map_extracts_base_and_quote() {
6798 let entries = vec![
6799 iso_entry(
6800 "BTCUSDT",
6801 iso_base("BTC", "0", "0", "0", "0"),
6802 iso_quote("USDT", "0", "0", "0", "0"),
6803 None,
6804 None,
6805 None,
6806 ),
6807 iso_entry(
6808 "ETHBTC",
6809 iso_base("ETH", "0", "0", "0", "0"),
6810 iso_quote("BTC", "0", "0", "0", "0"),
6811 None,
6812 None,
6813 None,
6814 ),
6815 ];
6816 let map = build_base_quote_map(&entries);
6817 assert_eq!(
6818 map.get(&btcusdt())
6819 .map(|(b, q)| (b.name().as_str(), q.name().as_str())),
6820 Some(("BTC", "USDT"))
6821 );
6822 assert_eq!(
6825 map.get(&InstrumentNameExchange::new("ETHBTC"))
6826 .map(|(b, q)| (b.name().as_str(), q.name().as_str())),
6827 Some(("ETH", "BTC"))
6828 );
6829 }
6830
6831 #[test]
6832 fn isolated_margin_level_change_names_its_subscriptions_pair_or_none() {
6833 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6834 let notice_on = |subscription_id: i64| {
6835 let mut buf = Vec::new();
6836 let frame = push_with_sub(
6837 subscription_id,
6838 documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE"),
6839 );
6840 assert_eq!(
6841 convert_margin_user_data_events_with(
6842 &frame,
6843 &mut buf,
6844 &mut cross_account_position_handler,
6845 &mut |id| isolated_notice_instrument(id, &sub_map),
6846 ),
6847 FrameOutcome::Pushed { terminated: false }
6848 );
6849 only_notice(&buf).instrument.clone()
6850 };
6851
6852 assert_eq!(notice_on(7), Some(btcusdt()));
6853 assert_eq!(notice_on(99), None);
6855 assert_eq!(isolated_notice_instrument(None, &sub_map), None);
6857 }
6858
6859 #[test]
6860 fn the_two_streams_key_one_notice_alike() {
6861 use crate::client::binance::margin_risk::{RiskFrame, convert_risk_frame};
6863
6864 let change = documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE");
6865 let mut from_user_data = Vec::new();
6866 convert_margin_user_data_events(&push(change.clone()), &mut from_user_data);
6867 let mut from_risk = Vec::new();
6868 let risk_frame = serde_json::json!({ "stream": "listen-key", "data": change });
6869 assert_eq!(
6870 convert_risk_frame(&risk_frame.to_string(), "listen-key", &mut from_risk),
6871 RiskFrame::Continue
6872 );
6873
6874 let dedup = new_dedup_cache();
6875 let key = |events: &[UnindexedAccountEvent]| dedup_key_from_event(&events[0]).unwrap();
6876 assert!(!is_duplicate(&dedup, key(&from_user_data)));
6877 assert!(
6878 is_duplicate(&dedup, key(&from_risk)),
6879 "the risk data stream's copy is dropped"
6880 );
6881 }
6882
6883 #[test]
6884 fn a_notice_from_both_streams_is_delivered_once() {
6885 let dedup = new_dedup_cache();
6889 let notice = |time_ms: i64| {
6890 let mut buf = Vec::new();
6891 convert_margin_user_data_events(
6892 &push(serde_json::json!({
6893 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": time_ms, "l": "1.1",
6894 "s": "MARGIN_CALL",
6895 })),
6896 &mut buf,
6897 );
6898 buf.pop().unwrap()
6899 };
6900 let is_new = |event: &UnindexedAccountEvent| {
6901 !is_duplicate(&dedup, dedup_key_from_event(event).unwrap())
6902 };
6903
6904 assert!(is_new(¬ice(1_700_000_000_000)));
6905 assert!(!is_new(¬ice(1_700_000_000_000)), "second copy dropped");
6906 assert!(is_new(¬ice(1_700_000_060_000)), "a later notice is new");
6907 }
6908
6909 #[test]
6910 fn isolated_outbound_position_routes_to_instrument_balance_update() {
6911 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6912 let base_quote = Arc::new(HashMap::from([(
6913 btcusdt(),
6914 (
6915 AssetNameExchange::new("BTC"),
6916 AssetNameExchange::new("USDT"),
6917 ),
6918 )]));
6919 let mut handler = isolated_handler(sub_map, base_quote);
6920
6921 let frame = push_with_sub(
6923 7,
6924 serde_json::json!({
6925 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6926 "B": [
6927 { "a": "USDT", "f": "1000.0", "l": "50.0" },
6928 { "a": "BTC", "f": "0.5", "l": "0.1" },
6929 ],
6930 }),
6931 );
6932 let mut buf = Vec::new();
6933 assert_eq!(
6934 convert_margin_user_data_events_with(
6935 &frame,
6936 &mut buf,
6937 &mut handler,
6938 &mut cross_notice_instrument,
6939 ),
6940 FrameOutcome::Pushed { terminated: false }
6941 );
6942 assert_eq!(
6945 isolated_balances(&buf),
6946 vec![
6947 ("BTCUSDT", "USDT", Decimal::from(1000), Decimal::from(50)),
6948 ("BTCUSDT", "BTC", Decimal::new(5, 1), Decimal::new(1, 1)),
6949 ]
6950 );
6951 let time = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
6952 assert!(buf.iter().all(|event| matches!(
6953 &event.kind,
6954 AccountEventKind::InstrumentBalanceUpdate(ibu) if ibu.balance.time_exchange == time
6955 )));
6956 }
6957
6958 #[test]
6959 fn isolated_outbound_position_unknown_subscription_dropped() {
6960 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6962 let base_quote = Arc::new(HashMap::from([(
6963 btcusdt(),
6964 (
6965 AssetNameExchange::new("BTC"),
6966 AssetNameExchange::new("USDT"),
6967 ),
6968 )]));
6969 let mut handler = isolated_handler(sub_map, base_quote);
6970
6971 let frame = push_with_sub(
6972 99,
6973 serde_json::json!({
6974 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6975 "B": [{ "a": "BTC", "f": "1", "l": "0" }, { "a": "USDT", "f": "1", "l": "0" }],
6976 }),
6977 );
6978 let mut buf = Vec::new();
6979 assert_eq!(
6980 convert_margin_user_data_events_with(
6981 &frame,
6982 &mut buf,
6983 &mut handler,
6984 &mut cross_notice_instrument,
6985 ),
6986 FrameOutcome::Pushed { terminated: false }
6987 );
6988 assert!(buf.is_empty(), "unmapped subscriptionId frame is dropped");
6989 }
6990
6991 #[test]
6992 fn isolated_outbound_position_with_one_side_emits_that_side() {
6993 for (listed, free) in [("BTC", Decimal::new(5, 1)), ("USDT", Decimal::from(20))] {
6996 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6997 let base_quote = Arc::new(HashMap::from([(
6998 btcusdt(),
6999 (
7000 AssetNameExchange::new("BTC"),
7001 AssetNameExchange::new("USDT"),
7002 ),
7003 )]));
7004 let mut handler = isolated_handler(sub_map, base_quote);
7005
7006 let frame = push_with_sub(
7007 7,
7008 serde_json::json!({
7009 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
7010 "B": [{ "a": listed, "f": free.to_string(), "l": "0" }],
7011 }),
7012 );
7013 let mut buf = Vec::new();
7014 assert_eq!(
7015 convert_margin_user_data_events_with(
7016 &frame,
7017 &mut buf,
7018 &mut handler,
7019 &mut cross_notice_instrument,
7020 ),
7021 FrameOutcome::Pushed { terminated: false }
7022 );
7023 assert_eq!(
7024 isolated_balances(&buf),
7025 vec![("BTCUSDT", listed, free, Decimal::ZERO)],
7026 "only {listed} listed"
7027 );
7028 }
7029 }
7030
7031 #[test]
7032 fn isolated_outbound_position_skips_assets_outside_the_pair() {
7033 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
7036 let base_quote = Arc::new(HashMap::from([(
7037 btcusdt(),
7038 (
7039 AssetNameExchange::new("BTC"),
7040 AssetNameExchange::new("USDT"),
7041 ),
7042 )]));
7043 let mut handler = isolated_handler(sub_map, base_quote);
7044
7045 let mixed = push_with_sub(
7046 7,
7047 serde_json::json!({
7048 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
7049 "B": [{ "a": "BNB", "f": "3", "l": "0" }, { "a": "USDT", "f": "20", "l": "1" }],
7050 }),
7051 );
7052 let mut buf = Vec::new();
7053 assert_eq!(
7054 convert_margin_user_data_events_with(
7055 &mixed,
7056 &mut buf,
7057 &mut handler,
7058 &mut cross_notice_instrument,
7059 ),
7060 FrameOutcome::Pushed { terminated: false }
7061 );
7062 assert_eq!(
7063 isolated_balances(&buf),
7064 vec![("BTCUSDT", "USDT", Decimal::from(20), Decimal::ONE)]
7065 );
7066
7067 let stray_only = push_with_sub(
7068 7,
7069 serde_json::json!({
7070 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
7071 "B": [{ "a": "BNB", "f": "3", "l": "0" }],
7072 }),
7073 );
7074 let mut buf = Vec::new();
7075 assert_eq!(
7076 convert_margin_user_data_events_with(
7077 &stray_only,
7078 &mut buf,
7079 &mut handler,
7080 &mut cross_notice_instrument,
7081 ),
7082 FrameOutcome::Pushed { terminated: false }
7083 );
7084 assert!(buf.is_empty(), "no asset of the pair → nothing emitted");
7085 }
7086
7087 #[test]
7088 fn isolated_outbound_position_unparseable_entry_keeps_the_other_side() {
7089 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
7091 let base_quote = Arc::new(HashMap::from([(
7092 btcusdt(),
7093 (
7094 AssetNameExchange::new("BTC"),
7095 AssetNameExchange::new("USDT"),
7096 ),
7097 )]));
7098 let mut handler = isolated_handler(sub_map, base_quote);
7099
7100 let frame = push_with_sub(
7101 7,
7102 serde_json::json!({
7103 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
7104 "B": [{ "a": "BTC", "f": "x", "l": "0" }, { "a": "USDT", "f": "20", "l": "1" }],
7105 }),
7106 );
7107 let mut buf = Vec::new();
7108 assert_eq!(
7109 convert_margin_user_data_events_with(
7110 &frame,
7111 &mut buf,
7112 &mut handler,
7113 &mut cross_notice_instrument,
7114 ),
7115 FrameOutcome::Pushed { terminated: false }
7116 );
7117 assert_eq!(
7118 isolated_balances(&buf),
7119 vec![("BTCUSDT", "USDT", Decimal::from(20), Decimal::ONE)]
7120 );
7121 }
7122
7123 #[test]
7124 fn isolated_execution_report_routes_by_inner_symbol_independent_of_map() {
7125 let sub_map = Arc::new(Mutex::new(HashMap::<i64, InstrumentNameExchange>::new()));
7129 let base_quote = Arc::new(HashMap::new());
7130 let mut handler = isolated_handler(sub_map, base_quote);
7131
7132 let frame = push_with_sub(
7133 42,
7134 serde_json::json!({
7135 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
7136 "x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 5_i64,
7137 "l": "1", "L": "100", "z": "1", "p": "100", "q": "1", "f": "GTC",
7138 "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
7139 }),
7140 );
7141 let mut buf = Vec::new();
7142 assert_eq!(
7143 convert_margin_user_data_events_with(
7144 &frame,
7145 &mut buf,
7146 &mut handler,
7147 &mut cross_notice_instrument,
7148 ),
7149 FrameOutcome::Pushed { terminated: false }
7150 );
7151 assert_eq!(buf.len(), 2, "the execution and its order snapshot");
7152 match &buf[0].kind {
7153 AccountEventKind::Trade(t) => assert_eq!(t.instrument.name().as_str(), "BTCUSDT"),
7154 other => panic!("expected Trade, got {other:?}"),
7155 }
7156 match &buf[1].kind {
7157 AccountEventKind::OrderSnapshot(snap) => {
7158 assert_eq!(snap.0.key.instrument.name().as_str(), "BTCUSDT");
7159 }
7160 other => panic!("expected OrderSnapshot, got {other:?}"),
7161 }
7162 }
7163
7164 struct MarginMyTradesVenue {
7175 trades: Vec<(i64, i64, i64)>,
7177 end_exclusive: bool,
7178 }
7179
7180 impl wiremock::Respond for MarginMyTradesVenue {
7181 fn respond(&self, request: &wiremock::Request) -> wiremock::ResponseTemplate {
7182 const DAY_MS: i64 = 24 * 60 * 60 * 1000;
7183 let query: std::collections::HashMap<String, String> =
7184 request.url.query_pairs().into_owned().collect();
7185 let param = |name: &str| query.get(name).map(|v| v.parse::<i64>().unwrap());
7186 let limit = usize::try_from(param("limit").unwrap_or(500)).unwrap();
7187 let order = param("orderId");
7188
7189 let mut selected: Vec<&(i64, i64, i64)> = self
7190 .trades
7191 .iter()
7192 .filter(|(_, _, order_id)| order.is_none_or(|order| *order_id == order))
7193 .collect();
7194 if let Some(from_id) = param("fromId") {
7195 selected.retain(|(id, _, _)| *id >= from_id);
7196 } else if order.is_none() {
7197 let start = param("startTime").expect("a query by time sends startTime");
7198 let end = match param("endTime") {
7199 Some(end) if end - start >= DAY_MS => {
7200 return wiremock::ResponseTemplate::new(400).set_body_json(
7201 serde_json::json!({"code": -1127, "msg": "More than 24 hours between startTime and endTime."}),
7202 );
7203 }
7204 Some(end) if self.end_exclusive => end - 1,
7205 Some(end) => end,
7206 None => start + DAY_MS - 1,
7207 };
7208 selected.retain(|(_, time, _)| (start..=end).contains(time));
7209 }
7210
7211 let page: Vec<serde_json::Value> = selected
7212 .into_iter()
7213 .take(limit)
7214 .map(|(id, time, order_id)| {
7215 serde_json::json!({
7216 "id": id, "symbol": "BTCUSDT", "time": time, "orderId": order_id,
7217 "price": "50000", "qty": "0.01", "commission": "0",
7218 "commissionAsset": "USDT", "isBuyer": true, "isMaker": false,
7219 "isBestMatch": true, "isIsolated": true,
7220 })
7221 })
7222 .collect();
7223 wiremock::ResponseTemplate::new(200).set_body_json(page)
7224 }
7225 }
7226
7227 fn since(start: i64) -> MyTradesFrom {
7229 MyTradesFrom::Span {
7230 start,
7231 end: i64::MAX,
7232 }
7233 }
7234
7235 async fn read_margin_trades(
7238 venue: MarginMyTradesVenue,
7239 from: MyTradesFrom,
7240 ) -> (Vec<i64>, Vec<std::collections::HashMap<String, String>>) {
7241 let server = wiremock::MockServer::start().await;
7242 wiremock::Mock::given(wiremock::matchers::method("GET"))
7243 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7244 .respond_with(venue)
7245 .mount(&server)
7246 .await;
7247 let rest = Arc::new(MarginTradingRestApi::from_config(
7248 ConfigurationRestApi::builder()
7249 .api_key("key")
7250 .api_secret("secret")
7251 .base_path(server.uri())
7252 .build()
7253 .unwrap(),
7254 ));
7255
7256 let read = paginate_margin_my_trades(
7257 &rest,
7258 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7259 &InstrumentNameExchange::new("BTCUSDT"),
7260 from,
7261 true,
7262 RequestKind::Query,
7263 )
7264 .await
7265 .unwrap();
7266
7267 let requests = server
7268 .received_requests()
7269 .await
7270 .unwrap()
7271 .iter()
7272 .map(|request| request.url.query_pairs().into_owned().collect())
7273 .collect();
7274 (read.into_iter().map(|t| t.id.unwrap()).collect(), requests)
7275 }
7276
7277 #[tokio::test]
7280 async fn recovered_margin_fills_carry_their_orders_client_order_ids() {
7281 recovered_margin_fills_carry_client_order_ids(true).await;
7282 recovered_margin_fills_carry_client_order_ids(false).await;
7283 }
7284
7285 async fn recovered_margin_fills_carry_client_order_ids(is_isolated: bool) {
7286 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7287 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7288 let fill_time = (disconnect + chrono::Duration::minutes(1)).timestamp_millis();
7289 let server = wiremock::MockServer::start().await;
7290 wiremock::Mock::given(wiremock::matchers::method("GET"))
7291 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7292 .respond_with(MarginMyTradesVenue {
7293 trades: vec![(1, fill_time, 1), (2, fill_time, 2)],
7294 end_exclusive: false,
7295 })
7296 .mount(&server)
7297 .await;
7298 wiremock::Mock::given(wiremock::matchers::method("GET"))
7299 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7300 .and(wiremock::matchers::query_param("orderId", "1"))
7301 .and(wiremock::matchers::query_param(
7302 "isIsolated",
7303 if is_isolated { "TRUE" } else { "FALSE" },
7304 ))
7305 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
7306 serde_json::json!({"symbol": "BTCUSDT", "orderId": 1, "clientOrderId": "repay-1"}),
7307 ))
7308 .mount(&server)
7309 .await;
7310 wiremock::Mock::given(wiremock::matchers::method("GET"))
7311 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7312 .and(wiremock::matchers::query_param("orderId", "2"))
7313 .respond_with(
7314 wiremock::ResponseTemplate::new(400).set_body_json(
7315 serde_json::json!({"code": -2013, "msg": "Order does not exist."}),
7316 ),
7317 )
7318 .mount(&server)
7319 .await;
7320 let rest = Arc::new(MarginTradingRestApi::from_config(
7321 ConfigurationRestApi::builder()
7322 .api_key("key")
7323 .api_secret("secret")
7324 .base_path(server.uri())
7325 .build()
7326 .unwrap(),
7327 ));
7328 let (tx, mut rx) = mpsc::unbounded_channel();
7329 let mut unrecovered = UnrecoveredFills::default();
7330 unrecovered.open(
7331 &[InstrumentNameExchange::new("BTCUSDT")],
7332 disconnect,
7333 reconnect,
7334 );
7335
7336 recover_margin_fills(
7337 &rest,
7338 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7339 &mut unrecovered,
7340 &tx,
7341 &new_dedup_cache(),
7342 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7343 is_isolated,
7344 )
7345 .await;
7346
7347 let mut cids: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
7348 .filter_map(|event| match event.kind {
7349 AccountEventKind::Trade(trade) => Some((
7350 trade.id.0.to_string(),
7351 trade.client_order_id,
7352 trade.order_filled_quantity.is_some(),
7353 )),
7354 _ => None,
7355 })
7356 .collect();
7357 cids.sort();
7358 assert_eq!(
7360 cids,
7361 [
7362 ("1".to_string(), Some(ClientOrderId::new("repay-1")), true),
7363 ("2".to_string(), None, true),
7364 ],
7365 "is_isolated: {is_isolated}"
7366 );
7367 let order_reads = server
7368 .received_requests()
7369 .await
7370 .unwrap()
7371 .iter()
7372 .filter(|request| request.url.path() == "/sapi/v1/margin/order")
7373 .count();
7374 assert_eq!(order_reads, 2, "one read per order");
7375 }
7376
7377 fn venue(trades: Vec<(i64, i64)>) -> MarginMyTradesVenue {
7379 MarginMyTradesVenue {
7380 trades: trades
7381 .into_iter()
7382 .map(|(id, time)| (id, time, id))
7383 .collect(),
7384 end_exclusive: false,
7385 }
7386 }
7387
7388 const HOUR_MS: i64 = 60 * 60 * 1000;
7389
7390 #[tokio::test]
7393 async fn margin_trades_read_a_lone_fill_beyond_the_first_day() {
7394 let now = Utc::now().timestamp_millis();
7395 let (ids, _) = read_margin_trades(
7396 venue(vec![(7, now - 72 * HOUR_MS)]),
7397 since(now - 120 * HOUR_MS),
7398 )
7399 .await;
7400
7401 assert_eq!(ids, vec![7]);
7402 }
7403
7404 #[tokio::test]
7407 async fn margin_trades_read_fills_across_several_windows_once_each() {
7408 let now = Utc::now().timestamp_millis();
7409 let trades = vec![
7410 (1, now - 119 * HOUR_MS),
7411 (2, now - 72 * HOUR_MS),
7412 (3, now - HOUR_MS),
7413 ];
7414 let (ids, _) = read_margin_trades(venue(trades), since(now - 144 * HOUR_MS)).await;
7415
7416 assert_eq!(ids, vec![1, 2, 3]);
7417 }
7418
7419 #[tokio::test]
7422 async fn margin_trades_read_a_trade_on_a_window_boundary_once() {
7423 for end_exclusive in [false, true] {
7424 let now = Utc::now().timestamp_millis();
7425 let from = now - 48 * HOUR_MS;
7426 let venue = MarginMyTradesVenue {
7427 end_exclusive,
7428 ..venue(vec![(5, from + MARGIN_MY_TRADES_WINDOW_MS)])
7429 };
7430 let (ids, _) = read_margin_trades(venue, since(from)).await;
7431
7432 assert_eq!(ids, vec![5], "end_exclusive: {end_exclusive}");
7433 }
7434 }
7435
7436 #[tokio::test]
7438 async fn margin_trades_page_on_by_id_past_a_full_window() {
7439 let now = Utc::now().timestamp_millis();
7440 let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
7441 let trades = (1..=count)
7442 .map(|id| (id, now - 48 * HOUR_MS + id))
7443 .collect();
7444 let (ids, _) = read_margin_trades(venue(trades), since(now - 50 * HOUR_MS)).await;
7445
7446 assert_eq!(ids, (1..=count).collect::<Vec<_>>());
7447 }
7448
7449 #[tokio::test]
7452 async fn margin_trades_page_on_past_a_full_final_window() {
7453 let now = Utc::now().timestamp_millis();
7454 let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
7455 let trades = (1..=count).map(|id| (id, now - HOUR_MS + id)).collect();
7456 let (ids, requests) = read_margin_trades(venue(trades), since(now - 2 * HOUR_MS)).await;
7457
7458 assert_eq!(ids, (1..=count).collect::<Vec<_>>());
7459 assert_eq!(requests.len(), 2);
7460 }
7461
7462 #[tokio::test]
7465 async fn margin_trades_quiet_lookback_sends_one_request_per_window() {
7466 let now = Utc::now().timestamp_millis();
7467 let (ids, requests) =
7468 read_margin_trades(venue(Vec::new()), since(now - 72 * HOUR_MS)).await;
7469
7470 assert!(ids.is_empty());
7471 assert_eq!(requests.len(), 4);
7473 }
7474
7475 #[tokio::test]
7478 async fn margin_trades_read_a_short_gap_in_one_request() {
7479 let now = Utc::now().timestamp_millis();
7480 let (ids, requests) =
7481 read_margin_trades(venue(vec![(9, now - HOUR_MS)]), since(now - 2 * HOUR_MS)).await;
7482
7483 assert_eq!(ids, vec![9]);
7484 assert_eq!(requests.len(), 1);
7485 }
7486
7487 #[tokio::test]
7489 async fn margin_trades_from_the_future_send_no_request() {
7490 let now = Utc::now().timestamp_millis();
7491 let (ids, requests) = read_margin_trades(venue(Vec::new()), since(now + HOUR_MS)).await;
7492
7493 assert!(ids.is_empty());
7494 assert!(requests.is_empty());
7495 }
7496
7497 #[tokio::test]
7500 async fn margin_trades_by_order_page_by_id_without_time_bounds() {
7501 let now = Utc::now().timestamp_millis();
7502 let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
7503 let trades = (1..=2 * count)
7505 .map(|id| {
7506 (
7507 id,
7508 now - 48 * HOUR_MS + id,
7509 if id % 2 == 0 { 42 } else { 7 },
7510 )
7511 })
7512 .collect();
7513 let venue = MarginMyTradesVenue {
7514 trades,
7515 end_exclusive: false,
7516 };
7517 let (ids, requests) = read_margin_trades(venue, MyTradesFrom::Order(42)).await;
7518
7519 assert_eq!(ids, (1..=count).map(|n| 2 * n).collect::<Vec<_>>());
7520 assert_eq!(requests.len(), 2);
7521 assert!(
7522 requests.iter().all(
7523 |query| query.get("orderId").map(String::as_str) == Some("42")
7524 && !query.contains_key("startTime")
7525 && !query.contains_key("endTime")
7526 )
7527 );
7528 }
7529
7530 async fn margin_trades_error(response: wiremock::ResponseTemplate) -> UnindexedClientError {
7533 let server = wiremock::MockServer::start().await;
7534 wiremock::Mock::given(wiremock::matchers::method("GET"))
7535 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7536 .respond_with(response)
7537 .mount(&server)
7538 .await;
7539 let rest = Arc::new(MarginTradingRestApi::from_config(
7540 ConfigurationRestApi::builder()
7541 .api_key("key")
7542 .api_secret("secret")
7543 .base_path(server.uri())
7544 .build()
7545 .unwrap(),
7546 ));
7547
7548 paginate_margin_my_trades(
7549 &rest,
7550 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7551 &InstrumentNameExchange::new("BTCUSDT"),
7552 since(Utc::now().timestamp_millis() - HOUR_MS),
7553 true,
7554 RequestKind::Query,
7555 )
7556 .await
7557 .unwrap_err()
7558 }
7559
7560 #[tokio::test]
7563 async fn margin_trades_report_a_venue_rejection_as_a_non_transient_request_rejection() {
7564 let err = margin_trades_error(wiremock::ResponseTemplate::new(400).set_body_json(
7565 serde_json::json!({"code": -1127, "msg": "More than 24 hours between startTime and endTime."}),
7566 ))
7567 .await;
7568
7569 let UnindexedClientError::Api(ApiError::RequestRejected(msg)) = &err else {
7570 panic!("expected RequestRejected, got {err:?}");
7571 };
7572 assert!(msg.contains("-1127"), "code missing from {msg:?}");
7573 assert!(!err.is_transient());
7574 }
7575
7576 #[tokio::test]
7578 async fn margin_trades_report_an_undecodable_response_as_internal() {
7579 let err = margin_trades_error(
7580 wiremock::ResponseTemplate::new(200)
7581 .set_body_json(serde_json::json!({"unexpected": 1})),
7582 )
7583 .await;
7584
7585 assert!(
7586 matches!(err, UnindexedClientError::Internal(_)),
7587 "got {err:?}"
7588 );
7589 assert!(!err.is_transient());
7590 }
7591
7592 #[tokio::test]
7595 async fn sapi_used_ip_weight_near_the_limit_pauses_queries() {
7596 for (used, pauses) in [("10799", false), ("10800", true), ("not a number", false)] {
7597 let server = wiremock::MockServer::start().await;
7598 wiremock::Mock::given(wiremock::matchers::method("GET"))
7599 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
7600 .respond_with(
7601 wiremock::ResponseTemplate::new(200)
7602 .insert_header("x-sapi-used-ip-weight-1m", used)
7603 .set_body_json(serde_json::json!([])),
7604 )
7605 .mount(&server)
7606 .await;
7607 let rest = Arc::new(MarginTradingRestApi::from_config(
7608 ConfigurationRestApi::builder()
7609 .api_key("key")
7610 .api_secret("secret")
7611 .base_path(server.uri())
7612 .build()
7613 .unwrap(),
7614 ));
7615 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7616
7617 fetch_margin_all_open_orders(rest, Arc::clone(&tracker), false)
7618 .await
7619 .unwrap();
7620
7621 let query_waits = tokio::time::timeout(
7622 Duration::from_millis(50),
7623 tracker.wait_if_blocked(RequestKind::Query),
7624 )
7625 .await
7626 .is_err();
7627 assert_eq!(query_waits, pauses, "used IP weight {used}");
7628 tokio::time::timeout(
7629 Duration::from_millis(50),
7630 tracker.wait_if_blocked(RequestKind::Order),
7631 )
7632 .await
7633 .expect("orders never wait on the pause")
7634 .expect("no cooldown too long to sleep through");
7635 }
7636 }
7637
7638 #[tokio::test]
7640 async fn margin_fill_recovery_does_not_wait_for_the_weight_pause() {
7641 let server = wiremock::MockServer::start().await;
7642 wiremock::Mock::given(wiremock::matchers::method("GET"))
7643 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7644 .respond_with(
7646 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([{
7647 "symbol": "BTCUSDT",
7648 "id": 1,
7649 "orderId": 7,
7650 "price": "100",
7651 "qty": "1",
7652 "commission": "0",
7653 "commissionAsset": "USDT",
7654 "time": Utc::now().timestamp_millis(),
7655 "isBuyer": true,
7656 "isMaker": false,
7657 }])),
7658 )
7659 .expect(2)
7660 .mount(&server)
7661 .await;
7662 let rest = Arc::new(MarginTradingRestApi::from_config(
7663 ConfigurationRestApi::builder()
7664 .api_key("key")
7665 .api_secret("secret")
7666 .base_path(server.uri())
7667 .build()
7668 .unwrap(),
7669 ));
7670 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7671 tracker.throttle(Duration::from_secs(60));
7672 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
7673 let mut unrecovered = UnrecoveredFills::default();
7674 unrecovered.open(
7675 &[InstrumentNameExchange::new("BTCUSDT")],
7676 Utc::now() - chrono::Duration::minutes(1),
7677 Utc::now(),
7678 );
7679
7680 tokio::time::timeout(
7682 Duration::from_millis(800),
7683 recover_margin_fills(
7684 &rest,
7685 &tracker,
7686 &mut unrecovered,
7687 &tx,
7688 &new_dedup_cache(),
7689 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7690 false,
7691 ),
7692 )
7693 .await
7694 .expect("recovery does not wait for the pause");
7695 }
7696
7697 #[tokio::test]
7700 async fn margin_trades_read_over_a_span_stop_at_its_end() {
7701 let now = Utc::now().timestamp_millis();
7702 let trades = vec![
7703 (1, now - 50 * HOUR_MS),
7704 (2, now - 49 * HOUR_MS),
7705 (3, now - 10 * HOUR_MS),
7706 ];
7707 let (ids, _) = read_margin_trades(
7708 venue(trades),
7709 MyTradesFrom::Span {
7710 start: now - 60 * HOUR_MS,
7711 end: now - 20 * HOUR_MS,
7712 },
7713 )
7714 .await;
7715
7716 assert_eq!(ids, vec![1, 2]);
7717 }
7718
7719 #[tokio::test]
7722 async fn a_margin_gap_whose_every_read_fails_is_reported_when_given_up() {
7723 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7724 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7725 let server = wiremock::MockServer::start().await;
7726 wiremock::Mock::given(wiremock::matchers::method("GET"))
7727 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7728 .respond_with(
7729 wiremock::ResponseTemplate::new(400)
7730 .set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
7731 )
7732 .mount(&server)
7733 .await;
7734 let rest = Arc::new(MarginTradingRestApi::from_config(
7735 ConfigurationRestApi::builder()
7736 .api_key("key")
7737 .api_secret("secret")
7738 .base_path(server.uri())
7739 .build()
7740 .expect("valid config"),
7741 ));
7742 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7743 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
7744 let dedup = new_dedup_cache();
7745 let btc = InstrumentNameExchange::new("BTCUSDT");
7746 let mut unrecovered = UnrecoveredFills::default();
7747 unrecovered.open(std::slice::from_ref(&btc), disconnect, reconnect);
7748
7749 for _ in 0..=crate::client::order_recovery::MAX_GAP_RETRIES {
7750 recover_margin_fills(
7751 &rest,
7752 &tracker,
7753 &mut unrecovered,
7754 &tx,
7755 &dedup,
7756 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7757 false,
7758 )
7759 .await;
7760 unrecovered.make_due();
7761 }
7762
7763 assert!(unrecovered.is_empty(), "the gap is given up");
7764 let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7765 let [event] = forwarded.as_slice() else {
7766 panic!("only the give-up is sent: {forwarded:?}");
7767 };
7768 let AccountEventKind::FillRecoveryGaveUp(gave_up) = &event.kind else {
7769 panic!("expected FillRecoveryGaveUp, got {event:?}");
7770 };
7771 assert_eq!(event.exchange, ExchangeId::BinanceMargin);
7772 assert_eq!(
7773 gave_up.scope,
7774 crate::FillRecoveryScope::Instruments(vec![btc])
7775 );
7776 assert_eq!(
7777 gave_up.start.timestamp_millis(),
7778 disconnect.timestamp_millis()
7779 );
7780 assert_eq!(
7781 gave_up.attempts,
7782 crate::client::order_recovery::MAX_GAP_RETRIES + 1
7783 );
7784 assert!(
7785 matches!(&gave_up.reason, FillRecoveryFailure::Request(reason) if reason.contains("rejected")),
7786 "the last read's error: {:?}",
7787 gave_up.reason
7788 );
7789 }
7790
7791 #[tokio::test]
7794 async fn a_failed_margin_recovery_keeps_its_closed_gap_for_the_retry() {
7795 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7796 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7797 let server = wiremock::MockServer::start().await;
7798 wiremock::Mock::given(wiremock::matchers::method("GET"))
7799 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7800 .respond_with(
7801 wiremock::ResponseTemplate::new(400)
7802 .set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
7803 )
7804 .up_to_n_times(1)
7805 .with_priority(1)
7806 .mount(&server)
7807 .await;
7808 wiremock::Mock::given(wiremock::matchers::method("GET"))
7809 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7810 .respond_with(venue(vec![
7811 (
7812 1,
7813 (disconnect + chrono::Duration::minutes(1)).timestamp_millis(),
7814 ),
7815 (
7816 2,
7817 (Utc::now() - chrono::Duration::minutes(1)).timestamp_millis(),
7818 ),
7819 ]))
7820 .mount(&server)
7821 .await;
7822 let rest = Arc::new(MarginTradingRestApi::from_config(
7823 ConfigurationRestApi::builder()
7824 .api_key("key")
7825 .api_secret("secret")
7826 .base_path(server.uri())
7827 .build()
7828 .unwrap(),
7829 ));
7830 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7831 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
7832 let dedup = new_dedup_cache();
7833 let mut unrecovered = UnrecoveredFills::default();
7834 unrecovered.open(
7835 &[InstrumentNameExchange::new("BTCUSDT")],
7836 disconnect,
7837 reconnect,
7838 );
7839
7840 recover_margin_fills(
7841 &rest,
7842 &tracker,
7843 &mut unrecovered,
7844 &tx,
7845 &dedup,
7846 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7847 false,
7848 )
7849 .await;
7850 assert!(!unrecovered.is_empty(), "the failed gap is kept");
7851 assert!(rx.try_recv().is_err(), "nothing forwarded");
7852
7853 unrecovered.make_due();
7854 recover_margin_fills(
7855 &rest,
7856 &tracker,
7857 &mut unrecovered,
7858 &tx,
7859 &dedup,
7860 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7861 false,
7862 )
7863 .await;
7864 assert!(unrecovered.is_empty(), "the retry recovered it");
7865 let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7866 assert_eq!(
7867 forwarded.len(),
7868 1,
7869 "only the fill inside the gap: {forwarded:?}"
7870 );
7871 }
7872
7873 #[tokio::test]
7876 async fn a_timed_out_margin_recovery_charges_only_the_gaps_it_started() {
7877 let server = wiremock::MockServer::start().await;
7878 wiremock::Mock::given(wiremock::matchers::method("GET"))
7879 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7880 .respond_with(
7881 wiremock::ResponseTemplate::new(200)
7882 .set_body_json(serde_json::json!([]))
7883 .set_delay(Duration::from_secs(3_600)),
7884 )
7885 .mount(&server)
7886 .await;
7887 let rest = Arc::new(MarginTradingRestApi::from_config(
7888 ConfigurationRestApi::builder()
7889 .api_key("key")
7890 .api_secret("secret")
7891 .base_path(server.uri())
7892 .timeout(3_600_000_u64)
7893 .retries(0_u32)
7894 .build()
7895 .unwrap(),
7896 ));
7897 let instruments: Vec<_> = (0..9)
7898 .map(|i| InstrumentNameExchange::new(format!("SYM{i}USDT")))
7899 .collect();
7900 let mut unrecovered = UnrecoveredFills::default();
7901 unrecovered.open(
7902 &instruments,
7903 Utc::now() - chrono::Duration::minutes(10),
7904 Utc::now(),
7905 );
7906 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
7907 tokio::time::pause();
7909
7910 recover_margin_fills(
7911 &rest,
7912 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7913 &mut unrecovered,
7914 &tx,
7915 &new_dedup_cache(),
7916 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7917 false,
7918 )
7919 .await;
7920
7921 let now = tokio::time::Instant::now();
7922 assert_eq!(
7923 unrecovered.due(now).len(),
7924 1,
7925 "the gap never started stays due"
7926 );
7927 assert_eq!(
7928 unrecovered
7929 .due(now + Duration::from_secs(crate::client::order_recovery::GAP_RETRY_BASE_SECS))
7930 .len(),
7931 9,
7932 "the eight started wait for their first retry"
7933 );
7934 }
7935
7936 fn margin_order_row(cid: &str, status: &str) -> serde_json::Value {
7942 serde_json::json!({
7943 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid, "price": "100",
7944 "origQty": "2", "executedQty": "1", "cummulativeQuoteQty": "105", "status": status,
7945 "timeInForce": "GTC", "type": "LIMIT", "side": "BUY", "isIsolated": false,
7946 "time": 1_700_000_000_000_i64, "updateTime": 1_700_000_060_000_i64,
7947 })
7948 }
7949
7950 fn margin_key(instrument: &str, cid: &str) -> UnindexedOrderKey {
7951 OrderKey::new(
7952 ExchangeId::BinanceMargin,
7953 InstrumentNameExchange::new(instrument),
7954 StrategyId::new("strategy"),
7955 ClientOrderId::new(cid),
7956 )
7957 }
7958
7959 #[test]
7960 fn a_margin_order_row_says_how_the_order_ended() {
7961 use crate::order::state::InactiveOrderState;
7962 use binance_sdk::margin_trading::rest_api::QueryMarginAccountsOrderResponse;
7963 let key = margin_key("BTCUSDT", "a");
7964 let ended = |row: serde_json::Value| {
7965 let row: QueryMarginAccountsOrderResponse = serde_json::from_value(row).unwrap();
7966 convert_ended_order(&row, ExchangeId::BinanceMargin, &key, None)
7967 .map(|order| order.state)
7968 };
7969
7970 let mut filled = margin_order_row("a", "FILLED");
7971 filled["executedQty"] = "2".into();
7972 filled["cummulativeQuoteQty"] = "210".into();
7973 let Some(InactiveOrderState::FullyFilled(filled)) = ended(filled) else {
7974 panic!("FILLED is fully filled");
7975 };
7976 assert_eq!(filled.avg_price, Some(Decimal::from(105)));
7977 assert!(matches!(
7978 ended(margin_order_row("a", "CANCELED")),
7979 Some(InactiveOrderState::Cancelled(cancelled)) if cancelled.filled_quantity == Some(Decimal::ONE)
7980 ));
7981 assert_eq!(ended(margin_order_row("a", "NEW")), None);
7982
7983 let mut unknown_quote = margin_order_row("a", "FILLED");
7985 unknown_quote["executedQty"] = "2".into();
7986 unknown_quote["cummulativeQuoteQty"] = "-1".into();
7987 let Some(InactiveOrderState::FullyFilled(filled)) = ended(unknown_quote) else {
7988 panic!("still fully filled");
7989 };
7990 assert_eq!(filled.avg_price, None, "no negative average price");
7991 }
7992
7993 fn margin_rest_at(server: &wiremock::MockServer) -> Arc<RestApi> {
7994 Arc::new(MarginTradingRestApi::from_config(
7995 ConfigurationRestApi::builder()
7996 .api_key("key")
7997 .api_secret("secret")
7998 .base_path(server.uri())
7999 .build()
8000 .unwrap(),
8001 ))
8002 }
8003
8004 pub(super) fn margin_client_at(
8006 server: &wiremock::MockServer,
8007 is_isolated: bool,
8008 ) -> BinanceMargin {
8009 let config = if is_isolated {
8010 BinanceMarginConfig::isolated(
8011 "key".into(),
8012 "secret".into(),
8013 vec![InstrumentNameExchange::new("BTCUSDT")],
8014 )
8015 } else {
8016 BinanceMarginConfig::cross_margin("key".into(), "secret".into())
8017 };
8018 let mut client = <BinanceMargin as ExecutionClient>::new(config);
8019 client.rest = margin_rest_at(server);
8020 client
8021 }
8022
8023 async fn mount_margin_order(
8025 server: &wiremock::MockServer,
8026 cid: &str,
8027 response: wiremock::ResponseTemplate,
8028 ) {
8029 wiremock::Mock::given(wiremock::matchers::method("GET"))
8030 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8031 .and(wiremock::matchers::query_param("origClientOrderId", cid))
8032 .respond_with(response)
8033 .mount(server)
8034 .await;
8035 }
8036
8037 fn margin_venue_error(code: i64) -> wiremock::ResponseTemplate {
8038 wiremock::ResponseTemplate::new(400)
8039 .set_body_json(serde_json::json!({"code": code, "msg": "refused"}))
8040 }
8041
8042 fn margin_row_response(cid: &str, status: &str) -> wiremock::ResponseTemplate {
8043 wiremock::ResponseTemplate::new(200).set_body_json(margin_order_row(cid, status))
8044 }
8045
8046 async fn margin_requests(server: &wiremock::MockServer) -> Vec<(String, Option<String>)> {
8048 server
8049 .received_requests()
8050 .await
8051 .unwrap()
8052 .iter()
8053 .map(|request| {
8054 let isolated = request
8055 .url
8056 .query_pairs()
8057 .find(|(key, _)| key == "isIsolated")
8058 .map(|(_, value)| value.into_owned());
8059 (request.url.path().to_owned(), isolated)
8060 })
8061 .collect()
8062 }
8063
8064 #[tokio::test]
8065 async fn margin_fetch_ended_orders_asks_the_configured_account_and_omits_the_rest() {
8066 use crate::order::state::InactiveOrderState;
8067 for (is_isolated, flag) in [(false, "FALSE"), (true, "TRUE")] {
8068 let server = wiremock::MockServer::start().await;
8069 mount_margin_order(&server, "filled", margin_row_response("filled", "FILLED")).await;
8070 mount_margin_order(&server, "live", margin_row_response("live", "NEW")).await;
8071 mount_margin_order(&server, "unknown", margin_venue_error(-2013)).await;
8072 mount_margin_order(&server, "no-symbol", margin_venue_error(-1121)).await;
8073 let client = margin_client_at(&server, is_isolated);
8074 let keys = ["filled", "live", "unknown"].map(|cid| margin_key("BTCUSDT", cid));
8075
8076 let ended = client
8077 .fetch_ended_orders(
8078 &[keys.to_vec(), vec![margin_key("NOSUCH", "no-symbol")]].concat(),
8079 )
8080 .await
8081 .unwrap();
8082
8083 let [order] = ended.as_slice() else {
8084 panic!("one order ended: {ended:?}");
8085 };
8086 assert_eq!(order.key, keys[0], "the strategy asked survives");
8087 assert!(matches!(order.state, InactiveOrderState::FullyFilled(_)));
8088 let requests = margin_requests(&server).await;
8089 assert_eq!(requests.len(), 4);
8090 for (path, isolated) in requests {
8091 assert_eq!(path, "/sapi/v1/margin/order");
8092 assert_eq!(
8093 isolated.as_deref(),
8094 Some(flag),
8095 "is_isolated = {is_isolated}"
8096 );
8097 }
8098 }
8099 }
8100
8101 #[tokio::test]
8102 async fn margin_fetch_ended_orders_fails_when_any_lookup_fails() {
8103 let server = wiremock::MockServer::start().await;
8104 mount_margin_order(
8105 &server,
8106 "cancelled",
8107 margin_row_response("cancelled", "CANCELED"),
8108 )
8109 .await;
8110 mount_margin_order(&server, "refused", margin_venue_error(-1100)).await;
8111 let client = margin_client_at(&server, false);
8112
8113 let result = client
8114 .fetch_ended_orders(&[
8115 margin_key("BTCUSDT", "cancelled"),
8116 margin_key("BTCUSDT", "refused"),
8117 ])
8118 .await;
8119
8120 assert!(
8121 matches!(
8122 result,
8123 Err(UnindexedClientError::Api(ApiError::RequestRejected(_)))
8124 ),
8125 "{result:?}"
8126 );
8127 }
8128
8129 fn held_margin_orders() -> SharedKnownLiveOrders {
8131 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
8132 for (n, cid) in ["live", "gone", "unknown"].into_iter().enumerate() {
8133 known.lock().live(
8134 &margin_key("BTCUSDT", cid),
8135 Decimal::TWO,
8136 OrderKind::Limit,
8137 &Open::new(
8138 VenueOrderId::Assigned(OrderId::new(n.to_string())),
8139 Utc::now(),
8140 Decimal::ZERO,
8141 ),
8142 );
8143 }
8144 known
8145 }
8146
8147 #[tokio::test]
8152 async fn an_ended_margin_trailing_stop_is_reported_with_the_kind_it_was_held_with() {
8153 let server = wiremock::MockServer::start().await;
8154 wiremock::Mock::given(wiremock::matchers::method("GET"))
8155 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
8156 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
8157 .mount(&server)
8158 .await;
8159 let mut row = margin_order_row("trailing", "CANCELED");
8160 row["type"] = "STOP_LOSS".into();
8161 row["price"] = "0.00000000".into();
8162 row["stopPrice"] = "0.00000000".into();
8163 mount_margin_order(
8164 &server,
8165 "trailing",
8166 wiremock::ResponseTemplate::new(200).set_body_json(row),
8167 )
8168 .await;
8169 let trailing_stop = OrderKind::TrailingStop {
8170 offset: Decimal::from(25),
8171 offset_type: TrailingOffsetType::BasisPoints,
8172 activation_price: None,
8173 };
8174 let hold = |known: &SharedKnownLiveOrders| {
8175 known.lock().live(
8176 &margin_key("BTCUSDT", "trailing"),
8177 Decimal::TWO,
8178 trailing_stop,
8179 &Open::new(
8180 VenueOrderId::Assigned(OrderId::new("7")),
8181 Utc::now(),
8182 Decimal::ZERO,
8183 ),
8184 );
8185 };
8186
8187 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
8188 hold(&known);
8189 let mut unchecked = UncheckedOrders::default();
8190 unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
8191 let (tx, mut rx) = mpsc::unbounded_channel();
8192 recover_margin_ended_orders(
8193 &margin_rest_at(&server),
8194 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
8195 &known,
8196 &mut unchecked,
8197 &UnrecoveredFills::default(),
8198 &tx,
8199 false,
8200 )
8201 .await;
8202 let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
8203 let [event] = sent.as_slice() else {
8204 panic!("the trailing stop ended: {sent:?}");
8205 };
8206 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
8207 panic!("an order snapshot: {event:?}");
8208 };
8209 assert_eq!(snapshot.0.kind, trailing_stop);
8210 assert!(matches!(
8211 snapshot.0.state,
8212 OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(_))
8213 ));
8214 assert!(!known.lock().contains(&ClientOrderId::new("trailing")));
8215
8216 let client = margin_client_at(&server, false);
8217 hold(&client.known_live);
8218 let ended = client
8219 .fetch_ended_orders(&[margin_key("BTCUSDT", "trailing")])
8220 .await
8221 .expect("lookup");
8222 let [order] = ended.as_slice() else {
8223 panic!("the trailing stop ended: {ended:?}");
8224 };
8225 assert_eq!(order.kind, trailing_stop);
8226 }
8227
8228 #[tokio::test]
8229 async fn a_margin_reconnect_reports_how_an_unlisted_order_ended_in_the_configured_account() {
8230 for (is_isolated, flag) in [(false, "FALSE"), (true, "TRUE")] {
8231 let server = wiremock::MockServer::start().await;
8232 wiremock::Mock::given(wiremock::matchers::method("GET"))
8233 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
8234 .and(wiremock::matchers::query_param("symbol", "BTCUSDT"))
8235 .respond_with(
8236 wiremock::ResponseTemplate::new(200)
8237 .set_body_json(serde_json::json!([margin_order_row("live", "NEW")])),
8238 )
8239 .mount(&server)
8240 .await;
8241 mount_margin_order(&server, "gone", margin_row_response("gone", "CANCELED")).await;
8242 mount_margin_order(&server, "unknown", margin_venue_error(-2013)).await;
8243 let known = held_margin_orders();
8244 let mut unchecked = UncheckedOrders::default();
8245 unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
8246 let (tx, mut rx) = mpsc::unbounded_channel();
8247
8248 recover_margin_ended_orders(
8249 &margin_rest_at(&server),
8250 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
8251 &known,
8252 &mut unchecked,
8253 &UnrecoveredFills::default(),
8254 &tx,
8255 is_isolated,
8256 )
8257 .await;
8258
8259 let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
8260 let [event] = sent.as_slice() else {
8261 panic!("one order ended: {sent:?}");
8262 };
8263 assert_eq!(event.exchange, ExchangeId::BinanceMargin);
8264 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
8265 panic!("an order snapshot: {event:?}");
8266 };
8267 assert_eq!(snapshot.0.key.cid, ClientOrderId::new("gone"));
8268 assert_eq!(snapshot.0.key.strategy, StrategyId::unknown());
8269 assert!(matches!(
8270 snapshot.0.state,
8271 OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(_))
8272 ));
8273 let mut requests = margin_requests(&server).await;
8274 requests.sort();
8275 let expected = |path: &str| (path.to_owned(), Some(flag.to_owned()));
8276 assert_eq!(
8277 requests,
8278 [
8279 expected("/sapi/v1/margin/openOrders"),
8280 expected("/sapi/v1/margin/order"),
8281 expected("/sapi/v1/margin/order"),
8282 ],
8283 "one listing, and a lookup of each unlisted order only"
8284 );
8285 let known = known.lock();
8286 assert!(known.contains(&ClientOrderId::new("live")));
8287 assert!(!known.contains(&ClientOrderId::new("gone")));
8288 assert!(
8289 !known.contains(&ClientOrderId::new("unknown")),
8290 "asking again cannot help"
8291 );
8292 assert!(unchecked.is_empty());
8293 }
8294 }
8295
8296 fn margin_open_request<'a>(
8297 instrument: &'a InstrumentNameExchange,
8298 cid: &str,
8299 ) -> OrderRequestOpen<ExchangeId, &'a InstrumentNameExchange> {
8300 OrderRequestOpen {
8301 key: OrderKey::new(
8302 ExchangeId::BinanceMargin,
8303 instrument,
8304 StrategyId::new("strategy"),
8305 ClientOrderId::new(cid),
8306 ),
8307 state: crate::order::request::RequestOpen {
8308 side: Side::Buy,
8309 price: Some(Decimal::from(100)),
8310 quantity: Decimal::TWO,
8311 kind: OrderKind::Limit,
8312 time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
8313 position_id: None,
8314 reduce_only: false,
8315 market: None,
8316 },
8317 }
8318 }
8319
8320 #[tokio::test]
8323 async fn a_margin_trailing_stop_is_sent_with_trailing_delta() {
8324 let server = wiremock::MockServer::start().await;
8325 let sent_as_trailing_stop = |request: &wiremock::Request| {
8326 let mut body_as_query = request.url.clone();
8328 body_as_query.set_query(Some(&String::from_utf8_lossy(&request.body)));
8329 let params: Vec<(String, String)> = request
8330 .url
8331 .query_pairs()
8332 .chain(body_as_query.query_pairs())
8333 .map(|(k, v)| (k.into_owned(), v.into_owned()))
8334 .collect();
8335 let has = |key: &str, value: &str| params.iter().any(|(k, v)| k == key && v == value);
8336 has("type", "STOP_LOSS")
8337 && has("trailingDelta", "25")
8338 && has("sideEffectType", "AUTO_BORROW_REPAY")
8339 && !params.iter().any(|(k, _)| k == "stopPrice" || k == "price")
8340 };
8341 wiremock::Mock::given(wiremock::matchers::method("POST"))
8342 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8343 .and(sent_as_trailing_stop)
8344 .respond_with(
8345 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
8346 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "trailing",
8347 "transactTime": 1_700_000_000_000_i64, "price": "0", "origQty": "2",
8348 "executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
8349 "timeInForce": "GTC", "type": "STOP_LOSS", "side": "SELL",
8350 })),
8351 )
8352 .expect(1)
8353 .mount(&server)
8354 .await;
8355 let client = margin_client_at(&server, false);
8356 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
8357 let trailing_request = |cid, offset| {
8358 let mut request = margin_open_request(&btcusdt, cid);
8359 request.state.side = Side::Sell;
8360 request.state.price = None;
8361 request.state.kind = OrderKind::TrailingStop {
8362 offset,
8363 offset_type: TrailingOffsetType::BasisPoints,
8364 activation_price: None,
8365 };
8366 request
8367 };
8368
8369 let placed = client
8370 .open_order(trailing_request("trailing", Decimal::from(25)))
8371 .await;
8372 assert!(
8373 matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
8374 "{placed:?}"
8375 );
8376
8377 let refused = client
8378 .open_order(trailing_request("fractional", Decimal::new(25, 1)))
8379 .await;
8380 assert!(
8381 matches!(
8382 refused.state,
8383 OrderState::Inactive(crate::order::state::InactiveOrderState::OpenFailed(
8384 OrderError::InvalidPrecision(_)
8385 ))
8386 ),
8387 "{refused:?}"
8388 );
8389 }
8390
8391 #[tokio::test]
8392 async fn placing_listing_and_cancelling_margin_orders_feed_the_known_live_set() {
8393 let server = wiremock::MockServer::start().await;
8394 wiremock::Mock::given(wiremock::matchers::method("POST"))
8395 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8396 .respond_with(
8397 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
8398 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "placed",
8399 "transactTime": 1_700_000_000_000_i64, "price": "100", "origQty": "2",
8400 "executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
8401 "timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
8402 })),
8403 )
8404 .mount(&server)
8405 .await;
8406 wiremock::Mock::given(wiremock::matchers::method("DELETE"))
8407 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8408 .respond_with(
8409 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
8410 "symbol": "BTCUSDT", "orderId": "7", "origClientOrderId": "placed",
8411 "clientOrderId": "cancel", "price": "100", "origQty": "2", "executedQty": "0",
8412 "cummulativeQuoteQty": "0", "status": "CANCELED", "timeInForce": "GTC",
8413 "type": "LIMIT", "side": "BUY",
8414 })),
8415 )
8416 .mount(&server)
8417 .await;
8418 wiremock::Mock::given(wiremock::matchers::method("GET"))
8419 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
8420 .respond_with(
8421 wiremock::ResponseTemplate::new(200)
8422 .set_body_json(serde_json::json!([margin_order_row("listed", "NEW")])),
8423 )
8424 .mount(&server)
8425 .await;
8426 let client = margin_client_at(&server, false);
8427 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
8428
8429 let placed = client
8430 .open_order(margin_open_request(&btcusdt, "placed"))
8431 .await;
8432 assert!(
8433 matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
8434 "{placed:?}"
8435 );
8436 assert!(
8437 client
8438 .known_live
8439 .lock()
8440 .contains(&ClientOrderId::new("placed"))
8441 );
8442
8443 client.fetch_open_orders(&[]).await.unwrap();
8444 assert!(
8445 client
8446 .known_live
8447 .lock()
8448 .contains(&ClientOrderId::new("listed"))
8449 );
8450
8451 let cancelled = client
8452 .cancel_order(OrderRequestCancel {
8453 key: OrderKey::new(
8454 ExchangeId::BinanceMargin,
8455 &btcusdt,
8456 StrategyId::new("strategy"),
8457 ClientOrderId::new("placed"),
8458 ),
8459 state: crate::order::request::RequestCancel {
8460 id: Some(VenueOrderId::Assigned(OrderId::new("7"))),
8461 },
8462 })
8463 .await;
8464 assert!(cancelled.state.is_ok(), "{cancelled:?}");
8465 assert!(
8466 !client
8467 .known_live
8468 .lock()
8469 .contains(&ClientOrderId::new("placed"))
8470 );
8471 }
8472
8473 #[tokio::test]
8475 async fn a_cancel_response_without_executed_qty_reports_the_fill_unknown() {
8476 let server = wiremock::MockServer::start().await;
8477 wiremock::Mock::given(wiremock::matchers::method("DELETE"))
8478 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8479 .respond_with(
8480 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
8481 "symbol": "BTCUSDT", "orderId": "7", "origClientOrderId": "placed",
8482 "clientOrderId": "cancel", "price": "100", "origQty": "2",
8483 "status": "CANCELED", "timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
8484 })),
8485 )
8486 .mount(&server)
8487 .await;
8488 let client = margin_client_at(&server, false);
8489
8490 let cancelled = client
8491 .cancel_order(OrderRequestCancel {
8492 key: OrderKey::new(
8493 ExchangeId::BinanceMargin,
8494 &InstrumentNameExchange::new("BTCUSDT"),
8495 StrategyId::new("strategy"),
8496 ClientOrderId::new("placed"),
8497 ),
8498 state: crate::order::request::RequestCancel {
8499 id: Some(VenueOrderId::Assigned(OrderId::new("7"))),
8500 },
8501 })
8502 .await;
8503 let Ok(cancelled) = cancelled.state else {
8504 panic!("expected Ok, got {cancelled:?}");
8505 };
8506 assert_eq!(cancelled.filled_quantity, None);
8507 }
8508
8509 #[test]
8510 fn a_reconnect_opens_a_gap_on_each_stream_instrument_and_checks_those_with_orders_held() {
8511 let [btc, eth, xrp] = ["BTCUSDT", "ETHUSDT", "XRPUSDT"].map(InstrumentNameExchange::new);
8512 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
8513 for (instrument, cid) in [(&btc, "btc"), (&xrp, "xrp")] {
8514 known.lock().live(
8515 &margin_key(instrument.name(), cid),
8516 Decimal::TWO,
8517 OrderKind::Limit,
8518 &Open::new(
8519 VenueOrderId::Assigned(OrderId::new(cid)),
8520 Utc::now(),
8521 Decimal::ZERO,
8522 ),
8523 );
8524 }
8525 let mut unrecovered = UnrecoveredFills::default();
8526 let mut unchecked = UncheckedOrders::default();
8527
8528 open_reconnect_recovery(
8529 &mut unrecovered,
8530 &mut unchecked,
8531 &known,
8532 &[btc.clone(), eth.clone()],
8533 Utc::now() - chrono::Duration::minutes(1),
8534 );
8535
8536 assert!(unrecovered.covers(&btc) && unrecovered.covers(ð));
8537 assert!(!unrecovered.covers(&xrp), "not the stream's");
8538 assert!(unchecked.contains(&btc));
8539 assert!(!unchecked.contains(ð), "no order held");
8540 assert!(!unchecked.contains(&xrp), "its fills are not recovered");
8541 }
8542
8543 #[tokio::test]
8546 async fn a_recovered_margin_fill_that_completes_a_held_order_ends_it() {
8547 let disconnect = Utc::now() - chrono::Duration::minutes(10);
8548 let server = wiremock::MockServer::start().await;
8549 wiremock::Mock::given(wiremock::matchers::method("GET"))
8550 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
8551 .respond_with(venue(vec![
8552 (
8553 1,
8554 (disconnect + chrono::Duration::minutes(1)).timestamp_millis(),
8555 ),
8556 (
8557 2,
8558 (disconnect + chrono::Duration::minutes(2)).timestamp_millis(),
8559 ),
8560 ]))
8561 .mount(&server)
8562 .await;
8563 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
8566 for (cid, order_id, quantity) in [("one", "1", "0.01"), ("two", "2", "0.02")] {
8567 known.lock().live(
8568 &margin_key("BTCUSDT", cid),
8569 Decimal::from_str(quantity).unwrap(),
8570 OrderKind::Limit,
8571 &Open::new(
8572 VenueOrderId::Assigned(OrderId::new(order_id)),
8573 Utc::now(),
8574 Decimal::ZERO,
8575 ),
8576 );
8577 }
8578 let mut unrecovered = UnrecoveredFills::default();
8579 unrecovered.open(
8580 &[InstrumentNameExchange::new("BTCUSDT")],
8581 disconnect,
8582 Utc::now(),
8583 );
8584 let (tx, mut rx) = mpsc::unbounded_channel();
8585
8586 recover_margin_fills(
8587 &margin_rest_at(&server),
8588 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
8589 &mut unrecovered,
8590 &tx,
8591 &new_dedup_cache(),
8592 &known,
8593 false,
8594 )
8595 .await;
8596
8597 assert_eq!(
8598 std::iter::from_fn(|| rx.try_recv().ok()).count(),
8599 2,
8600 "both fills"
8601 );
8602 let known = known.lock();
8603 assert!(!known.contains(&ClientOrderId::new("one")), "completed");
8604 assert!(known.contains(&ClientOrderId::new("two")), "part-filled");
8605 }
8606
8607 #[tokio::test]
8610 async fn a_margin_order_that_expired_on_placement_is_not_reported_or_held_open() {
8611 use rust_decimal_macros::dec;
8612
8613 for (cid, filled, quote) in [("unfilled", "0", "0"), ("partly", "1", "100")] {
8614 let server = wiremock::MockServer::start().await;
8615 wiremock::Mock::given(wiremock::matchers::method("POST"))
8616 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8617 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
8618 serde_json::json!({
8619 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid,
8620 "transactTime": 1_700_000_000_000_i64, "price": "100", "origQty": "2",
8621 "executedQty": filled, "cummulativeQuoteQty": quote,
8622 "status": "EXPIRED", "timeInForce": "IOC", "type": "LIMIT", "side": "BUY",
8623 }),
8624 ))
8625 .mount(&server)
8626 .await;
8627 let client = margin_client_at(&server, false);
8628 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
8629
8630 let placed = client.open_order(margin_open_request(&btcusdt, cid)).await;
8631
8632 let OrderState::Inactive(crate::order::state::InactiveOrderState::Expired(expired)) =
8633 &placed.state
8634 else {
8635 panic!("{cid}: expired, not open: {placed:?}");
8636 };
8637 assert_eq!(expired.filled_quantity, Decimal::from_str(filled).ok());
8638 let late = Open::new(
8639 VenueOrderId::Assigned(OrderId::new("7")),
8640 Utc::now(),
8641 Decimal::ZERO,
8642 );
8643 let mut known = client.known_live.lock();
8644 known.live(&placed.key, dec!(2), OrderKind::Limit, &late);
8645 assert!(
8646 !known.contains(&ClientOrderId::new(cid)),
8647 "{cid}: not held, even after a late live report"
8648 );
8649 }
8650 }
8651
8652 async fn mount_borrow_terms(
8656 server: &wiremock::MockServer,
8657 pair: serde_json::Value,
8658 asset: serde_json::Value,
8659 rate: serde_json::Value,
8660 max_borrow: serde_json::Value,
8661 ) {
8662 use wiremock::{
8663 Mock, ResponseTemplate,
8664 matchers::{method, path},
8665 };
8666 for (route, body) in [
8667 ("/sapi/v1/margin/allPairs", pair.clone()),
8668 ("/sapi/v1/margin/isolated/allPairs", pair),
8669 ("/sapi/v1/margin/allAssets", asset),
8670 ("/sapi/v1/margin/next-hourly-interest-rate", rate),
8671 ("/sapi/v1/margin/maxBorrowable", max_borrow),
8672 ("/sapi/v1/margin/delist-schedule", serde_json::json!([])),
8673 ] {
8674 Mock::given(method("GET"))
8675 .and(path(route))
8676 .respond_with(ResponseTemplate::new(200).set_body_json(body))
8677 .mount(server)
8678 .await;
8679 }
8680 }
8681
8682 async fn mount_delist_schedule(server: &wiremock::MockServer, schedule: serde_json::Value) {
8685 use wiremock::{
8686 Mock, ResponseTemplate,
8687 matchers::{method, path},
8688 };
8689 Mock::given(method("GET"))
8690 .and(path("/sapi/v1/margin/delist-schedule"))
8691 .respond_with(ResponseTemplate::new(200).set_body_json(schedule))
8692 .with_priority(1)
8693 .mount(server)
8694 .await;
8695 }
8696
8697 fn btcusdt_pair(is_margin_trade: bool, is_sell_allowed: bool) -> serde_json::Value {
8698 serde_json::json!([{
8699 "base": "BTC", "quote": "USDT", "symbol": "BTCUSDT",
8700 "isMarginTrade": is_margin_trade, "isSellAllowed": is_sell_allowed, "isBuyAllowed": true,
8701 }])
8702 }
8703
8704 fn btc_asset(is_borrowable: bool) -> serde_json::Value {
8705 serde_json::json!([{ "assetName": "BTC", "isBorrowable": is_borrowable }])
8706 }
8707
8708 fn btc_rate(rate: &str) -> serde_json::Value {
8709 serde_json::json!([{ "asset": "BTC", "nextHourlyInterestRate": rate }])
8710 }
8711
8712 async fn query_value(server: &wiremock::MockServer, route: &str, key: &str) -> Option<String> {
8714 server
8715 .received_requests()
8716 .await
8717 .unwrap()
8718 .iter()
8719 .find(|request| request.url.path() == route)
8720 .and_then(|request| {
8721 request
8722 .url
8723 .query_pairs()
8724 .find(|(name, _)| name == key)
8725 .map(|(_, value)| value.into_owned())
8726 })
8727 }
8728
8729 #[tokio::test]
8730 async fn margin_shortability_reads_the_pair_the_asset_and_the_annualised_rate() {
8731 use crate::client::ShortabilityClient;
8732 use rust_decimal_macros::dec;
8733 for (is_isolated, pairs_route, flag) in [
8734 (false, "/sapi/v1/margin/allPairs", "FALSE"),
8735 (true, "/sapi/v1/margin/isolated/allPairs", "TRUE"),
8736 ] {
8737 let server = wiremock::MockServer::start().await;
8738 mount_borrow_terms(
8739 &server,
8740 btcusdt_pair(true, true),
8741 btc_asset(true),
8742 btc_rate("0.00000571"),
8743 serde_json::json!({}),
8744 )
8745 .await;
8746
8747 let shortability = margin_client_at(&server, is_isolated)
8748 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8749 .await
8750 .unwrap();
8751
8752 assert_eq!(
8753 shortability,
8754 Shortability::new()
8755 .with_shortable(true)
8756 .with_fee_rate(dec!(0.00000571) * dec!(8760)),
8757 "isolated: {is_isolated}"
8758 );
8759 assert_eq!(
8760 query_value(&server, pairs_route, "symbol").await.as_deref(),
8761 Some("BTCUSDT"),
8762 "the pair list is asked for the one symbol"
8763 );
8764 assert_eq!(
8765 query_value(&server, "/sapi/v1/margin/allAssets", "asset")
8766 .await
8767 .as_deref(),
8768 Some("BTC"),
8769 "the base asset is the one borrowed"
8770 );
8771 let rate = "/sapi/v1/margin/next-hourly-interest-rate";
8772 assert_eq!(
8773 query_value(&server, rate, "assets").await.as_deref(),
8774 Some("BTC")
8775 );
8776 assert_eq!(
8777 query_value(&server, rate, "isIsolated").await.as_deref(),
8778 Some(flag)
8779 );
8780 }
8781 }
8782
8783 #[tokio::test]
8784 async fn margin_shortability_is_refused_by_any_flag_and_unknown_without_them() {
8785 use crate::client::ShortabilityClient;
8786 let cases = [
8787 (btcusdt_pair(false, true), btc_asset(true), Some(false)),
8788 (btcusdt_pair(true, false), btc_asset(true), Some(false)),
8789 (btcusdt_pair(true, true), btc_asset(false), Some(false)),
8790 (btcusdt_pair(true, true), serde_json::json!([]), None),
8792 ];
8793 for (pair, asset, shortable) in cases {
8794 let server = wiremock::MockServer::start().await;
8795 mount_borrow_terms(
8796 &server,
8797 pair.clone(),
8798 asset.clone(),
8799 serde_json::json!([]),
8800 serde_json::json!({}),
8801 )
8802 .await;
8803
8804 let shortability = margin_client_at(&server, false)
8805 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8806 .await
8807 .unwrap();
8808
8809 assert_eq!(shortability.shortable, shortable, "{pair} {asset}");
8810 assert_eq!(
8811 shortability.fee_rate, None,
8812 "no rate returned is an unknown fee"
8813 );
8814 assert_eq!(shortability.easy_to_borrow, None);
8815 assert_eq!(shortability.available, None);
8816 }
8817 }
8818
8819 #[tokio::test]
8823 async fn margin_shortability_reads_the_delisting_schedule_for_this_margin() {
8824 use crate::client::ShortabilityClient;
8825 let past = Utc::now() - chrono::Duration::hours(1);
8826 let soon = Utc::now() + chrono::Duration::days(3);
8827 let later = soon + chrono::Duration::days(7);
8828 let entry = |time: DateTime<Utc>, assets: &[&str], symbols: &[&str]| {
8829 serde_json::json!({
8830 "delistTime": time.timestamp_millis(),
8831 "crossMarginAssets": assets,
8832 "isolatedMarginSymbols": symbols,
8833 })
8834 };
8835 let millis = |time: DateTime<Utc>| DateTime::from_timestamp_millis(time.timestamp_millis());
8836 let cases = [
8837 (
8839 false,
8840 vec![entry(soon, &["BTC"], &[])],
8841 millis(soon),
8842 Some(true),
8843 ),
8844 (
8845 false,
8846 vec![entry(soon, &["USDT"], &[])],
8847 millis(soon),
8848 Some(true),
8849 ),
8850 (
8851 false,
8852 vec![
8853 entry(later, &["BTC"], &[]),
8854 entry(soon, &["ETH", "USDT"], &[]),
8855 ],
8856 millis(soon),
8857 Some(true),
8858 ),
8859 (
8860 false,
8861 vec![entry(soon, &["ETH"], &["BTCUSDT"])],
8862 None,
8863 Some(true),
8864 ),
8865 (
8866 false,
8867 vec![entry(past, &["BTC"], &[])],
8868 millis(past),
8869 Some(false),
8870 ),
8871 (
8873 true,
8874 vec![entry(soon, &[], &["BTCUSDT"])],
8875 millis(soon),
8876 Some(true),
8877 ),
8878 (
8879 true,
8880 vec![entry(soon, &["BTC", "USDT"], &["ETHUSDT"])],
8881 None,
8882 Some(true),
8883 ),
8884 (
8885 true,
8886 vec![entry(past, &[], &["BTCUSDT"])],
8887 millis(past),
8888 Some(false),
8889 ),
8890 ];
8891 for (is_isolated, schedule, delisting_at, shortable) in cases {
8892 let server = wiremock::MockServer::start().await;
8893 mount_borrow_terms(
8894 &server,
8895 btcusdt_pair(true, true),
8896 btc_asset(true),
8897 serde_json::json!([]),
8898 serde_json::json!({}),
8899 )
8900 .await;
8901 let schedule = serde_json::Value::from(schedule);
8902 mount_delist_schedule(&server, schedule.clone()).await;
8903
8904 let shortability = margin_client_at(&server, is_isolated)
8905 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8906 .await
8907 .unwrap();
8908
8909 let case = format!("isolated: {is_isolated}, {schedule}");
8910 assert_eq!(shortability.delisting_at, delisting_at, "{case}");
8911 assert_eq!(shortability.shortable, shortable, "{case}");
8912 }
8913 }
8914
8915 #[tokio::test]
8917 async fn margin_shortability_of_a_scheduled_pair_without_a_time_is_an_error() {
8918 use crate::client::ShortabilityClient;
8919 let server = wiremock::MockServer::start().await;
8920 mount_borrow_terms(
8921 &server,
8922 btcusdt_pair(true, true),
8923 btc_asset(true),
8924 serde_json::json!([]),
8925 serde_json::json!({}),
8926 )
8927 .await;
8928 mount_delist_schedule(
8929 &server,
8930 serde_json::json!([{ "crossMarginAssets": ["BTC"] }]),
8931 )
8932 .await;
8933
8934 let result = margin_client_at(&server, false)
8935 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8936 .await;
8937
8938 assert!(
8939 matches!(result, Err(UnindexedClientError::Internal(ref message))
8940 if message.contains("delistTime")),
8941 "{result:?}"
8942 );
8943 }
8944
8945 #[tokio::test]
8946 async fn margin_shortability_of_an_unlisted_pair_is_an_invalid_instrument() {
8947 use crate::client::ShortabilityClient;
8948 let server = wiremock::MockServer::start().await;
8949 mount_borrow_terms(
8950 &server,
8951 serde_json::json!([]),
8952 btc_asset(true),
8953 btc_rate("0.00000571"),
8954 serde_json::json!({}),
8955 )
8956 .await;
8957
8958 let result = margin_client_at(&server, false)
8959 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8960 .await;
8961
8962 assert!(
8963 matches!(
8964 result,
8965 Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(ref instrument, _)))
8966 if instrument.name().as_str() == "BTCUSDT"
8967 ),
8968 "{result:?}"
8969 );
8970 }
8971
8972 #[tokio::test]
8973 async fn margin_borrow_capacity_asks_for_the_asset_the_side_borrows() {
8974 use crate::client::BorrowCapacityClient;
8975 use rust_decimal_macros::dec;
8976 for (is_isolated, side, asset, isolated_symbol) in [
8977 (false, Side::Sell, "BTC", None),
8978 (false, Side::Buy, "USDT", None),
8979 (true, Side::Sell, "BTC", Some("BTCUSDT")),
8980 ] {
8981 let server = wiremock::MockServer::start().await;
8982 mount_borrow_terms(
8983 &server,
8984 btcusdt_pair(true, true),
8985 btc_asset(true),
8986 btc_rate("0.00000571"),
8987 serde_json::json!({ "amount": "1.69248805", "borrowLimit": "60" }),
8988 )
8989 .await;
8990
8991 let capacity = margin_client_at(&server, is_isolated)
8992 .fetch_borrow_capacity(&InstrumentNameExchange::new("BTCUSDT"), side)
8993 .await
8994 .unwrap();
8995
8996 assert_eq!(
8997 capacity,
8998 BorrowCapacity::new(AssetNameExchange::new(asset))
8999 .with_borrowable_now(dec!(1.69248805))
9000 .with_account_limit(dec!(60)),
9001 "{side:?} isolated: {is_isolated}"
9002 );
9003 let max_borrow = "/sapi/v1/margin/maxBorrowable";
9004 assert_eq!(
9005 query_value(&server, max_borrow, "asset").await.as_deref(),
9006 Some(asset)
9007 );
9008 assert_eq!(
9009 query_value(&server, max_borrow, "isolatedSymbol")
9010 .await
9011 .as_deref(),
9012 isolated_symbol,
9013 "only isolated margin names the pair's account"
9014 );
9015 }
9016 }
9017
9018 #[tokio::test]
9019 async fn margin_borrow_capacity_keeps_an_omitted_amount_unknown_and_fails_a_malformed_one() {
9020 use crate::client::BorrowCapacityClient;
9021 use rust_decimal_macros::dec;
9022 let instrument = InstrumentNameExchange::new("BTCUSDT");
9023
9024 let server = wiremock::MockServer::start().await;
9025 mount_borrow_terms(
9026 &server,
9027 btcusdt_pair(true, true),
9028 btc_asset(true),
9029 btc_rate("0"),
9030 serde_json::json!({ "borrowLimit": "60" }),
9031 )
9032 .await;
9033 let capacity = margin_client_at(&server, false)
9034 .fetch_borrow_capacity(&instrument, Side::Sell)
9035 .await
9036 .unwrap();
9037 assert_eq!(capacity.borrowable_now, None);
9038 assert_eq!(capacity.account_limit, Some(dec!(60)));
9039
9040 let server = wiremock::MockServer::start().await;
9041 mount_borrow_terms(
9042 &server,
9043 btcusdt_pair(true, true),
9044 btc_asset(true),
9045 btc_rate("0"),
9046 serde_json::json!({ "amount": "lots", "borrowLimit": "60" }),
9047 )
9048 .await;
9049 let result = margin_client_at(&server, false)
9050 .fetch_borrow_capacity(&instrument, Side::Sell)
9051 .await;
9052 assert!(
9053 matches!(result, Err(UnindexedClientError::Internal(ref message)) if message.contains("amount")),
9054 "{result:?}"
9055 );
9056 }
9057
9058 #[tokio::test]
9059 async fn margin_borrow_capacity_without_an_isolated_account_names_the_pair() {
9060 use crate::client::BorrowCapacityClient;
9061 use wiremock::{
9062 Mock,
9063 matchers::{method, path},
9064 };
9065 let server = wiremock::MockServer::start().await;
9066 Mock::given(method("GET"))
9067 .and(path("/sapi/v1/margin/maxBorrowable"))
9068 .respond_with(
9069 wiremock::ResponseTemplate::new(400).set_body_json(serde_json::json!({
9070 "code": -11001, "msg": "Isolated margin account does not exist."
9071 })),
9072 )
9073 .mount(&server)
9074 .await;
9075 mount_borrow_terms(
9076 &server,
9077 btcusdt_pair(true, true),
9078 btc_asset(true),
9079 btc_rate("0"),
9080 serde_json::json!({}),
9081 )
9082 .await;
9083
9084 let err = margin_client_at(&server, true)
9085 .fetch_borrow_capacity(&InstrumentNameExchange::new("BTCUSDT"), Side::Sell)
9086 .await
9087 .unwrap_err();
9088
9089 assert_eq!(
9090 err.to_string(),
9091 "API: request rejected: BTCUSDT: -11001 Isolated margin account does not exist."
9092 );
9093 assert!(!err.is_transient());
9094 }
9095}