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, ExecutionReport, ExponentialBackoff, 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, UserDataFrame,
72 WeightPool, binance_filled_qty, classify_order_kind_tif, classify_rest_order_error,
73 classify_rest_query_error, convert_ended_order, convert_execution_report,
74 convert_open_order_listing, convert_open_order_owned_symbol, dedup_key_from_event, drop_after,
75 frame_excerpt, gap_failed, gap_time, is_duplicate, is_unknown_order, log_unhandled_event,
76 log_unrecognised_frame, new_dedup_cache, parse_user_data_frame, placed_order_state,
77 recovered_order_cids, recovered_order_totals, response_decode_error, rest_call_with_retry,
78 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 },
93 emit_stream_terminated,
94 error::{
95 ApiError, ConnectivityError, OrderError, StreamTerminationReason, UnindexedClientError,
96 },
97 order::{
98 Order, OrderKey, OrderKind, TimeInForce, UnindexedInactiveOrder, UnindexedOrderKey,
99 id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
100 request::{OrderRequestCancel, OrderRequestOpen, UnindexedOrderResponseCancel},
101 state::{Cancelled, Open, OrderState, UnindexedOrderState},
102 },
103 position::PositionReport,
104 shortability::Shortability,
105 trade::{AssetFees, Trade, TradeId, TradesRead},
106};
107use binance_sdk::{
108 common::{
109 config::{ConfigurationRestApi, ConfigurationWebsocketApi},
110 constants::{MARGIN_TRADING_REST_API_PROD_URL, SPOT_WS_API_PROD_URL},
111 models::WebsocketEvent,
112 websocket::{
113 SendWebsocketMessageResult, Subscription, WebsocketApi as WsApiBase,
114 WebsocketMessageSendOptions,
115 },
116 },
117 margin_trading::{
118 MarginTradingRestApi,
119 rest_api::{
120 GetAllCrossMarginPairsParams, GetAllIsolatedMarginSymbolParams,
121 GetAllMarginAssetsParams, GetFutureHourlyInterestRateIsIsolatedEnum,
122 GetFutureHourlyInterestRateParams, MarginAccountCancelOrderIsIsolatedEnum,
123 MarginAccountCancelOrderParams, MarginAccountNewOrderIsIsolatedEnum,
124 MarginAccountNewOrderNewOrderRespTypeEnum, MarginAccountNewOrderParams,
125 MarginAccountNewOrderSideEffectTypeEnum, MarginAccountNewOrderSideEnum,
126 MarginAccountNewOrderTimeInForceEnum, MarginAccountNewOrderTypeEnum,
127 QueryCrossMarginAccountDetailsParams,
128 QueryCrossMarginAccountDetailsResponseUserAssetsInner,
129 QueryIsolatedMarginAccountInfoParams,
130 QueryIsolatedMarginAccountInfoResponseAssetsInner,
131 QueryMarginAccountsOpenOrdersIsIsolatedEnum, QueryMarginAccountsOpenOrdersParams,
132 QueryMarginAccountsOrderIsIsolatedEnum, QueryMarginAccountsOrderParams,
133 QueryMarginAccountsTradeListIsIsolatedEnum, QueryMarginAccountsTradeListParams,
134 QueryMarginAccountsTradeListResponseInner, QueryMaxBorrowParams, RestApi,
135 },
136 websocket_streams::{
137 MarginLevelStatusChange, OutboundAccountPosition, UserLiabilityChange,
138 },
139 },
140};
141use chrono::{DateTime, TimeZone, Utc};
142use fnv::FnvHashSet;
143use futures::stream::BoxStream;
144use rust_decimal::Decimal;
145use rustrade_instrument::{
146 Side,
147 asset::name::AssetNameExchange,
148 exchange::ExchangeId,
149 instrument::{kind::InstrumentKindDiscriminant, name::InstrumentNameExchange},
150};
151use serde::{Deserialize, Serialize};
152use smol_str::{SmolStr, format_smolstr};
153use std::{
154 collections::{BTreeMap, HashMap},
155 str::FromStr,
156 sync::{
157 Arc, Mutex,
158 atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
159 },
160 time::Duration,
161};
162use tokio::sync::{mpsc, oneshot};
163use tracing::{debug, error, info, trace, warn};
164
165#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
183#[serde(rename_all = "snake_case")]
184pub enum MarginSideEffect {
185 #[default]
193 AutoBorrowRepay,
194 NoBorrow,
198}
199
200impl MarginSideEffect {
201 pub fn as_binance_str(self) -> &'static str {
203 match self {
204 MarginSideEffect::AutoBorrowRepay => "AUTO_BORROW_REPAY",
205 MarginSideEffect::NoBorrow => "NO_SIDE_EFFECT",
206 }
207 }
208
209 fn as_sdk_enum(self) -> MarginAccountNewOrderSideEffectTypeEnum {
215 match self {
216 MarginSideEffect::AutoBorrowRepay => {
217 MarginAccountNewOrderSideEffectTypeEnum::AutoBorrowRepay
218 }
219 MarginSideEffect::NoBorrow => MarginAccountNewOrderSideEffectTypeEnum::NoSideEffect,
220 }
221 }
222}
223
224#[derive(Clone, Deserialize)]
234pub struct BinanceMarginConfig {
235 api_key: String,
238 secret_key: String,
239 pub testnet: bool,
245 #[serde(default)]
252 pub is_isolated: bool,
253 #[serde(default)]
264 pub isolated_symbols: Vec<InstrumentNameExchange>,
265 #[serde(default)]
269 pub side_effect: MarginSideEffect,
270}
271
272impl std::fmt::Debug for BinanceMarginConfig {
274 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
275 f.debug_struct("BinanceMarginConfig")
276 .field("api_key", &"***")
277 .field("secret_key", &"***")
278 .field("testnet", &self.testnet)
279 .field("is_isolated", &self.is_isolated)
280 .field("isolated_symbols", &self.isolated_symbols)
281 .field("side_effect", &self.side_effect)
282 .finish()
283 }
284}
285
286impl BinanceMarginConfig {
287 pub fn new(
297 api_key: String,
298 secret_key: String,
299 testnet: bool,
300 is_isolated: bool,
301 isolated_symbols: Vec<InstrumentNameExchange>,
302 side_effect: MarginSideEffect,
303 ) -> Self {
304 Self {
305 api_key,
306 secret_key,
307 testnet,
308 is_isolated,
309 isolated_symbols,
310 side_effect,
311 }
312 }
313
314 pub fn cross_margin(api_key: String, secret_key: String) -> Self {
321 Self::new(
322 api_key,
323 secret_key,
324 false,
325 false,
326 Vec::new(),
327 MarginSideEffect::default(),
328 )
329 }
330
331 pub fn isolated(
339 api_key: String,
340 secret_key: String,
341 symbols: Vec<InstrumentNameExchange>,
342 ) -> Self {
343 Self::new(
344 api_key,
345 secret_key,
346 false,
347 true,
348 symbols,
349 MarginSideEffect::default(),
350 )
351 }
352
353 pub fn api_key(&self) -> &str {
355 &self.api_key
356 }
357}
358
359#[derive(Clone)]
466pub struct BinanceMargin {
467 config: Arc<BinanceMarginConfig>,
468 rest: Arc<RestApi>,
470 rest_config: Arc<ConfigurationRestApi>,
474 ws_config: ConfigurationWebsocketApi,
478 rate_limiter: Arc<RateLimitTracker>,
480 known_live: SharedKnownLiveOrders,
483}
484
485impl std::fmt::Debug for BinanceMargin {
486 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
487 f.debug_struct("BinanceMargin")
488 .field("testnet", &self.config.testnet)
489 .field("is_isolated", &self.config.is_isolated)
490 .field("side_effect", &self.config.side_effect)
491 .finish_non_exhaustive()
492 }
493}
494
495impl BinanceMargin {
496 fn remember_live<'a>(
498 &self,
499 orders: impl IntoIterator<Item = &'a Order<ExchangeId, InstrumentNameExchange, Open>>,
500 ) {
501 let mut known = self.known_live.lock();
502 for order in orders {
503 known.live(&order.key, order.quantity, order.kind, &order.state);
504 }
505 }
506
507 #[allow(clippy::expect_used)] fn build_rest_config(config: &BinanceMarginConfig) -> ConfigurationRestApi {
517 ConfigurationRestApi::builder()
518 .api_key(config.api_key.clone())
519 .api_secret(config.secret_key.clone())
520 .base_path(MARGIN_TRADING_REST_API_PROD_URL)
521 .build()
522 .expect("failed to build Binance margin REST configuration")
523 }
524
525 #[allow(clippy::expect_used)] fn build_ws_config(config: &BinanceMarginConfig) -> ConfigurationWebsocketApi {
536 ConfigurationWebsocketApi::builder()
537 .api_key(config.api_key.clone())
538 .api_secret(config.secret_key.clone())
539 .ws_url(SPOT_WS_API_PROD_URL)
543 .build()
544 .expect("failed to build Binance margin WebSocket configuration")
545 }
546
547 fn effective_isolated_set(
560 &self,
561 instruments: &[InstrumentNameExchange],
562 ) -> Vec<InstrumentNameExchange> {
563 if instruments.is_empty() {
564 return self.config.isolated_symbols.clone();
565 }
566 instruments
567 .iter()
568 .filter(|inst| {
569 let in_set = self.config.isolated_symbols.contains(*inst);
570 if !in_set {
571 warn!(
572 instrument = %inst.name(),
573 "BinanceMargin isolated: requested instrument not in configured isolated_symbols — skipping"
574 );
575 }
576 in_set
577 })
578 .cloned()
579 .collect()
580 }
581
582 async fn isolated_account_snapshot(
586 &self,
587 instruments: &[InstrumentNameExchange],
588 ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
589 use futures::{StreamExt as _, TryStreamExt as _};
590
591 let effective = self.effective_isolated_set(instruments);
592 if effective.is_empty() {
593 return Ok(AccountSnapshot::new(
596 ExchangeId::BinanceMargin,
597 Vec::new(),
598 Vec::new(),
599 ));
600 }
601
602 let mut balances_by_symbol = convert_isolated_margin_assets(
604 fetch_isolated_margin_account_info(
605 self.rest.clone(),
606 self.rate_limiter.clone(),
607 effective.clone(),
608 )
609 .await?,
610 );
611
612 let instrument_snapshots: Vec<_> =
615 futures::stream::iter(effective.into_iter().map(|instrument| {
616 fetch_margin_open_orders_for_instrument(
617 self.rest.clone(),
618 self.rate_limiter.clone(),
619 instrument,
620 true,
621 RequestKind::Query,
622 )
623 }))
624 .buffer_unordered(8)
625 .map(|result| {
626 let (inst, listing) = result?;
627 self.remember_live(&listing.orders);
628 let wrapped = listing
629 .orders
630 .into_iter()
631 .map(active_order_snapshot)
632 .collect();
633 let isolated = balances_by_symbol.remove(&inst);
634 if isolated.is_none() {
635 warn!(
639 instrument = %inst.name(),
640 "BinanceMargin isolated: no isolated balance entry returned for instrument — snapshot will carry isolated: None"
641 );
642 }
643 Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
644 inst,
645 wrapped,
646 listing.complete,
647 PositionReport::Unreported,
648 isolated,
649 ))
650 })
651 .try_collect()
652 .await?;
653
654 Ok(AccountSnapshot::new(
657 ExchangeId::BinanceMargin,
658 Vec::new(),
659 instrument_snapshots,
660 ))
661 }
662}
663
664impl ExecutionClient for BinanceMargin {
665 const EXCHANGE: ExchangeId = ExchangeId::BinanceMargin;
666
667 const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] =
670 &[InstrumentKindDiscriminant::Spot];
671 type Config = BinanceMarginConfig;
672 type AccountStream = BoxStream<'static, UnindexedAccountEvent>;
673
674 fn new(config: Self::Config) -> Self {
687 if config.testnet {
688 warn!(
689 "BinanceMarginConfig.testnet = true is ignored: Binance margin has no testnet; \
690 using production endpoints"
691 );
692 }
693 assert!(
694 !(config.is_isolated && config.isolated_symbols.is_empty()),
695 "BinanceMarginConfig: is_isolated = true requires a non-empty isolated_symbols"
696 );
697 let rest_config = Self::build_rest_config(&config);
702 let rest = Arc::new(MarginTradingRestApi::production(rest_config.clone()));
703 let rest_config = Arc::new(rest_config);
704 let ws_config = Self::build_ws_config(&config);
705 Self {
706 config: Arc::new(config),
707 rest,
708 rest_config,
709 ws_config,
710 rate_limiter: Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
711 known_live: KnownLiveOrders::shared(ExchangeId::BinanceMargin),
712 }
713 }
714
715 async fn open_order(
734 &self,
735 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
736 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
737 let instrument = request.key.instrument.clone();
738 let side = request.state.side;
739 let price = request.state.price;
740 let quantity = request.state.quantity;
741 let kind = request.state.kind;
742 let time_in_force = request.state.time_in_force;
743 let cid = request.key.cid.clone();
744
745 let order_key = OrderKey::new(
746 ExchangeId::BinanceMargin,
747 instrument.clone(),
748 request.key.strategy.clone(),
749 cid.clone(),
750 );
751
752 let inactive = |state: UnindexedOrderState| Order {
755 key: order_key.clone(),
756 side,
757 price,
758 quantity,
759 kind,
760 time_in_force,
761 state,
762 };
763
764 let params = match build_new_order_params(
765 instrument.name().to_string(),
766 side,
767 price,
768 quantity,
769 kind,
770 time_in_force,
771 cid.0.to_string(),
772 self.config.side_effect,
773 self.config.is_isolated,
774 ) {
775 Ok(params) => params,
776 Err(BuildOrderError::Unsupported) => {
777 return inactive(OrderState::inactive(OrderError::UnsupportedOrderType(
778 format!(
779 "BinanceMargin does not support OrderKind::{kind:?} with {time_in_force:?}"
780 ),
781 )));
782 }
783 Err(BuildOrderError::Trailing(refused)) => {
784 return inactive(OrderState::inactive(refused.into_order_error()));
785 }
786 Err(BuildOrderError::Build(msg)) => {
787 error!(%msg, "BinanceMargin failed to build new order params");
788 return inactive(OrderState::inactive(OrderError::Rejected(
789 ApiError::OrderRejected(msg),
790 )));
791 }
792 };
793
794 let response =
795 match rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Order, |rest| {
796 let params = params.clone();
797 Box::pin(async move { rest.margin_account_new_order(params).await })
798 })
799 .await
800 {
801 Ok(response) => response,
802 Err(e) => {
803 return inactive(OrderState::inactive(classify_rest_order_error(
804 &e,
805 &instrument,
806 )));
807 }
808 };
809
810 let data = match response.data().await {
811 Ok(data) => data,
812 Err(e) => {
813 return inactive(OrderState::inactive(OrderError::Rejected(
816 ApiError::OrderRejected(e.to_string()),
817 )));
818 }
819 };
820
821 let time_exchange = data
822 .transact_time
823 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
824 .unwrap_or_else(Utc::now);
825
826 let exchange_order_id = match data.order_id {
827 Some(id) => OrderId(format_smolstr!("{id}")),
828 None => {
829 error!("BinanceMargin open_order response missing orderId");
830 return inactive(OrderState::inactive(OrderError::Rejected(
831 ApiError::OrderRejected("open_order response missing orderId".into()),
832 )));
833 }
834 };
835
836 let state = placed_order_state(
839 ExchangeId::BinanceMargin,
840 &instrument,
841 quantity,
842 PlacementResponse {
843 status: data.status.as_deref(),
844 order_id: exchange_order_id,
845 time_exchange,
846 executed_qty: data.executed_qty.as_deref(),
847 cumulative_quote_qty: data.cummulative_quote_qty.as_deref(),
848 },
849 );
850 self.known_live
851 .lock()
852 .placed(&order_key, quantity, kind, &state);
853
854 Order {
855 key: order_key,
856 side,
857 price,
858 quantity,
859 kind,
860 time_in_force,
861 state,
862 }
863 }
864
865 async fn cancel_order(
875 &self,
876 request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
877 ) -> UnindexedOrderResponseCancel {
878 let instrument = request.key.instrument.clone();
879 let key = OrderKey {
880 exchange: request.key.exchange,
881 instrument: instrument.clone(),
882 strategy: request.key.strategy.clone(),
883 cid: request.key.cid.clone(),
884 };
885
886 let params = match build_cancel_order_params(
887 instrument.name().to_string(),
888 request.state.id.as_ref().and_then(VenueOrderId::assigned),
889 &request.key.cid,
890 self.config.is_isolated,
891 ) {
892 Ok(p) => p,
893 Err(e) => {
894 error!(%e, "BinanceMargin failed to build cancel order params");
895 return UnindexedOrderResponseCancel {
896 key,
897 state: Err(OrderError::Rejected(ApiError::OrderRejected(e))),
898 };
899 }
900 };
901
902 let response =
903 match rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Order, |rest| {
904 let params = params.clone();
905 Box::pin(async move { rest.margin_account_cancel_order(params).await })
906 })
907 .await
908 {
909 Ok(response) => response,
910 Err(e) => {
911 return UnindexedOrderResponseCancel {
912 key,
913 state: Err(classify_rest_order_error(&e, &instrument)),
914 };
915 }
916 };
917
918 let data = match response.data().await {
919 Ok(data) => data,
920 Err(e) => {
921 return UnindexedOrderResponseCancel {
924 key,
925 state: Err(OrderError::Rejected(ApiError::OrderRejected(e.to_string()))),
926 };
927 }
928 };
929
930 let time_exchange = Utc::now();
932
933 let exchange_order_id = match data.order_id {
934 Some(id) => OrderId(SmolStr::new(id)),
937 None => {
938 error!("BinanceMargin cancel response missing orderId");
939 return UnindexedOrderResponseCancel {
940 key,
941 state: Err(OrderError::Rejected(ApiError::OrderRejected(
942 "cancel response missing orderId".into(),
943 ))),
944 };
945 }
946 };
947
948 let filled_qty = binance_filled_qty(
949 ExchangeId::BinanceMargin,
950 &exchange_order_id,
951 data.executed_qty.as_deref(),
952 );
953
954 self.known_live.lock().ended(&key.cid);
955 UnindexedOrderResponseCancel {
956 key,
957 state: Ok(Cancelled::new(exchange_order_id, time_exchange, filled_qty)),
958 }
959 }
960
961 async fn account_snapshot(
976 &self,
977 assets: &[AssetNameExchange],
978 instruments: &[InstrumentNameExchange],
979 ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
980 if self.config.is_isolated {
981 return self.isolated_account_snapshot(instruments).await;
982 }
983 let response =
984 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
985 Box::pin(async move {
986 let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
987 rest.query_cross_margin_account_details(params).await
988 })
989 })
990 .await
991 .map_err(|e| classify_rest_query_error(&e, None))?;
992
993 let account = response.data().await.map_err(response_decode_error)?;
994
995 let balances =
996 filter_and_convert_margin_balances(account.user_assets.unwrap_or_default(), assets);
997
998 use futures::{StreamExt as _, TryStreamExt as _};
1003 let instrument_snapshots: Vec<_> =
1004 futures::stream::iter(instruments.iter().cloned().map(|instrument| {
1005 fetch_margin_open_orders_for_instrument(
1006 self.rest.clone(),
1007 self.rate_limiter.clone(),
1008 instrument,
1009 false,
1011 RequestKind::Query,
1012 )
1013 }))
1014 .buffer_unordered(8)
1015 .map(|result| {
1016 let (inst, listing) = result?;
1017 self.remember_live(&listing.orders);
1018 let wrapped = listing
1019 .orders
1020 .into_iter()
1021 .map(active_order_snapshot)
1022 .collect();
1023 Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
1024 inst,
1025 wrapped,
1026 listing.complete,
1027 PositionReport::Unreported,
1028 None,
1029 ))
1030 })
1031 .try_collect()
1032 .await?;
1033
1034 Ok(AccountSnapshot::new(
1035 ExchangeId::BinanceMargin,
1036 balances,
1037 instrument_snapshots,
1038 ))
1039 }
1040
1041 async fn fetch_balances(
1051 &self,
1052 assets: &[AssetNameExchange],
1053 ) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
1054 if self.config.is_isolated {
1055 return Ok(Vec::new());
1056 }
1057 let response =
1058 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1059 Box::pin(async move {
1060 let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
1061 rest.query_cross_margin_account_details(params).await
1062 })
1063 })
1064 .await
1065 .map_err(|e| classify_rest_query_error(&e, None))?;
1066
1067 let account = response.data().await.map_err(response_decode_error)?;
1068
1069 Ok(filter_and_convert_margin_balances(
1070 account.user_assets.unwrap_or_default(),
1071 assets,
1072 ))
1073 }
1074
1075 async fn fetch_open_orders(
1094 &self,
1095 instruments: &[InstrumentNameExchange],
1096 ) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
1097 use futures::{StreamExt as _, TryStreamExt as _};
1098
1099 if self.config.is_isolated {
1103 let effective = self.effective_isolated_set(instruments);
1104 let capacity = effective.len();
1105 return futures::stream::iter(effective.into_iter().map(|instrument| {
1106 fetch_margin_open_orders_for_instrument(
1107 self.rest.clone(),
1108 self.rate_limiter.clone(),
1109 instrument,
1110 true,
1111 RequestKind::Query,
1112 )
1113 }))
1114 .buffer_unordered(8)
1115 .try_fold(
1116 Vec::with_capacity(capacity),
1117 |mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>,
1118 (_, listing)| async move {
1119 acc.extend(listing.orders);
1120 Ok(acc)
1121 },
1122 )
1123 .await
1124 .inspect(|orders| self.remember_live(orders));
1125 }
1126
1127 if instruments.is_empty() {
1130 let orders =
1131 fetch_margin_all_open_orders(self.rest.clone(), self.rate_limiter.clone(), false)
1132 .await?;
1133 self.remember_live(&orders);
1134 return Ok(orders);
1135 }
1136 futures::stream::iter(instruments.iter().cloned().map(|instrument| {
1137 fetch_margin_open_orders_for_instrument(
1138 self.rest.clone(),
1139 self.rate_limiter.clone(),
1140 instrument,
1141 false,
1142 RequestKind::Query,
1143 )
1144 }))
1145 .buffer_unordered(8)
1146 .try_fold(
1147 Vec::with_capacity(instruments.len()),
1148 |mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, (_, listing)| async move {
1149 acc.extend(listing.orders);
1150 Ok(acc)
1151 },
1152 )
1153 .await
1154 .inspect(|orders| self.remember_live(orders))
1155 }
1156
1157 async fn fetch_trades(
1186 &self,
1187 start: DateTime<Utc>,
1188 end: DateTime<Utc>,
1189 instruments: &[InstrumentNameExchange],
1190 ) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
1191 use futures::StreamExt as _;
1192
1193 if start > end {
1194 return Ok(TradesRead::complete(Vec::new()));
1195 }
1196
1197 let effective: Vec<InstrumentNameExchange> = if self.config.is_isolated {
1200 self.effective_isolated_set(instruments)
1201 } else {
1202 instruments.to_vec()
1203 };
1204 if effective.is_empty() {
1205 debug!(
1208 is_isolated = self.config.is_isolated,
1209 "BinanceMargin fetch_trades: empty effective instrument set — returning empty result"
1210 );
1211 return Ok(TradesRead::complete(Vec::new()));
1212 }
1213 let span = MyTradesFrom::Span {
1215 start: start.timestamp_millis(),
1216 end: end.timestamp_millis(),
1217 };
1218 let is_isolated = self.config.is_isolated;
1219 let mut all_trades = Vec::new();
1220
1221 let mut stream = futures::stream::iter(effective.into_iter().map(|inst| {
1224 let rest = self.rest.clone();
1225 let rate_limiter = self.rate_limiter.clone();
1226 async move {
1227 let pages = paginate_margin_my_trades(
1228 &rest,
1229 &rate_limiter,
1230 &inst,
1231 span,
1232 is_isolated,
1233 RequestKind::Query,
1234 )
1235 .await?;
1236 Ok::<_, UnindexedClientError>((inst, pages))
1237 }
1238 }))
1239 .buffer_unordered(8);
1240 while let Some(result) = stream.next().await {
1241 let (instrument, trades_data) = result?;
1242 for t in trades_data {
1243 if let Some(trade) = convert_margin_trade(&t, &instrument)
1244 && (start..=end).contains(&trade.time_exchange)
1245 {
1246 all_trades.push(trade);
1247 }
1248 }
1249 }
1250
1251 Ok(TradesRead::complete(all_trades))
1252 }
1253
1254 async fn account_stream(
1388 &self,
1389 _assets: &[AssetNameExchange],
1392 instruments: &[InstrumentNameExchange],
1393 ) -> Result<Self::AccountStream, UnindexedClientError> {
1394 let (tx, rx) = mpsc::unbounded_channel::<UnindexedAccountEvent>();
1395 let dedup = new_dedup_cache();
1396 let rest = self.rest.clone();
1397 let rest_config = self.rest_config.clone();
1398 let ws_config = self.ws_config.clone();
1399 let rate_limiter = self.rate_limiter.clone();
1400 let known_live = self.known_live.clone();
1401
1402 let cm_handle = if self.config.is_isolated {
1403 let symbols = self.config.isolated_symbols.clone();
1407 debug_assert!(
1410 symbols.iter().all(|i| i.name().len() <= 23),
1411 "instrument name exceeds SmolStr inline capacity: {:?}",
1412 symbols.iter().find(|i| i.name().len() > 23)
1413 );
1414
1415 let assets = fetch_isolated_margin_account_info(
1421 rest.clone(),
1422 rate_limiter.clone(),
1423 symbols.clone(),
1424 )
1425 .await?;
1426 let base_quote = Arc::new(build_base_quote_map(&assets));
1427 for sym in &symbols {
1428 if !base_quote.contains_key(sym) {
1429 warn!(
1430 symbol = %sym.name(),
1431 "BinanceMargin isolated: account info returned no base/quote for configured \
1432 symbol — its live per-pair balance frames will be dropped (fills/orders \
1433 unaffected; balances available via account_snapshot)"
1434 );
1435 }
1436 }
1437
1438 let tokens = acquire_all_isolated_tokens(&rest_config, &rate_limiter, &symbols).await?;
1442 let live = isolated_connect_and_subscribe(
1443 &ws_config,
1444 &tx,
1445 &dedup,
1446 &known_live,
1447 &base_quote,
1448 &tokens,
1449 )
1450 .await
1451 .map_err(|e| {
1452 UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
1453 })?;
1454
1455 tokio::spawn(isolated_connection_manager(
1456 tx,
1457 dedup,
1458 ws_config,
1459 rest_config,
1460 rest,
1461 rate_limiter,
1462 known_live,
1463 symbols,
1464 base_quote,
1465 Some(live),
1466 ))
1467 } else {
1468 let instruments = instruments.to_vec();
1470 debug_assert!(
1474 instruments.iter().all(|i| i.name().len() <= 23),
1475 "instrument name exceeds SmolStr inline capacity: {:?}",
1476 instruments.iter().find(|i| i.name().len() > 23)
1477 );
1478
1479 let initial_token =
1485 acquire_user_listen_token(&rest_config, &rate_limiter, None).await?;
1486 let initial_ws = connect_margin_ws(&ws_config).await.map_err(|e| {
1487 UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
1488 })?;
1489
1490 let risk = RiskStreamTask::spawn(
1494 SapiRiskListenKeys {
1495 rest: rest.clone(),
1496 rate_limiter: rate_limiter.clone(),
1497 },
1498 RISK_STREAM_URL.to_owned(),
1499 tx.clone(),
1500 dedup.clone(),
1501 );
1502 let manager = margin_connection_manager(
1503 tx,
1504 dedup,
1505 ws_config,
1506 rest_config,
1507 rest,
1508 rate_limiter,
1509 known_live,
1510 instruments,
1511 Some((initial_ws, initial_token)),
1512 );
1513 tokio::spawn(async move {
1514 let _risk = risk;
1515 manager.await;
1516 })
1517 };
1518
1519 let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
1520 let guarded_stream = AbortOnDropStream::new(rx_stream, cm_handle);
1521 Ok(futures::StreamExt::boxed(guarded_stream))
1522 }
1523}
1524
1525impl OrderStatusClient for BinanceMargin {
1542 async fn fetch_ended_orders(
1543 &self,
1544 orders: &[UnindexedOrderKey],
1545 ) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError> {
1546 fetch_ended_by_key(orders, |key| {
1547 let recorded_kind = self.known_live.lock().kind_of(&key.cid);
1548 fetch_margin_order_lookup(
1549 self.rest.clone(),
1550 self.rate_limiter.clone(),
1551 key,
1552 self.config.is_isolated,
1553 RequestKind::Query,
1554 recorded_kind,
1555 )
1556 })
1557 .await
1558 }
1559}
1560
1561const HOURS_PER_YEAR: u32 = 24 * 365;
1567
1568impl ShortabilityClient for BinanceMargin {
1569 async fn fetch_shortability(
1592 &self,
1593 instrument: &InstrumentNameExchange,
1594 ) -> Result<Shortability, UnindexedClientError> {
1595 let pair = self.fetch_margin_pair(instrument).await?;
1596 let (borrowable, hourly_rate) = tokio::try_join!(
1597 self.fetch_asset_borrowable(&pair.base),
1598 self.fetch_next_hourly_interest_rate(&pair.base),
1599 )?;
1600
1601 let mut shortability = Shortability::new();
1602 let flags = [pair.is_margin_trade, pair.is_sell_allowed, borrowable];
1604 if flags.contains(&Some(false)) {
1605 shortability = shortability.with_shortable(false);
1606 } else if flags.iter().all(|flag| *flag == Some(true)) {
1607 shortability = shortability.with_shortable(true);
1608 }
1609 if let Some(hourly_rate) = hourly_rate {
1610 shortability = shortability.with_fee_rate(hourly_rate * Decimal::from(HOURS_PER_YEAR));
1611 }
1612 Ok(shortability)
1613 }
1614}
1615
1616impl BorrowCapacityClient for BinanceMargin {
1617 async fn fetch_borrow_capacity(
1634 &self,
1635 instrument: &InstrumentNameExchange,
1636 side: Side,
1637 ) -> Result<BorrowCapacity, UnindexedClientError> {
1638 let pair = self.fetch_margin_pair(instrument).await?;
1639 let asset = match side {
1640 Side::Buy => pair.quote,
1641 Side::Sell => pair.base,
1642 };
1643 let isolated_symbol = self
1644 .config
1645 .is_isolated
1646 .then(|| instrument.name().to_string());
1647 let response =
1648 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1649 let asset = asset.name().to_string();
1650 let isolated_symbol = isolated_symbol.clone();
1651 Box::pin(async move {
1652 let params = QueryMaxBorrowParams::builder(asset)
1653 .isolated_symbol(isolated_symbol)
1654 .build()?;
1655 rest.query_max_borrow(params).await
1656 })
1657 })
1658 .await
1659 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
1660 let max_borrow = response.data().await.map_err(response_decode_error)?;
1661
1662 let mut capacity = BorrowCapacity::new(asset);
1663 if let Some(amount) = optional_decimal("amount", max_borrow.amount.as_deref())? {
1664 capacity = capacity.with_borrowable_now(amount);
1665 }
1666 if let Some(limit) = optional_decimal("borrowLimit", max_borrow.borrow_limit.as_deref())? {
1667 capacity = capacity.with_account_limit(limit);
1668 }
1669 Ok(capacity)
1670 }
1671}
1672
1673#[derive(Debug)]
1675struct MarginPair {
1676 base: AssetNameExchange,
1677 quote: AssetNameExchange,
1678 is_margin_trade: Option<bool>,
1679 is_sell_allowed: Option<bool>,
1680}
1681
1682impl BinanceMargin {
1683 async fn fetch_margin_pair(
1687 &self,
1688 instrument: &InstrumentNameExchange,
1689 ) -> Result<MarginPair, UnindexedClientError> {
1690 let symbol = instrument.name().to_string();
1691 let listed = if self.config.is_isolated {
1692 let response =
1693 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1694 let symbol = symbol.clone();
1695 Box::pin(async move {
1696 let params = GetAllIsolatedMarginSymbolParams::builder()
1697 .symbol(symbol)
1698 .build()?;
1699 rest.get_all_isolated_margin_symbol(params).await
1700 })
1701 })
1702 .await
1703 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
1704 response
1705 .data()
1706 .await
1707 .map_err(response_decode_error)?
1708 .into_iter()
1709 .find(|pair| pair.symbol.as_deref() == Some(symbol.as_str()))
1710 .map(|pair| {
1711 (
1712 pair.base,
1713 pair.quote,
1714 pair.is_margin_trade,
1715 pair.is_sell_allowed,
1716 )
1717 })
1718 } else {
1719 let response =
1720 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1721 let symbol = symbol.clone();
1722 Box::pin(async move {
1723 let params = GetAllCrossMarginPairsParams::builder()
1724 .symbol(symbol)
1725 .build()?;
1726 rest.get_all_cross_margin_pairs(params).await
1727 })
1728 })
1729 .await
1730 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
1731 response
1732 .data()
1733 .await
1734 .map_err(response_decode_error)?
1735 .into_iter()
1736 .find(|pair| pair.symbol.as_deref() == Some(symbol.as_str()))
1737 .map(|pair| {
1738 (
1739 pair.base,
1740 pair.quote,
1741 pair.is_margin_trade,
1742 pair.is_sell_allowed,
1743 )
1744 })
1745 };
1746 let margin = if self.config.is_isolated {
1747 "isolated"
1748 } else {
1749 "cross"
1750 };
1751 let Some((base, quote, is_margin_trade, is_sell_allowed)) = listed else {
1752 return Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(
1753 instrument.clone(),
1754 format!("Binance lists no {margin} margin pair {symbol}"),
1755 )));
1756 };
1757 let (Some(base), Some(quote)) = (base, quote) else {
1758 return Err(UnindexedClientError::Internal(format!(
1759 "Binance lists {margin} margin pair {symbol} without its base or quote asset"
1760 )));
1761 };
1762 Ok(MarginPair {
1763 base: AssetNameExchange::new(base),
1764 quote: AssetNameExchange::new(quote),
1765 is_margin_trade,
1766 is_sell_allowed,
1767 })
1768 }
1769
1770 async fn fetch_asset_borrowable(
1773 &self,
1774 asset: &AssetNameExchange,
1775 ) -> Result<Option<bool>, UnindexedClientError> {
1776 let name = asset.name().to_string();
1777 let response =
1778 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1779 let name = name.clone();
1780 Box::pin(async move {
1781 let params = GetAllMarginAssetsParams::builder().asset(name).build()?;
1782 rest.get_all_margin_assets(params).await
1783 })
1784 })
1785 .await
1786 .map_err(|e| classify_rest_query_error(&e, None))?;
1787 Ok(response
1788 .data()
1789 .await
1790 .map_err(response_decode_error)?
1791 .into_iter()
1792 .find(|listed| listed.asset_name.as_deref() == Some(name.as_str()))
1793 .and_then(|listed| listed.is_borrowable))
1794 }
1795
1796 async fn fetch_next_hourly_interest_rate(
1800 &self,
1801 asset: &AssetNameExchange,
1802 ) -> Result<Option<Decimal>, UnindexedClientError> {
1803 let name = asset.name().to_string();
1804 let is_isolated = self.config.is_isolated;
1805 let response =
1806 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
1807 let name = name.clone();
1808 Box::pin(async move {
1809 let params = GetFutureHourlyInterestRateParams::builder(
1810 name,
1811 IsolatedFlag::from_flag(is_isolated),
1812 )
1813 .build()?;
1814 rest.get_future_hourly_interest_rate(params).await
1815 })
1816 })
1817 .await
1818 .map_err(|e| classify_rest_query_error(&e, None))?;
1819 let rate = response
1820 .data()
1821 .await
1822 .map_err(response_decode_error)?
1823 .into_iter()
1824 .find(|listed| listed.asset.as_deref() == Some(name.as_str()))
1825 .and_then(|listed| listed.next_hourly_interest_rate);
1826 optional_decimal("nextHourlyInterestRate", rate.as_deref())
1827 }
1828}
1829
1830fn optional_decimal(
1833 field: &str,
1834 value: Option<&str>,
1835) -> Result<Option<Decimal>, UnindexedClientError> {
1836 value
1837 .map(|value| {
1838 Decimal::from_str(value).map_err(|error| {
1839 UnindexedClientError::Internal(format!(
1840 "decoding Binance response: {field} {value:?}: {error}"
1841 ))
1842 })
1843 })
1844 .transpose()
1845}
1846
1847const TOKEN_RENEW_MARGIN_SECS: i64 = 300; const TOKEN_MIN_LIFETIME_SECS: u64 = 60;
1855
1856struct UserListenToken {
1858 token: String,
1859 expiration_time_ms: i64,
1860}
1861
1862fn build_listen_token_query(
1866 symbol: Option<&InstrumentNameExchange>,
1867) -> BTreeMap<String, serde_json::Value> {
1868 let mut query = BTreeMap::new();
1869 if let Some(sym) = symbol {
1870 query.insert(
1871 "isIsolated".to_string(),
1872 serde_json::Value::String("TRUE".to_string()),
1873 );
1874 query.insert(
1875 "symbol".to_string(),
1876 serde_json::Value::String(sym.name().to_string()),
1877 );
1878 }
1879 query
1880}
1881
1882#[derive(Deserialize)]
1884struct UserListenTokenResponse {
1885 #[serde(alias = "listenToken")]
1887 token: String,
1888 #[serde(rename = "expirationTime")]
1889 expiration_time: i64,
1890}
1891
1892async fn acquire_user_listen_token(
1907 rest_config: &Arc<ConfigurationRestApi>,
1908 rate_limiter: &RateLimitTracker,
1909 symbol: Option<&InstrumentNameExchange>,
1910) -> Result<UserListenToken, UnindexedClientError> {
1911 let query = build_listen_token_query(symbol);
1914 let response = rest_call_with_retry(rest_config, rate_limiter, RequestKind::Essential, |cfg| {
1915 let query = query.clone();
1916 Box::pin(async move {
1917 binance_sdk::common::utils::send_request::<UserListenTokenResponse>(
1918 &cfg,
1919 "/sapi/v1/userListenToken",
1920 reqwest::Method::POST,
1921 query,
1924 BTreeMap::new(),
1925 None,
1926 true, )
1928 .await
1929 })
1930 })
1931 .await
1932 .map_err(|e| classify_rest_query_error(&e, symbol))?;
1933
1934 let data = response.data().await.map_err(response_decode_error)?;
1935 Ok(UserListenToken {
1936 token: data.token,
1937 expiration_time_ms: data.expiration_time,
1938 })
1939}
1940
1941fn token_renew_after(expiration_time_ms: i64) -> Duration {
1944 let now_ms = Utc::now().timestamp_millis();
1945 let remaining_secs = expiration_time_ms
1949 .saturating_sub(now_ms)
1950 .saturating_div(1_000)
1951 .saturating_sub(TOKEN_RENEW_MARGIN_SECS);
1952 let secs = u64::try_from(remaining_secs).unwrap_or(0);
1953 if secs < TOKEN_MIN_LIFETIME_SECS {
1956 warn!(
1957 computed_secs = secs,
1958 floor_secs = TOKEN_MIN_LIFETIME_SECS,
1959 "BinanceMargin token renewal wait clamped to floor (near-expiry or odd expirationTime)"
1960 );
1961 }
1962 Duration::from_secs(secs.max(TOKEN_MIN_LIFETIME_SECS))
1963}
1964
1965async fn connect_margin_ws(
1972 ws_config: &ConfigurationWebsocketApi,
1973) -> anyhow::Result<Arc<WsApiBase>> {
1974 let ws = WsApiBase::new(ws_config.clone(), vec![]);
1975 tokio::time::timeout(
1978 Duration::from_secs(CONNECT_TIMEOUT_SECS),
1979 ws.clone().connect(),
1980 )
1981 .await
1982 .map_err(|_| {
1983 anyhow::anyhow!("BinanceMargin WS connect timed out after {CONNECT_TIMEOUT_SECS}s")
1984 })??;
1985 Ok(ws)
1986}
1987
1988async fn subscribe_listen_token(ws: &Arc<WsApiBase>, token: &str) -> anyhow::Result<()> {
1995 let mut payload = BTreeMap::new();
1998 payload.insert(
1999 "listenToken".to_string(),
2000 serde_json::Value::String(token.to_string()),
2001 );
2002 ws.send_message::<serde_json::Value>(
2003 "userDataStream.subscribe.listenToken",
2004 payload,
2005 WebsocketMessageSendOptions::new(),
2006 )
2007 .await
2008 .map_err(|e| anyhow::anyhow!("userDataStream.subscribe.listenToken failed: {e}"))?;
2009 Ok(())
2010}
2011
2012async fn subscribe_listen_token_capture(
2025 ws: &Arc<WsApiBase>,
2026 token: &str,
2027) -> anyhow::Result<Option<i64>> {
2028 #[derive(Deserialize)]
2031 struct SubscribeAck {
2032 #[serde(rename = "subscriptionId", default)]
2033 subscription_id: Option<i64>,
2034 }
2035
2036 let mut payload = BTreeMap::new();
2037 payload.insert(
2038 "listenToken".to_string(),
2039 serde_json::Value::String(token.to_string()),
2040 );
2041 let result = ws
2042 .send_message::<SubscribeAck>(
2043 "userDataStream.subscribe.listenToken",
2044 payload,
2045 WebsocketMessageSendOptions::new(),
2046 )
2047 .await
2048 .map_err(|e| anyhow::anyhow!("userDataStream.subscribe.listenToken failed: {e}"))?;
2049 let ack = match result {
2050 SendWebsocketMessageResult::Single(resp) => resp.data()?,
2051 SendWebsocketMessageResult::Multiple(responses) => responses
2053 .into_iter()
2054 .next()
2055 .ok_or_else(|| anyhow::anyhow!("empty subscribe ack"))?
2056 .data()?,
2057 };
2058 Ok(ack.subscription_id)
2059}
2060
2061fn cross_account_position_handler(
2065 position: OutboundAccountPosition,
2066 _subscription_id: Option<i64>,
2067 buf: &mut Vec<UnindexedAccountEvent>,
2068) {
2069 convert_margin_account_position(position, buf);
2070}
2071
2072fn cross_notice_instrument(_subscription_id: Option<i64>) -> Option<InstrumentNameExchange> {
2076 None
2077}
2078
2079fn isolated_notice_instrument(
2085 subscription_id: Option<i64>,
2086 sub_map: &Mutex<HashMap<i64, InstrumentNameExchange>>,
2087) -> Option<InstrumentNameExchange> {
2088 let Some(sub_id) = subscription_id else {
2089 warn!(
2090 "BinanceMargin isolated: MARGIN_LEVEL_STATUS_CHANGE without subscriptionId — \
2091 delivering the notice without an instrument"
2092 );
2093 return None;
2094 };
2095 let instrument = sub_map
2096 .lock()
2097 .unwrap_or_else(|p| p.into_inner())
2098 .get(&sub_id)
2099 .cloned();
2100 if instrument.is_none() {
2101 warn!(
2102 subscription_id = sub_id,
2103 "BinanceMargin isolated: MARGIN_LEVEL_STATUS_CHANGE for unmapped subscriptionId — \
2104 delivering the notice without an instrument"
2105 );
2106 }
2107 instrument
2108}
2109
2110#[cfg(test)]
2116fn convert_margin_user_data_events(frame: &str, buf: &mut Vec<UnindexedAccountEvent>) -> bool {
2117 convert_margin_user_data_events_with(
2118 frame,
2119 buf,
2120 &mut cross_account_position_handler,
2121 &mut cross_notice_instrument,
2122 )
2123}
2124
2125fn convert_margin_user_data_events_with(
2142 frame: &str,
2143 buf: &mut Vec<UnindexedAccountEvent>,
2144 handle_position: &mut impl FnMut(
2145 OutboundAccountPosition,
2146 Option<i64>,
2147 &mut Vec<UnindexedAccountEvent>,
2148 ),
2149 notice_instrument: &mut impl FnMut(Option<i64>) -> Option<InstrumentNameExchange>,
2150) -> bool {
2151 let (subscription_id, event_type, event_raw) = match parse_user_data_frame(frame) {
2152 UserDataFrame::Response => return false,
2154 UserDataFrame::Event {
2155 subscription_id,
2156 event_type,
2157 event,
2158 } => (subscription_id, event_type, event),
2159 UserDataFrame::Unrecognised => {
2160 static SEEN: AtomicU64 = AtomicU64::new(0);
2161 log_unrecognised_frame("BinanceMargin", &SEEN, frame);
2162 return false;
2163 }
2164 };
2165 match event_type {
2166 "executionReport" => {
2167 match serde_json::from_str::<ExecutionReport>(event_raw) {
2170 Ok(report) => {
2171 debug!(
2175 symbol = report.symbol.as_deref().unwrap_or("?"),
2176 order_id = ?report.order_id,
2177 client_order_id = report.client_order_id.as_deref().unwrap_or("?"),
2178 order_type = report.order_type.as_deref().unwrap_or("?"),
2179 execution_type = report.execution_type.as_deref().unwrap_or("?"),
2180 order_status = report.order_status.as_deref().unwrap_or("?"),
2181 "BinanceMargin executionReport"
2182 );
2183 convert_execution_report(&report, ExchangeId::BinanceMargin, buf);
2184 }
2185 Err(e) => {
2186 warn!(error = %e, frame = frame_excerpt(event_raw), "BinanceMargin: undeserializable executionReport, dropping")
2187 }
2188 }
2189 false
2190 }
2191 "outboundAccountPosition" => {
2192 match serde_json::from_str::<OutboundAccountPosition>(event_raw) {
2193 Ok(position) => handle_position(position, subscription_id, buf),
2195 Err(e) => {
2196 warn!(error = %e, frame = frame_excerpt(event_raw), "BinanceMargin: undeserializable outboundAccountPosition, dropping")
2197 }
2198 }
2199 false
2200 }
2201 "balanceUpdate" => {
2202 false
2206 }
2207 "USER_LIABILITY_CHANGE" | "userLiabilityChange" => {
2211 match serde_json::from_str::<UserLiabilityChange>(event_raw) {
2218 Ok(c) => log_liability_change(&c, "user data"),
2219 Err(e) => warn!(
2220 error = %e,
2221 frame = frame_excerpt(event_raw),
2222 "BinanceMargin: undeserializable USER_LIABILITY_CHANGE, dropping"
2223 ),
2224 }
2225 false
2226 }
2227 "MARGIN_LEVEL_STATUS_CHANGE" | "marginLevelStatusChange" => {
2228 match serde_json::from_str::<MarginLevelStatusChange>(event_raw) {
2231 Ok(change) => {
2232 log_margin_level_change(&change, "user data");
2233 buf.extend(margin_level_notice(
2234 change,
2235 notice_instrument(subscription_id),
2236 ));
2237 }
2238 Err(e) => warn!(
2239 error = %e,
2240 frame = frame_excerpt(event_raw),
2241 "BinanceMargin: undeserializable MARGIN_LEVEL_STATUS_CHANGE, dropping"
2242 ),
2243 }
2244 false
2245 }
2246 "eventStreamTerminated" => {
2247 warn!("BinanceMargin user data stream terminated by exchange, signalling reconnect");
2248 true
2249 }
2250 "listStatus" => {
2251 trace!("BinanceMargin ignoring listStatus user data event");
2254 false
2255 }
2256 other => {
2257 static SEEN: UnhandledEvents = UnhandledEvents::new();
2258 log_unhandled_event("BinanceMargin", &SEEN, other, event_raw);
2259 false
2260 }
2261 }
2262}
2263
2264fn convert_margin_account_position(
2269 position: OutboundAccountPosition,
2270 buf: &mut Vec<UnindexedAccountEvent>,
2271) {
2272 let time_exchange = position
2273 .u
2274 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
2275 .unwrap_or_else(Utc::now);
2276
2277 for b in position.b_uppercase.unwrap_or_default() {
2278 let asset = match b.a {
2279 Some(a) => AssetNameExchange::new(a),
2280 None => {
2281 warn!("BinanceMargin account position entry missing asset name");
2282 continue;
2283 }
2284 };
2285 let free = match b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2286 Some(v) => v,
2287 None => {
2288 warn!(%asset, "BinanceMargin account position missing/unparseable 'free' field");
2289 continue;
2290 }
2291 };
2292 let locked = match b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
2293 Some(v) => v,
2294 None => {
2295 warn!(%asset, "BinanceMargin account position missing/unparseable 'locked' field");
2296 continue;
2297 }
2298 };
2299 let update =
2300 AssetBalanceUpdate::new(asset, BalanceUpdate::new(free, locked), time_exchange);
2301 buf.push(UnindexedAccountEvent::new(
2302 ExchangeId::BinanceMargin,
2303 AccountEventKind::BalanceStreamUpdate(
2304 rustrade_integration::collection::snapshot::Snapshot::new(update),
2305 ),
2306 ));
2307 }
2308}
2309
2310fn route_isolated_account_position(
2321 position: OutboundAccountPosition,
2322 subscription_id: Option<i64>,
2323 sub_map: &Mutex<HashMap<i64, InstrumentNameExchange>>,
2324 base_quote: &HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>,
2325 buf: &mut Vec<UnindexedAccountEvent>,
2326) {
2327 let Some(sub_id) = subscription_id else {
2328 warn!(
2329 "BinanceMargin isolated: outboundAccountPosition without subscriptionId — dropping \
2330 (per-instrument balance routing requires it)"
2331 );
2332 return;
2333 };
2334 let instrument = {
2335 let map = sub_map.lock().unwrap_or_else(|p| p.into_inner());
2338 match map.get(&sub_id) {
2339 Some(inst) => inst.clone(),
2340 None => {
2341 warn!(
2342 subscription_id = sub_id,
2343 "BinanceMargin isolated: outboundAccountPosition for unmapped subscriptionId — dropping"
2344 );
2345 return;
2346 }
2347 }
2348 };
2349 let Some((base_asset, quote_asset)) = base_quote.get(&instrument) else {
2350 warn!(
2351 instrument = %instrument.name(),
2352 "BinanceMargin isolated: no base/quote known for instrument — dropping balance frame"
2353 );
2354 return;
2355 };
2356
2357 let time_exchange = position
2358 .u
2359 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
2360 .unwrap_or_else(Utc::now);
2361
2362 let mut base_update: Option<AssetBalanceUpdate<AssetNameExchange>> = None;
2363 let mut quote_update: Option<AssetBalanceUpdate<AssetNameExchange>> = None;
2364 for b in position.b_uppercase.unwrap_or_default() {
2365 let Some(asset) = b.a.as_deref() else {
2366 warn!(instrument = %instrument.name(), "BinanceMargin isolated account position entry missing asset name");
2367 continue;
2368 };
2369 let side = if asset == base_asset.name().as_str() {
2371 &mut base_update
2372 } else if asset == quote_asset.name().as_str() {
2373 &mut quote_update
2374 } else {
2375 continue;
2376 };
2377 let (Some(free), Some(locked)) = (
2378 b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()),
2379 b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()),
2380 ) else {
2381 warn!(%asset, instrument = %instrument.name(), "BinanceMargin isolated account position missing/unparseable free/locked");
2382 continue;
2383 };
2384 *side = Some(AssetBalanceUpdate::new(
2385 AssetNameExchange::new(asset),
2386 BalanceUpdate::new(free, locked),
2387 time_exchange,
2388 ));
2389 }
2390
2391 let (Some(base), Some(quote)) = (base_update, quote_update) else {
2395 warn!(
2396 instrument = %instrument.name(),
2397 "BinanceMargin isolated: outboundAccountPosition missing base or quote side — dropping (no partial InstrumentBalanceUpdate)"
2398 );
2399 return;
2400 };
2401 buf.push(UnindexedAccountEvent::new(
2402 ExchangeId::BinanceMargin,
2403 AccountEventKind::InstrumentBalanceUpdate(InstrumentBalanceUpdate::new(
2404 instrument, base, quote,
2405 )),
2406 ));
2407}
2408
2409fn register_user_data_listener(
2422 ws: &Arc<WsApiBase>,
2423 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
2424 dedup: SharedDedupCache,
2425 known: SharedKnownLiveOrders,
2426 heartbeat_flag: Arc<AtomicBool>,
2427 signal_tx: oneshot::Sender<()>,
2428 mut handle_position: impl FnMut(
2429 OutboundAccountPosition,
2430 Option<i64>,
2431 &mut Vec<UnindexedAccountEvent>,
2432 ) + Send
2433 + 'static,
2434 mut notice_instrument: impl FnMut(Option<i64>) -> Option<InstrumentNameExchange> + Send + 'static,
2435) -> Subscription {
2436 let mut signal_tx_opt = Some(signal_tx);
2437 let mut event_tx = Some(tx);
2438 let mut event_buf = Vec::with_capacity(32);
2442
2443 ws.common.events.subscribe(move |event| {
2447 let Some(ref sender) = event_tx else { return };
2448 match event {
2449 WebsocketEvent::Message(json_str) => {
2450 heartbeat_flag.store(true, Ordering::Release);
2451 let terminated = convert_margin_user_data_events_with(
2454 &json_str,
2455 &mut event_buf,
2456 &mut handle_position,
2457 &mut notice_instrument,
2458 );
2459 for ev in event_buf.drain(..) {
2460 if let Some(key) = dedup_key_from_event(&ev)
2461 && is_duplicate(&dedup, key)
2462 {
2463 trace!("BinanceMargin dedup: skipping duplicate event");
2464 continue;
2465 }
2466 let held = KnownLiveOrders::observes(&ev.kind).then(|| {
2470 let mut held = known.lock();
2471 held.observe(&ev);
2472 held
2473 });
2474 if sender.send(ev).is_err() {
2475 drop(held);
2476 warn!("BinanceMargin account_stream receiver dropped, suppressing sends");
2477 event_tx.take();
2478 if let Some(s) = signal_tx_opt.take() {
2479 let _ = s.send(());
2480 }
2481 return;
2482 }
2483 }
2484 if terminated {
2485 event_tx.take();
2486 if let Some(s) = signal_tx_opt.take() {
2487 let _ = s.send(());
2488 }
2489 }
2490 }
2491 WebsocketEvent::Ping | WebsocketEvent::Pong => {
2492 heartbeat_flag.store(true, Ordering::Release);
2493 }
2494 WebsocketEvent::Error(e) => {
2495 warn!(%e, "BinanceMargin WebSocket error, will attempt reconnect");
2496 event_tx.take();
2497 if let Some(s) = signal_tx_opt.take() {
2498 let _ = s.send(());
2499 }
2500 }
2501 WebsocketEvent::Close(code, reason) => {
2502 warn!(code, %reason, "BinanceMargin WebSocket closed");
2503 event_tx.take();
2504 if let Some(s) = signal_tx_opt.take() {
2505 let _ = s.send(());
2506 }
2507 }
2508 _ => {
2509 trace!("BinanceMargin ignoring unhandled WebsocketEvent variant");
2510 }
2511 }
2512 })
2513}
2514
2515async fn recover_margin_fills(
2537 rest: &Arc<RestApi>,
2538 rate_limiter: &Arc<RateLimitTracker>,
2539 unrecovered: &mut UnrecoveredFills,
2540 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
2541 dedup: &SharedDedupCache,
2542 known: &SharedKnownLiveOrders,
2543 is_isolated: bool,
2544) {
2545 use futures::StreamExt as _;
2546
2547 let due = unrecovered.due(tokio::time::Instant::now());
2548 let Some(oldest) = due.iter().map(|(_, gap)| gap.start_ms).min() else {
2549 return;
2550 };
2551 info!(
2552 gaps = due.len(),
2553 oldest = %gap_time(oldest),
2554 is_isolated,
2555 "BinanceMargin recovering fills missed while disconnected"
2556 );
2557
2558 let mut settled = vec![false; due.len()];
2561 let started = AtomicUsize::new(0);
2562 let recovery = async {
2563 let order_executions_deadline = tokio::time::Instant::now() + ORDER_EXECUTIONS_BUDGET;
2564 let mut recovered = 0u32;
2565 let mut duplicates = 0u32;
2566
2567 let mut stream =
2568 futures::stream::iter(due.iter().cloned().enumerate().map(|(index, (inst, gap))| {
2569 started.store(index + 1, Ordering::Relaxed);
2570 let rest = rest.clone();
2571 let rl = rate_limiter.clone();
2572 async move {
2573 let span = MyTradesFrom::Span {
2574 start: gap.start_ms,
2575 end: gap.end_ms,
2576 };
2577 let raw = match paginate_margin_my_trades(
2578 &rest,
2579 &rl,
2580 &inst,
2581 span,
2582 is_isolated,
2583 RequestKind::Essential,
2584 )
2585 .await
2586 {
2587 Ok(pages) => pages,
2588 Err(e) => return (index, Err(e)),
2589 };
2590 let totals = recovered_order_totals(
2594 ExchangeId::BinanceMargin,
2595 &inst,
2596 &raw,
2597 order_executions_deadline,
2598 |order_id| {
2599 paginate_margin_my_trades(
2600 &rest,
2601 &rl,
2602 &inst,
2603 MyTradesFrom::Order(order_id),
2604 is_isolated,
2605 RequestKind::Essential,
2606 )
2607 },
2608 )
2609 .await;
2610 let cids = recovered_order_cids(
2613 ExchangeId::BinanceMargin,
2614 &inst,
2615 &raw,
2616 order_executions_deadline,
2617 |order_id| fetch_margin_order_cid(&rest, &rl, &inst, order_id, is_isolated),
2618 )
2619 .await;
2620 let trades = raw
2621 .iter()
2622 .filter_map(|t| {
2623 let mut trade = convert_margin_trade(t, &inst)?;
2624 trade.order_filled_quantity =
2625 t.id.and_then(|id| totals.get(&id).copied());
2626 trade.client_order_id =
2627 t.order_id.and_then(|id| cids.get(&id).cloned());
2628 Some(trade)
2629 })
2630 .collect::<Vec<_>>();
2631 (index, Ok(trades))
2632 }
2633 }))
2634 .buffer_unordered(8);
2635 while let Some((index, result)) = stream.next().await {
2636 let (inst, gap) = &due[index];
2637 settled[index] = true;
2638 let trades = match result {
2639 Ok(trades) => trades,
2640 Err(e) => {
2641 gap_failed(
2642 unrecovered,
2643 tx,
2644 ExchangeId::BinanceMargin,
2645 inst,
2646 gap,
2647 FillRecoveryFailure::Request(e.to_string()),
2648 );
2649 continue;
2650 }
2651 };
2652 for trade in trades {
2653 let event = UnindexedAccountEvent::new(
2654 ExchangeId::BinanceMargin,
2655 AccountEventKind::Trade(trade),
2656 );
2657 if let Some(key) = dedup_key_from_event(&event)
2658 && is_duplicate(dedup, key)
2659 {
2660 duplicates += 1;
2661 continue;
2662 }
2663 let sent = {
2666 let mut known = known.lock();
2667 known.observe(&event);
2668 tx.send(event)
2669 };
2670 if sent.is_err() {
2671 debug!("BinanceMargin fill recovery: consumer dropped during recovery");
2672 return;
2673 }
2674 recovered += 1;
2675 }
2676 unrecovered.recovered(inst, gap);
2677 }
2678 info!(
2679 recovered,
2680 duplicates, is_isolated, "BinanceMargin fill recovery complete"
2681 );
2682 };
2683 if tokio::time::timeout(Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS), recovery)
2687 .await
2688 .is_err()
2689 {
2690 let started = started.load(Ordering::Relaxed);
2691 for ((inst, gap), _) in due[..started]
2692 .iter()
2693 .zip(&settled)
2694 .filter(|(_, settled)| !**settled)
2695 {
2696 gap_failed(
2697 unrecovered,
2698 tx,
2699 ExchangeId::BinanceMargin,
2700 inst,
2701 gap,
2702 FillRecoveryFailure::TimedOut {
2703 timeout_secs: FILL_RECOVERY_TIMEOUT_SECS,
2704 },
2705 );
2706 }
2707 }
2708}
2709
2710fn open_reconnect_recovery(
2715 unrecovered: &mut UnrecoveredFills,
2716 unchecked: &mut UncheckedOrders,
2717 known: &SharedKnownLiveOrders,
2718 instruments: &[InstrumentNameExchange],
2719 disconnected: DateTime<Utc>,
2720) {
2721 unrecovered.open(instruments, disconnected, Utc::now());
2722 let held = known.lock().instruments_among(instruments);
2723 unchecked.open(held);
2724}
2725
2726#[allow(
2733 clippy::cognitive_complexity,
2734 reason = "inherent reconnect-loop complexity (token + connect + subscribe + callback + recovery \
2735 + monitor + cleanup + backoff); mirrors spot's `connection_manager`, not worth splitting further"
2736)]
2737async fn margin_connection_manager(
2738 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
2739 dedup: SharedDedupCache,
2740 ws_config: ConfigurationWebsocketApi,
2741 rest_config: Arc<ConfigurationRestApi>,
2742 rest: Arc<RestApi>,
2743 rate_limiter: Arc<RateLimitTracker>,
2744 known: SharedKnownLiveOrders,
2745 instruments: Vec<InstrumentNameExchange>,
2746 initial: Option<(Arc<WsApiBase>, UserListenToken)>,
2747) {
2748 enum DisconnectReason {
2749 Signal,
2750 HeartbeatTimeout,
2751 TokenRefresh,
2752 ConsumerDropped,
2753 }
2754
2755 let mut backoff = ExponentialBackoff::new();
2756 let mut disconnect_time: Option<DateTime<Utc>> = None;
2757 let mut unrecovered = UnrecoveredFills::default();
2759 let mut unchecked = UncheckedOrders::default();
2761 let (mut current_ws, mut current_token) = match initial {
2762 Some((ws, token)) => (Some(ws), Some(token)),
2763 None => (None, None),
2764 };
2765
2766 loop {
2767 let token = match current_token.take() {
2769 Some(t) => t,
2770 None => match acquire_user_listen_token(&rest_config, &rate_limiter, None).await {
2771 Ok(t) => t,
2772 Err(e) => {
2773 error!(%e, "BinanceMargin userListenToken acquisition failed");
2774 if !backoff.wait().await {
2775 error!("BinanceMargin max reconnect attempts exhausted");
2776 emit_stream_terminated(
2777 &tx,
2778 ExchangeId::BinanceMargin,
2779 StreamTerminationReason::ReconnectBudgetExhausted {
2780 attempts: backoff.attempts(),
2781 last_error: e.to_string(),
2782 },
2783 );
2784 break;
2785 }
2786 continue;
2787 }
2788 },
2789 };
2790
2791 let ws = match current_ws.take() {
2793 Some(ws) => ws,
2794 None => match connect_margin_ws(&ws_config).await {
2795 Ok(ws) => ws,
2796 Err(e) => {
2797 error!(%e, "BinanceMargin WS connect failed");
2798 if !backoff.wait().await {
2799 error!("BinanceMargin max reconnect attempts exhausted");
2800 emit_stream_terminated(
2801 &tx,
2802 ExchangeId::BinanceMargin,
2803 StreamTerminationReason::ReconnectBudgetExhausted {
2804 attempts: backoff.attempts(),
2805 last_error: e.to_string(),
2806 },
2807 );
2808 break;
2809 }
2810 continue;
2811 }
2812 },
2813 };
2814
2815 let (signal_tx, signal_rx) = oneshot::channel::<()>();
2817 let heartbeat_flag = Arc::new(AtomicBool::new(true));
2819 let subscription = register_user_data_listener(
2822 &ws,
2823 tx.clone(),
2824 dedup.clone(),
2825 known.clone(),
2826 heartbeat_flag.clone(),
2827 signal_tx,
2828 cross_account_position_handler,
2829 cross_notice_instrument,
2830 );
2831
2832 if let Err(e) = subscribe_listen_token(&ws, &token.token).await {
2834 warn!(%e, "BinanceMargin user-data subscribe failed, cleaning up and retrying");
2835 subscription.unsubscribe();
2836 if let Err(de) = ws.disconnect().await {
2837 warn!(%de, "BinanceMargin failed to disconnect after subscribe failure");
2838 }
2839 if !backoff.wait().await {
2841 error!("BinanceMargin max reconnect attempts exhausted");
2842 emit_stream_terminated(
2843 &tx,
2844 ExchangeId::BinanceMargin,
2845 StreamTerminationReason::ReconnectBudgetExhausted {
2846 attempts: backoff.attempts(),
2847 last_error: e.to_string(),
2848 },
2849 );
2850 break;
2851 }
2852 continue;
2853 }
2854 info!("BinanceMargin account_stream connected and subscribed");
2855 backoff.reset();
2859
2860 let token_deadline =
2863 tokio::time::Instant::now() + token_renew_after(token.expiration_time_ms);
2864
2865 if let Some(dt) = disconnect_time.take() {
2870 open_reconnect_recovery(&mut unrecovered, &mut unchecked, &known, &instruments, dt);
2871 }
2872 recover_margin_fills(
2875 &rest,
2876 &rate_limiter,
2877 &mut unrecovered,
2878 &tx,
2879 &dedup,
2880 &known,
2881 false,
2882 )
2883 .await;
2884 recover_margin_ended_orders(
2885 &rest,
2886 &rate_limiter,
2887 &known,
2888 &mut unchecked,
2889 &unrecovered,
2890 &tx,
2891 false,
2892 )
2893 .await;
2894
2895 let reason = {
2897 let mut signal_rx = signal_rx;
2898 let retry_gaps = async {
2904 loop {
2905 let next_check = unchecked.next_due(&unrecovered);
2906 match unrecovered.next_due().into_iter().chain(next_check).min() {
2907 Some(due) => tokio::time::sleep_until(due).await,
2908 None => std::future::pending::<()>().await,
2909 }
2910 recover_margin_fills(
2911 &rest,
2912 &rate_limiter,
2913 &mut unrecovered,
2914 &tx,
2915 &dedup,
2916 &known,
2917 false,
2918 )
2919 .await;
2920 recover_margin_ended_orders(
2921 &rest,
2922 &rate_limiter,
2923 &known,
2924 &mut unchecked,
2925 &unrecovered,
2926 &tx,
2927 false,
2928 )
2929 .await;
2930 }
2931 };
2932 tokio::pin!(retry_gaps);
2933 loop {
2934 tokio::select! {
2935 biased;
2938 _ = tx.closed() => {
2939 debug!("BinanceMargin account_stream consumer dropped, terminating");
2940 break DisconnectReason::ConsumerDropped;
2941 }
2942 _ = &mut signal_rx => {
2943 warn!("BinanceMargin WS disconnected, will attempt reconnect");
2944 break DisconnectReason::Signal;
2945 }
2946 () = tokio::time::sleep_until(token_deadline) => {
2947 info!("BinanceMargin userListenToken nearing expiry, renewing");
2948 break DisconnectReason::TokenRefresh;
2949 }
2950 () = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
2951 if heartbeat_flag.swap(false, Ordering::AcqRel) {
2952 continue;
2956 }
2957 warn!("BinanceMargin heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
2958 break DisconnectReason::HeartbeatTimeout;
2959 }
2960 () = &mut retry_gaps => {}
2962 }
2963 }
2964 };
2965 let should_reconnect = !matches!(reason, DisconnectReason::ConsumerDropped);
2966
2967 if should_reconnect {
2968 disconnect_time = Some(match reason {
2969 DisconnectReason::HeartbeatTimeout => {
2970 Utc::now()
2971 - chrono::Duration::seconds(HEARTBEAT_TIMEOUT_SECS as i64)
2972 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
2973 }
2974 DisconnectReason::TokenRefresh => {
2975 Utc::now()
2980 - chrono::Duration::seconds(CONNECT_TIMEOUT_SECS as i64)
2981 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
2982 }
2983 _ => Utc::now() - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS),
2984 });
2985 }
2986
2987 subscription.unsubscribe();
2989 if let Err(e) = ws.disconnect().await {
2990 warn!(%e, "BinanceMargin failed to disconnect WebSocket");
2991 }
2992
2993 if !should_reconnect || tx.is_closed() {
2994 debug!("BinanceMargin connection manager exiting");
2996 break;
2997 }
2998
2999 if !matches!(reason, DisconnectReason::TokenRefresh) && !backoff.wait().await {
3002 error!("BinanceMargin max reconnect attempts exhausted, stream terminating");
3003 let last_error = match reason {
3006 DisconnectReason::HeartbeatTimeout => {
3007 format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
3008 }
3009 _ => "WebSocket disconnected (server close/error)".to_string(),
3010 };
3011 emit_stream_terminated(
3012 &tx,
3013 ExchangeId::BinanceMargin,
3014 StreamTerminationReason::ReconnectBudgetExhausted {
3015 attempts: backoff.attempts(),
3016 last_error,
3017 },
3018 );
3019 break;
3020 }
3021 }
3022}
3023
3024fn build_base_quote_map(
3036 assets: &[QueryIsolatedMarginAccountInfoResponseAssetsInner],
3037) -> HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)> {
3038 let mut map = HashMap::with_capacity(assets.len());
3039 for entry in assets {
3040 let (Some(symbol), Some(base), Some(quote)) = (
3041 entry.symbol.as_deref(),
3042 entry.base_asset.as_deref().and_then(|b| b.asset.as_deref()),
3043 entry
3044 .quote_asset
3045 .as_deref()
3046 .and_then(|q| q.asset.as_deref()),
3047 ) else {
3048 warn!(
3049 "BinanceMargin isolated: account-info entry missing symbol/base/quote asset — \
3050 skipping base/quote map entry"
3051 );
3052 continue;
3053 };
3054 map.insert(
3055 InstrumentNameExchange::new(symbol),
3056 (AssetNameExchange::new(base), AssetNameExchange::new(quote)),
3057 );
3058 }
3059 map
3060}
3061
3062async fn acquire_all_isolated_tokens(
3075 rest_config: &Arc<ConfigurationRestApi>,
3076 rate_limiter: &Arc<RateLimitTracker>,
3077 symbols: &[InstrumentNameExchange],
3078) -> Result<Vec<(InstrumentNameExchange, UserListenToken)>, UnindexedClientError> {
3079 use futures::{StreamExt as _, TryStreamExt as _};
3080
3081 futures::stream::iter(symbols.iter().cloned().map(|sym| {
3082 let rest_config = rest_config.clone();
3083 let rate_limiter = rate_limiter.clone();
3084 async move {
3085 let token = acquire_user_listen_token(&rest_config, &rate_limiter, Some(&sym)).await?;
3086 Ok::<_, UnindexedClientError>((sym, token))
3087 }
3088 }))
3089 .buffer_unordered(8)
3090 .try_collect()
3091 .await
3092}
3093
3094fn earliest_token_expiry_ms(tokens: &[(InstrumentNameExchange, UserListenToken)]) -> i64 {
3099 tokens
3100 .iter()
3101 .map(|(_, t)| t.expiration_time_ms)
3102 .min()
3103 .unwrap_or(i64::MAX)
3104}
3105
3106struct IsolatedLiveConn {
3109 ws: Arc<WsApiBase>,
3110 subscription: Subscription,
3111 signal_rx: oneshot::Receiver<()>,
3112 heartbeat_flag: Arc<AtomicBool>,
3113 earliest_expiry_ms: i64,
3116}
3117
3118async fn isolated_connect_and_subscribe(
3128 ws_config: &ConfigurationWebsocketApi,
3129 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3130 dedup: &SharedDedupCache,
3131 known: &SharedKnownLiveOrders,
3132 base_quote: &Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
3133 tokens: &[(InstrumentNameExchange, UserListenToken)],
3134) -> anyhow::Result<IsolatedLiveConn> {
3135 let ws = connect_margin_ws(ws_config).await?;
3136 let sub_map = Arc::new(Mutex::new(
3142 HashMap::<i64, InstrumentNameExchange>::with_capacity(tokens.len()),
3143 ));
3144 let (signal_tx, signal_rx) = oneshot::channel::<()>();
3145 let heartbeat_flag = Arc::new(AtomicBool::new(true));
3147
3148 let handle_position = {
3149 let sub_map = sub_map.clone();
3150 let base_quote = base_quote.clone();
3151 move |position, subscription_id, buf: &mut Vec<UnindexedAccountEvent>| {
3152 route_isolated_account_position(position, subscription_id, &sub_map, &base_quote, buf);
3153 }
3154 };
3155 let notice_instrument = {
3156 let sub_map = sub_map.clone();
3157 move |subscription_id| isolated_notice_instrument(subscription_id, &sub_map)
3158 };
3159 let subscription = register_user_data_listener(
3161 &ws,
3162 tx.clone(),
3163 dedup.clone(),
3164 known.clone(),
3165 heartbeat_flag.clone(),
3166 signal_tx,
3167 handle_position,
3168 notice_instrument,
3169 );
3170
3171 let earliest_expiry_ms = earliest_token_expiry_ms(tokens);
3172 for (sym, token) in tokens {
3173 match subscribe_listen_token_capture(&ws, &token.token).await {
3174 Ok(Some(id)) => {
3175 sub_map
3176 .lock()
3177 .unwrap_or_else(|p| p.into_inner())
3178 .insert(id, sym.clone());
3179 }
3180 Ok(None) => warn!(
3181 symbol = %sym.name(),
3182 "BinanceMargin isolated: subscribe ack carried no subscriptionId — live per-pair \
3183 balances unavailable for this symbol (fills/orders unaffected; balances via snapshot)"
3184 ),
3185 Err(e) => {
3186 warn!(%e, symbol = %sym.name(), "BinanceMargin isolated subscribe failed");
3189 subscription.unsubscribe();
3190 if let Err(de) = ws.disconnect().await {
3191 warn!(%de, "BinanceMargin isolated: disconnect after subscribe failure failed");
3192 }
3193 return Err(e);
3194 }
3195 }
3196 }
3197
3198 Ok(IsolatedLiveConn {
3199 ws,
3200 subscription,
3201 signal_rx,
3202 heartbeat_flag,
3203 earliest_expiry_ms,
3204 })
3205}
3206
3207#[allow(
3216 clippy::cognitive_complexity,
3217 reason = "inherent reconnect-loop complexity (tokens + connect + subscribe + monitor + cleanup \
3218 + backoff); mirrors the cross manager, not worth splitting further"
3219)]
3220async fn isolated_connection_manager(
3221 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
3222 dedup: SharedDedupCache,
3223 ws_config: ConfigurationWebsocketApi,
3224 rest_config: Arc<ConfigurationRestApi>,
3225 rest: Arc<RestApi>,
3226 rate_limiter: Arc<RateLimitTracker>,
3227 known: SharedKnownLiveOrders,
3228 symbols: Vec<InstrumentNameExchange>,
3229 base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
3230 initial: Option<IsolatedLiveConn>,
3231) {
3232 enum DisconnectReason {
3233 Signal,
3234 HeartbeatTimeout,
3235 TokenRefresh,
3236 ConsumerDropped,
3237 }
3238
3239 let mut backoff = ExponentialBackoff::new();
3240 let mut disconnect_time: Option<DateTime<Utc>> = None;
3241 let mut unrecovered = UnrecoveredFills::default();
3243 let mut unchecked = UncheckedOrders::default();
3245 let mut current = initial;
3246
3247 loop {
3248 let IsolatedLiveConn {
3250 ws,
3251 subscription,
3252 signal_rx,
3253 heartbeat_flag,
3254 earliest_expiry_ms,
3255 } = match current.take() {
3256 Some(live) => live,
3258 None => {
3259 let tokens = match acquire_all_isolated_tokens(
3262 &rest_config,
3263 &rate_limiter,
3264 &symbols,
3265 )
3266 .await
3267 {
3268 Ok(t) => t,
3269 Err(e) => {
3270 error!(%e, "BinanceMargin isolated userListenToken acquisition failed");
3271 if !backoff.wait().await {
3272 error!("BinanceMargin isolated max reconnect attempts exhausted");
3273 emit_stream_terminated(
3274 &tx,
3275 ExchangeId::BinanceMargin,
3276 StreamTerminationReason::ReconnectBudgetExhausted {
3277 attempts: backoff.attempts(),
3278 last_error: e.to_string(),
3279 },
3280 );
3281 break;
3282 }
3283 continue;
3284 }
3285 };
3286 match isolated_connect_and_subscribe(
3287 &ws_config,
3288 &tx,
3289 &dedup,
3290 &known,
3291 &base_quote,
3292 &tokens,
3293 )
3294 .await
3295 {
3296 Ok(live) => live,
3297 Err(e) => {
3298 error!(%e, "BinanceMargin isolated connect/subscribe failed");
3299 if !backoff.wait().await {
3300 error!("BinanceMargin isolated max reconnect attempts exhausted");
3301 emit_stream_terminated(
3302 &tx,
3303 ExchangeId::BinanceMargin,
3304 StreamTerminationReason::ReconnectBudgetExhausted {
3305 attempts: backoff.attempts(),
3306 last_error: e.to_string(),
3307 },
3308 );
3309 break;
3310 }
3311 continue;
3312 }
3313 }
3314 }
3315 };
3316
3317 info!(
3318 symbols = symbols.len(),
3319 "BinanceMargin isolated account_stream connected and subscribed"
3320 );
3321 backoff.reset();
3324
3325 let token_deadline = tokio::time::Instant::now() + token_renew_after(earliest_expiry_ms);
3328
3329 if let Some(dt) = disconnect_time.take() {
3332 open_reconnect_recovery(&mut unrecovered, &mut unchecked, &known, &symbols, dt);
3333 }
3334 recover_margin_fills(
3336 &rest,
3337 &rate_limiter,
3338 &mut unrecovered,
3339 &tx,
3340 &dedup,
3341 &known,
3342 true,
3343 )
3344 .await;
3345 recover_margin_ended_orders(
3346 &rest,
3347 &rate_limiter,
3348 &known,
3349 &mut unchecked,
3350 &unrecovered,
3351 &tx,
3352 true,
3353 )
3354 .await;
3355
3356 let reason = {
3359 let mut signal_rx = signal_rx;
3360 let retry_gaps = async {
3363 loop {
3364 let next_check = unchecked.next_due(&unrecovered);
3365 match unrecovered.next_due().into_iter().chain(next_check).min() {
3366 Some(due) => tokio::time::sleep_until(due).await,
3367 None => std::future::pending::<()>().await,
3368 }
3369 recover_margin_fills(
3370 &rest,
3371 &rate_limiter,
3372 &mut unrecovered,
3373 &tx,
3374 &dedup,
3375 &known,
3376 true,
3377 )
3378 .await;
3379 recover_margin_ended_orders(
3380 &rest,
3381 &rate_limiter,
3382 &known,
3383 &mut unchecked,
3384 &unrecovered,
3385 &tx,
3386 true,
3387 )
3388 .await;
3389 }
3390 };
3391 tokio::pin!(retry_gaps);
3392 loop {
3393 tokio::select! {
3394 biased;
3395 _ = tx.closed() => {
3396 debug!("BinanceMargin isolated consumer dropped, terminating");
3397 break DisconnectReason::ConsumerDropped;
3398 }
3399 _ = &mut signal_rx => {
3400 warn!("BinanceMargin isolated WS disconnected, will attempt reconnect");
3401 break DisconnectReason::Signal;
3402 }
3403 () = tokio::time::sleep_until(token_deadline) => {
3404 info!("BinanceMargin isolated userListenToken nearing expiry, renewing all");
3405 break DisconnectReason::TokenRefresh;
3406 }
3407 () = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
3408 if heartbeat_flag.swap(false, Ordering::AcqRel) {
3409 continue;
3410 }
3411 warn!("BinanceMargin isolated heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
3412 break DisconnectReason::HeartbeatTimeout;
3413 }
3414 () = &mut retry_gaps => {}
3416 }
3417 }
3418 };
3419 let should_reconnect = !matches!(reason, DisconnectReason::ConsumerDropped);
3420
3421 if should_reconnect {
3422 disconnect_time = Some(match reason {
3423 DisconnectReason::HeartbeatTimeout => {
3424 Utc::now()
3425 - chrono::Duration::seconds(HEARTBEAT_TIMEOUT_SECS as i64)
3426 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3427 }
3428 DisconnectReason::TokenRefresh => {
3429 Utc::now()
3433 - chrono::Duration::seconds(CONNECT_TIMEOUT_SECS as i64)
3434 - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
3435 }
3436 _ => Utc::now() - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS),
3437 });
3438 }
3439
3440 subscription.unsubscribe();
3442 if let Err(e) = ws.disconnect().await {
3443 warn!(%e, "BinanceMargin isolated failed to disconnect WebSocket");
3444 }
3445
3446 if !should_reconnect || tx.is_closed() {
3447 debug!("BinanceMargin isolated connection manager exiting");
3449 break;
3450 }
3451
3452 if !matches!(reason, DisconnectReason::TokenRefresh) && !backoff.wait().await {
3455 error!("BinanceMargin isolated max reconnect attempts exhausted, stream terminating");
3456 let last_error = match reason {
3459 DisconnectReason::HeartbeatTimeout => {
3460 format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
3461 }
3462 _ => "WebSocket disconnected (server close/error)".to_string(),
3463 };
3464 emit_stream_terminated(
3465 &tx,
3466 ExchangeId::BinanceMargin,
3467 StreamTerminationReason::ReconnectBudgetExhausted {
3468 attempts: backoff.attempts(),
3469 last_error,
3470 },
3471 );
3472 break;
3473 }
3474 }
3475}
3476
3477async fn fetch_margin_open_orders_for_instrument(
3484 rest: Arc<RestApi>,
3485 rate_limiter: Arc<RateLimitTracker>,
3486 instrument: InstrumentNameExchange,
3487 is_isolated: bool,
3488 kind: RequestKind,
3489) -> Result<(InstrumentNameExchange, OpenOrderListing), UnindexedClientError> {
3490 let symbol_str = instrument.name().to_string();
3492 let isolated = QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(is_isolated);
3493 let response = rest_call_with_retry(&rest, &rate_limiter, kind, |rest| {
3494 let sym = symbol_str.clone();
3495 let isolated = isolated.clone();
3496 Box::pin(async move {
3497 let params = QueryMarginAccountsOpenOrdersParams::builder()
3498 .symbol(sym)
3499 .is_isolated(isolated)
3500 .build()?;
3501 rest.query_margin_accounts_open_orders(params).await
3502 })
3503 })
3504 .await
3505 .map_err(|e| classify_rest_query_error(&e, Some(&instrument)))?;
3506
3507 let orders_data = response.data().await.map_err(response_decode_error)?;
3508
3509 let listing = convert_open_order_listing(&orders_data, ExchangeId::BinanceMargin, &instrument);
3510
3511 Ok((instrument, listing))
3512}
3513
3514async fn fetch_margin_all_open_orders(
3521 rest: Arc<RestApi>,
3522 rate_limiter: Arc<RateLimitTracker>,
3523 is_isolated: bool,
3524) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
3525 let isolated = QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(is_isolated);
3526 let response = rest_call_with_retry(&rest, &rate_limiter, RequestKind::Query, |rest| {
3527 let isolated = isolated.clone();
3528 Box::pin(async move {
3529 let params = QueryMarginAccountsOpenOrdersParams::builder()
3530 .is_isolated(isolated)
3531 .build()?;
3532 rest.query_margin_accounts_open_orders(params).await
3533 })
3534 })
3535 .await
3536 .map_err(|e| classify_rest_query_error(&e, None))?;
3537
3538 let orders_data = response.data().await.map_err(response_decode_error)?;
3539
3540 let orders = orders_data
3541 .into_iter()
3542 .filter_map(|o| convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin))
3543 .collect();
3544
3545 Ok(orders)
3546}
3547
3548async fn fetch_margin_order_lookup(
3555 rest: Arc<RestApi>,
3556 rate_limiter: Arc<RateLimitTracker>,
3557 key: UnindexedOrderKey,
3558 is_isolated: bool,
3559 kind: RequestKind,
3560 recorded_kind: Option<OrderKind>,
3561) -> Result<OrderLookup, UnindexedClientError> {
3562 let symbol = key.instrument.name().to_string();
3564 let cid = key.cid.0.to_string();
3565 let isolated = QueryMarginAccountsOrderIsIsolatedEnum::from_flag(is_isolated);
3566 let response = match rest_call_with_retry(&rest, &rate_limiter, kind, |rest| {
3567 let (symbol, cid, isolated) = (symbol.clone(), cid.clone(), isolated.clone());
3568 Box::pin(async move {
3569 let params = QueryMarginAccountsOrderParams::builder(symbol)
3570 .orig_client_order_id(cid)
3571 .is_isolated(isolated)
3572 .build()?;
3573 rest.query_margin_accounts_order(params).await
3574 })
3575 })
3576 .await
3577 {
3578 Ok(response) => response,
3579 Err(e) if is_unknown_order(&e) => {
3580 debug!(instrument = %key.instrument, cid = %key.cid, error = %e, "BinanceMargin does not know this order");
3581 return Ok(OrderLookup::Unknown);
3582 }
3583 Err(e) => return Err(classify_rest_query_error(&e, Some(&key.instrument))),
3584 };
3585
3586 let row = response.data().await.map_err(response_decode_error)?;
3587
3588 Ok(
3589 convert_ended_order(&row, ExchangeId::BinanceMargin, &key, recorded_kind)
3590 .map_or(OrderLookup::NotEnded, |order| {
3591 OrderLookup::Ended(Box::new(order))
3592 }),
3593 )
3594}
3595
3596async fn fetch_margin_order_cid(
3601 rest: &Arc<RestApi>,
3602 rate_limiter: &RateLimitTracker,
3603 instrument: &InstrumentNameExchange,
3604 order_id: i64,
3605 is_isolated: bool,
3606) -> Result<Option<ClientOrderId>, UnindexedClientError> {
3607 let symbol = instrument.name().to_string();
3608 let isolated = QueryMarginAccountsOrderIsIsolatedEnum::from_flag(is_isolated);
3609 let response = match rest_call_with_retry(rest, rate_limiter, RequestKind::Essential, |rest| {
3610 let (symbol, isolated) = (symbol.clone(), isolated.clone());
3611 Box::pin(async move {
3612 let params = QueryMarginAccountsOrderParams::builder(symbol)
3613 .order_id(order_id)
3614 .is_isolated(isolated)
3615 .build()?;
3616 rest.query_margin_accounts_order(params).await
3617 })
3618 })
3619 .await
3620 {
3621 Ok(response) => response,
3622 Err(e) if is_unknown_order(&e) => {
3623 debug!(%instrument, order_id, error = %e, "BinanceMargin does not know this order");
3624 return Ok(None);
3625 }
3626 Err(e) => return Err(classify_rest_query_error(&e, Some(instrument))),
3627 };
3628 let row = response.data().await.map_err(response_decode_error)?;
3629 Ok(row
3630 .client_order_id
3631 .filter(|cid| !cid.is_empty())
3632 .map(ClientOrderId::new))
3633}
3634
3635async fn listed_margin_open_cids(
3638 rest: Arc<RestApi>,
3639 rate_limiter: Arc<RateLimitTracker>,
3640 instruments: Vec<InstrumentNameExchange>,
3641 is_isolated: bool,
3642) -> Result<FnvHashSet<ClientOrderId>, UnindexedClientError> {
3643 let mut listed = FnvHashSet::default();
3644 for instrument in instruments {
3645 let (_, listing) = fetch_margin_open_orders_for_instrument(
3646 rest.clone(),
3647 rate_limiter.clone(),
3648 instrument,
3649 is_isolated,
3650 RequestKind::Essential,
3651 )
3652 .await?;
3653 listed.extend(listing.orders.into_iter().map(|order| order.key.cid));
3654 }
3655 Ok(listed)
3656}
3657
3658async fn recover_margin_ended_orders(
3661 rest: &Arc<RestApi>,
3662 rate_limiter: &Arc<RateLimitTracker>,
3663 known: &SharedKnownLiveOrders,
3664 unchecked: &mut UncheckedOrders,
3665 unrecovered: &UnrecoveredFills,
3666 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3667 is_isolated: bool,
3668) {
3669 recover_ended_orders(
3670 ExchangeId::BinanceMargin,
3671 known,
3672 unchecked,
3673 unrecovered,
3674 tx,
3675 OpenListing::PerInstrument,
3676 |instruments| {
3677 listed_margin_open_cids(rest.clone(), rate_limiter.clone(), instruments, is_isolated)
3678 },
3679 |key| {
3680 let recorded_kind = known.lock().kind_of(&key.cid);
3682 fetch_margin_order_lookup(
3683 rest.clone(),
3684 rate_limiter.clone(),
3685 key,
3686 is_isolated,
3687 RequestKind::Essential,
3688 recorded_kind,
3689 )
3690 },
3691 )
3692 .await;
3693}
3694
3695const MARGIN_MY_TRADES_WINDOW_MS: i64 = 23 * 60 * 60 * 1000;
3701
3702#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3705enum MarginTradesQuery {
3706 Window { start: i64, end: i64 },
3708 FromId(i64),
3710 Order(i64),
3712 OrderFromId { order_id: i64, from_id: i64 },
3714}
3715
3716impl MarginTradesQuery {
3717 fn first(from: MyTradesFrom, now_ms: i64) -> Option<Self> {
3721 match from {
3722 MyTradesFrom::Span { start, .. } => {
3723 (start <= now_ms).then(|| Self::window(start, now_ms))
3724 }
3725 MyTradesFrom::Order(order_id) => Some(Self::Order(order_id)),
3726 }
3727 }
3728
3729 fn window(start: i64, now_ms: i64) -> Self {
3731 Self::Window {
3732 start,
3733 end: start.saturating_add(MARGIN_MY_TRADES_WINDOW_MS).min(now_ms),
3734 }
3735 }
3736
3737 fn next(self, page_len: usize, last_id: Option<i64>, now_ms: i64) -> Option<Self> {
3740 let full = page_len >= BINANCE_MAX_TRADES;
3741 let after_last = || last_id?.checked_add(1);
3742 match self {
3743 Self::Window { end, .. } if page_len == 0 => {
3748 (end < now_ms).then(|| Self::window(end, now_ms))
3749 }
3750 Self::Window { end, .. } if end >= now_ms && !full => None,
3752 Self::Window { .. } => after_last().map(Self::FromId),
3755 Self::FromId(_) if full => after_last().map(Self::FromId),
3756 Self::Order(order_id) | Self::OrderFromId { order_id, .. } if full => {
3757 after_last().map(|from_id| Self::OrderFromId { order_id, from_id })
3758 }
3759 Self::FromId(_) | Self::Order(_) | Self::OrderFromId { .. } => None,
3760 }
3761 }
3762}
3763
3764async fn paginate_margin_my_trades(
3786 rest: &Arc<RestApi>,
3787 rate_limiter: &Arc<RateLimitTracker>,
3788 instrument: &InstrumentNameExchange,
3789 from: MyTradesFrom,
3790 is_isolated: bool,
3791 kind: RequestKind,
3792) -> Result<Vec<QueryMarginAccountsTradeListResponseInner>, UnindexedClientError> {
3793 let symbol_str = instrument.name().to_string();
3795 let isolated = QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(is_isolated);
3796 #[allow(clippy::cast_possible_wrap)]
3801 let limit = BINANCE_MAX_TRADES as i64;
3802 let now_ms = match from {
3805 MyTradesFrom::Span { end, .. } => end.min(Utc::now().timestamp_millis()),
3806 MyTradesFrom::Order(_) => Utc::now().timestamp_millis(),
3807 };
3808 let mut all_pages = Vec::new();
3809 let mut requests = 0u32;
3810 let mut next = MarginTradesQuery::first(from, now_ms);
3811 while let Some(query) = next {
3812 requests += 1;
3813 let response = rest_call_with_retry(rest, rate_limiter, kind, |rest| {
3814 let sym = symbol_str.clone();
3815 let isolated = isolated.clone();
3816 Box::pin(async move {
3817 let builder = QueryMarginAccountsTradeListParams::builder(sym)
3818 .is_isolated(isolated)
3819 .limit(limit);
3820 let params = match query {
3821 MarginTradesQuery::Window { start, end } => {
3822 builder.start_time(start).end_time(end)
3823 }
3824 MarginTradesQuery::FromId(id) => builder.from_id(id),
3825 MarginTradesQuery::Order(order_id) => builder.order_id(order_id),
3826 MarginTradesQuery::OrderFromId { order_id, from_id } => {
3827 builder.order_id(order_id).from_id(from_id)
3828 }
3829 }
3830 .build()?;
3831 rest.query_margin_accounts_trade_list(params).await
3832 })
3833 })
3834 .await
3835 .map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
3836
3837 let mut page = response.data().await.map_err(response_decode_error)?;
3838
3839 let page_len = page.len();
3840 let last_id = page.last().and_then(|t| t.id);
3841 let past_end = match from {
3843 MyTradesFrom::Span { end, .. } => drop_after(&mut page, end),
3844 MyTradesFrom::Order(_) => false,
3845 };
3846 all_pages.extend(page);
3847 if past_end {
3848 break;
3849 }
3850
3851 if page_len > 0 && last_id.is_none() {
3852 warn!(%instrument, "BinanceMargin paginate_my_trades: trade missing ID, cannot page past it");
3853 break;
3854 }
3855 next = query.next(page_len, last_id, now_ms);
3856 }
3857 debug!(
3858 %instrument,
3859 requests,
3860 trades = all_pages.len(),
3861 "BinanceMargin paginate_my_trades: walk complete"
3862 );
3863 Ok(all_pages)
3864}
3865
3866fn convert_margin_balance_entry(
3876 b: QueryCrossMarginAccountDetailsResponseUserAssetsInner,
3877 now: DateTime<Utc>,
3878) -> Option<AssetBalance<AssetNameExchange>> {
3879 let asset_name = AssetNameExchange::new(b.asset.as_deref()?);
3880 let free = match b.free.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
3881 Some(v) => v,
3882 None => {
3883 warn!(%asset_name, "BinanceMargin balance missing/unparseable 'free' field");
3884 return None;
3885 }
3886 };
3887 let locked = match b.locked.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
3888 Some(v) => v,
3889 None => {
3890 warn!(%asset_name, "BinanceMargin balance missing/unparseable 'locked' field");
3891 return None;
3892 }
3893 };
3894 let borrowed = b
3899 .borrowed
3900 .as_deref()
3901 .and_then(|s| Decimal::from_str(s).ok())
3902 .unwrap_or(Decimal::ZERO);
3903 let interest = b
3904 .interest
3905 .as_deref()
3906 .and_then(|s| Decimal::from_str(s).ok())
3907 .unwrap_or(Decimal::ZERO);
3908 Some(AssetBalance::new(
3909 asset_name,
3910 Balance::new_margin(free + locked, free, borrowed, interest),
3911 now,
3912 ))
3913}
3914
3915fn filter_and_convert_margin_balances(
3918 user_assets: Vec<QueryCrossMarginAccountDetailsResponseUserAssetsInner>,
3919 assets: &[AssetNameExchange],
3920) -> Vec<AssetBalance<AssetNameExchange>> {
3921 let now = Utc::now();
3922 if assets.is_empty() {
3924 return user_assets
3925 .into_iter()
3926 .filter_map(|b| convert_margin_balance_entry(b, now))
3927 .collect();
3928 }
3929 if assets.len() <= 16 {
3931 return user_assets
3932 .into_iter()
3933 .filter_map(|b| {
3934 let asset_name_str = b.asset.as_deref()?;
3935 if !assets.iter().any(|a| a.name().as_str() == asset_name_str) {
3936 return None;
3937 }
3938 convert_margin_balance_entry(b, now)
3939 })
3940 .collect();
3941 }
3942 use std::collections::HashSet;
3943 let asset_set: HashSet<&str> = assets.iter().map(|a| a.name().as_str()).collect();
3944 user_assets
3945 .into_iter()
3946 .filter_map(|b| {
3947 let asset_name_str = b.asset.as_deref()?;
3948 if !asset_set.contains(asset_name_str) {
3949 return None;
3950 }
3951 convert_margin_balance_entry(b, now)
3952 })
3953 .collect()
3954}
3955
3956fn active_order_snapshot(
3963 o: Order<ExchangeId, InstrumentNameExchange, Open>,
3964) -> Order<ExchangeId, InstrumentNameExchange, OrderState<AssetNameExchange, InstrumentNameExchange>>
3965{
3966 Order {
3967 key: o.key,
3968 side: o.side,
3969 price: o.price,
3970 quantity: o.quantity,
3971 kind: o.kind,
3972 time_in_force: o.time_in_force,
3973 state: OrderState::active(o.state),
3974 }
3975}
3976
3977fn convert_isolated_asset_balance(
3987 asset: Option<&str>,
3988 free: Option<&str>,
3989 locked: Option<&str>,
3990 borrowed: Option<&str>,
3991 interest: Option<&str>,
3992 now: DateTime<Utc>,
3993) -> Option<AssetBalance<AssetNameExchange>> {
3994 let asset_name = AssetNameExchange::new(asset?);
3995 let free = match free.and_then(|s| Decimal::from_str(s).ok()) {
3996 Some(v) => v,
3997 None => {
3998 warn!(%asset_name, "BinanceMargin isolated balance missing/unparseable 'free' field");
3999 return None;
4000 }
4001 };
4002 let locked = match locked.and_then(|s| Decimal::from_str(s).ok()) {
4003 Some(v) => v,
4004 None => {
4005 warn!(%asset_name, "BinanceMargin isolated balance missing/unparseable 'locked' field");
4006 return None;
4007 }
4008 };
4009 let borrowed = borrowed
4010 .and_then(|s| Decimal::from_str(s).ok())
4011 .unwrap_or(Decimal::ZERO);
4012 let interest = interest
4013 .and_then(|s| Decimal::from_str(s).ok())
4014 .unwrap_or(Decimal::ZERO);
4015 Some(AssetBalance::new(
4016 asset_name,
4017 Balance::new_margin(free + locked, free, borrowed, interest),
4018 now,
4019 ))
4020}
4021
4022fn convert_isolated_margin_assets(
4030 assets: Vec<QueryIsolatedMarginAccountInfoResponseAssetsInner>,
4031) -> HashMap<InstrumentNameExchange, IsolatedInstrumentState<AssetNameExchange>> {
4032 let now = Utc::now();
4035 let mut map = HashMap::with_capacity(assets.len());
4036 for entry in assets {
4037 let Some(symbol) = entry.symbol.as_deref() else {
4038 warn!("BinanceMargin isolated asset entry missing 'symbol' — skipping");
4039 continue;
4040 };
4041 let instrument = InstrumentNameExchange::new(symbol);
4042
4043 let Some(base_raw) = entry.base_asset.as_deref() else {
4044 warn!(%instrument, "BinanceMargin isolated entry missing baseAsset — skipping");
4045 continue;
4046 };
4047 let Some(quote_raw) = entry.quote_asset.as_deref() else {
4048 warn!(%instrument, "BinanceMargin isolated entry missing quoteAsset — skipping");
4049 continue;
4050 };
4051
4052 let Some(base) = convert_isolated_asset_balance(
4053 base_raw.asset.as_deref(),
4054 base_raw.free.as_deref(),
4055 base_raw.locked.as_deref(),
4056 base_raw.borrowed.as_deref(),
4057 base_raw.interest.as_deref(),
4058 now,
4059 ) else {
4060 warn!(%instrument, "BinanceMargin isolated baseAsset missing required field — skipping");
4061 continue;
4062 };
4063 let Some(quote) = convert_isolated_asset_balance(
4064 quote_raw.asset.as_deref(),
4065 quote_raw.free.as_deref(),
4066 quote_raw.locked.as_deref(),
4067 quote_raw.borrowed.as_deref(),
4068 quote_raw.interest.as_deref(),
4069 now,
4070 ) else {
4071 warn!(%instrument, "BinanceMargin isolated quoteAsset missing required field — skipping");
4072 continue;
4073 };
4074
4075 let risk = IsolatedMarginRisk {
4076 margin_level: entry
4077 .margin_level
4078 .as_deref()
4079 .and_then(|s| Decimal::from_str(s).ok()),
4080 margin_ratio: entry
4081 .margin_ratio
4082 .as_deref()
4083 .and_then(|s| Decimal::from_str(s).ok()),
4084 liquidation_price: entry
4085 .liquidate_price
4086 .as_deref()
4087 .and_then(|s| Decimal::from_str(s).ok()),
4088 };
4089
4090 map.insert(instrument, IsolatedInstrumentState { base, quote, risk });
4091 }
4092 map
4093}
4094
4095fn chunk_symbols(symbols: &[InstrumentNameExchange]) -> Vec<String> {
4098 symbols
4099 .chunks(5)
4100 .map(|chunk| {
4101 chunk
4102 .iter()
4103 .map(|s| s.name().as_str())
4104 .collect::<Vec<_>>()
4105 .join(",")
4106 })
4107 .collect()
4108}
4109
4110async fn fetch_isolated_margin_account_info(
4116 rest: Arc<RestApi>,
4117 rate_limiter: Arc<RateLimitTracker>,
4118 symbols: Vec<InstrumentNameExchange>,
4119) -> Result<Vec<QueryIsolatedMarginAccountInfoResponseAssetsInner>, UnindexedClientError> {
4120 use futures::{StreamExt as _, TryStreamExt as _};
4121
4122 let chunks = chunk_symbols(&symbols);
4123
4124 futures::stream::iter(chunks.into_iter().map(|symbols_param| {
4128 let rest = rest.clone();
4129 let rate_limiter = rate_limiter.clone();
4130 async move {
4131 let response = rest_call_with_retry(&rest, &rate_limiter, RequestKind::Query, |rest| {
4132 let symbols_param = symbols_param.clone();
4133 Box::pin(async move {
4134 let params = QueryIsolatedMarginAccountInfoParams::builder()
4135 .symbols(symbols_param)
4136 .build()?;
4137 rest.query_isolated_margin_account_info(params).await
4138 })
4139 })
4140 .await
4141 .map_err(|e| classify_rest_query_error(&e, None))?;
4142
4143 let info = response.data().await.map_err(response_decode_error)?;
4144 Ok::<_, UnindexedClientError>(info.assets.unwrap_or_default())
4145 }
4146 }))
4147 .buffer_unordered(8)
4148 .try_fold(
4149 Vec::with_capacity(symbols.len()),
4150 |mut acc, assets| async move {
4151 acc.extend(assets);
4152 Ok(acc)
4153 },
4154 )
4155 .await
4156}
4157
4158fn convert_margin_trade(
4165 t: &QueryMarginAccountsTradeListResponseInner,
4166 instrument: &InstrumentNameExchange,
4167) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
4168 let trade_id_raw = match t.id {
4169 Some(id) => id,
4170 None => {
4171 warn!(%instrument, "BinanceMargin trade missing id");
4172 return None;
4173 }
4174 };
4175 let trade_id = TradeId(format_smolstr!("{trade_id_raw}"));
4176 let order_id = match t.order_id {
4177 Some(id) => OrderId(format_smolstr!("{id}")),
4178 None => {
4179 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing orderId");
4180 return None;
4181 }
4182 };
4183 let side = match t.is_buyer {
4184 Some(true) => Side::Buy,
4185 Some(false) => Side::Sell,
4186 None => {
4187 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing isBuyer");
4188 return None;
4189 }
4190 };
4191 let price = match t.price.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
4192 Some(v) => v,
4193 None => {
4194 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable price");
4195 return None;
4196 }
4197 };
4198 let quantity = match t.qty.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
4199 Some(v) => v,
4200 None => {
4201 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable qty");
4202 return None;
4203 }
4204 };
4205 let commission = t
4206 .commission
4207 .as_deref()
4208 .and_then(|s| Decimal::from_str(s).ok())
4209 .unwrap_or(Decimal::ZERO);
4210 let time_exchange = match t.time.and_then(|ms| Utc.timestamp_millis_opt(ms).single()) {
4211 Some(ts) => ts,
4212 None => {
4213 warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable time, using now");
4214 Utc::now()
4215 }
4216 };
4217
4218 let fee_asset = t
4222 .commission_asset
4223 .as_deref()
4224 .map(AssetNameExchange::from)
4225 .unwrap_or_else(|| AssetNameExchange::from("UNKNOWN"));
4226
4227 Some(Trade::new(
4228 trade_id,
4229 order_id,
4230 instrument.clone(),
4231 StrategyId::unknown(), time_exchange,
4233 side,
4234 price,
4235 quantity,
4236 None,
4238 AssetFees::new(fee_asset, commission, None),
4239 ))
4240}
4241
4242#[derive(Debug)]
4248enum BuildOrderError {
4249 Unsupported,
4251 Build(String),
4253 Trailing(TrailingOrderError),
4255}
4256
4257trait IsolatedFlag {
4268 fn from_flag(is_isolated: bool) -> Self;
4270}
4271
4272macro_rules! impl_isolated_flag {
4277 ($($t:ty),* $(,)?) => {
4278 $(
4279 impl IsolatedFlag for $t {
4280 fn from_flag(is_isolated: bool) -> Self {
4281 if is_isolated { Self::True } else { Self::False }
4282 }
4283 }
4284 )*
4285 };
4286}
4287
4288impl_isolated_flag!(
4289 MarginAccountNewOrderIsIsolatedEnum,
4290 MarginAccountCancelOrderIsIsolatedEnum,
4291 QueryMarginAccountsOpenOrdersIsIsolatedEnum,
4292 QueryMarginAccountsOrderIsIsolatedEnum,
4293 QueryMarginAccountsTradeListIsIsolatedEnum,
4294 GetFutureHourlyInterestRateIsIsolatedEnum,
4295);
4296
4297fn build_cancel_order_params(
4305 symbol: String,
4306 id: Option<&OrderId>,
4307 cid: &ClientOrderId,
4308 is_isolated: bool,
4309) -> Result<MarginAccountCancelOrderParams, String> {
4310 let mut builder = MarginAccountCancelOrderParams::builder(symbol).is_isolated(
4311 MarginAccountCancelOrderIsIsolatedEnum::from_flag(is_isolated),
4312 );
4313
4314 match id {
4315 Some(order_id) => match order_id.0.parse::<i64>() {
4316 Ok(id) => builder = builder.order_id(id),
4317 Err(_) => {
4318 error!(
4320 order_id = %order_id.0,
4321 "BinanceMargin cancel: exchange orderId not parseable as i64, falling back to clientOrderId"
4322 );
4323 builder = builder.orig_client_order_id(cid.0.to_string());
4324 }
4325 },
4326 None => builder = builder.orig_client_order_id(cid.0.to_string()),
4327 }
4328
4329 builder.build().map_err(|e| e.to_string())
4330}
4331
4332#[allow(clippy::too_many_arguments)] fn build_new_order_params(
4341 symbol: String,
4342 side: Side,
4343 price: Option<Decimal>,
4344 quantity: Decimal,
4345 kind: OrderKind,
4346 time_in_force: TimeInForce,
4347 new_client_order_id: String,
4348 side_effect: MarginSideEffect,
4349 is_isolated: bool,
4350) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
4351 let binance_side = match side {
4352 Side::Buy => MarginAccountNewOrderSideEnum::Buy,
4353 Side::Sell => MarginAccountNewOrderSideEnum::Sell,
4354 };
4355
4356 let (binance_type, binance_tif) =
4357 convert_order_kind_tif_margin(kind, time_in_force).ok_or(BuildOrderError::Unsupported)?;
4358
4359 let mut builder = MarginAccountNewOrderParams::builder(symbol, binance_side, binance_type)
4361 .quantity(quantity)
4362 .is_isolated(MarginAccountNewOrderIsIsolatedEnum::from_flag(is_isolated))
4363 .side_effect_type(side_effect.as_sdk_enum())
4364 .new_client_order_id(new_client_order_id)
4365 .new_order_resp_type(MarginAccountNewOrderNewOrderRespTypeEnum::Full);
4366
4367 if side_effect == MarginSideEffect::AutoBorrowRepay {
4370 builder = builder.auto_repay_at_cancel(true);
4371 }
4372
4373 if let Some(tif) = binance_tif {
4374 builder = builder.time_in_force(tif);
4375 }
4376
4377 if let Some(trailing) = trailing_params(&kind, price).map_err(BuildOrderError::Trailing)? {
4380 builder = builder.trailing_delta(i64::from(trailing.trailing_delta));
4381 if let Some(limit) = trailing.price {
4382 builder = builder.price(limit);
4383 }
4384 if let Some(activation_price) = trailing.stop_price {
4385 builder = builder.stop_price(activation_price);
4386 }
4387 }
4388
4389 match kind {
4393 OrderKind::Limit => {
4394 builder = builder.price(price);
4395 }
4396 OrderKind::Stop { trigger_price } | OrderKind::TakeProfit { trigger_price } => {
4397 builder = builder.stop_price(trigger_price);
4398 }
4399 OrderKind::StopLimit { trigger_price } | OrderKind::TakeProfitLimit { trigger_price } => {
4400 builder = builder.price(price).stop_price(trigger_price);
4401 }
4402 OrderKind::TrailingStop { .. }
4405 | OrderKind::TrailingStopFixedLimit { .. }
4406 | OrderKind::Market
4407 | OrderKind::TrailingStopLimit { .. } => {}
4408 }
4409
4410 builder
4411 .build()
4412 .map_err(|e| BuildOrderError::Build(e.to_string()))
4413}
4414
4415fn convert_order_kind_tif_margin(
4426 kind: OrderKind,
4427 tif: TimeInForce,
4428) -> Option<(
4429 MarginAccountNewOrderTypeEnum,
4430 Option<MarginAccountNewOrderTimeInForceEnum>,
4431)> {
4432 let (binance_type, binance_tif) = classify_order_kind_tif(kind, tif)?;
4433 let margin_type = match binance_type {
4434 BinanceOrderType::Market => MarginAccountNewOrderTypeEnum::Market,
4435 BinanceOrderType::Limit => MarginAccountNewOrderTypeEnum::Limit,
4436 BinanceOrderType::LimitMaker => MarginAccountNewOrderTypeEnum::LimitMaker,
4437 BinanceOrderType::StopLoss => MarginAccountNewOrderTypeEnum::StopLoss,
4438 BinanceOrderType::StopLossLimit => MarginAccountNewOrderTypeEnum::StopLossLimit,
4439 BinanceOrderType::TakeProfit => MarginAccountNewOrderTypeEnum::TakeProfit,
4440 BinanceOrderType::TakeProfitLimit => MarginAccountNewOrderTypeEnum::TakeProfitLimit,
4441 };
4442 let margin_tif = binance_tif.map(|t| match t {
4443 BinanceTimeInForce::Gtc => MarginAccountNewOrderTimeInForceEnum::Gtc,
4444 BinanceTimeInForce::Ioc => MarginAccountNewOrderTimeInForceEnum::Ioc,
4445 BinanceTimeInForce::Fok => MarginAccountNewOrderTimeInForceEnum::Fok,
4446 });
4447 Some((margin_type, margin_tif))
4448}
4449
4450#[cfg(test)]
4451#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
4453 use super::*;
4454 use crate::client::binance::shared::convert_open_order;
4455 use crate::notice::{AccountNotice, NoticeKind};
4456 use crate::order::state::ActiveOrderState;
4457 use binance_sdk::margin_trading::rest_api::QueryMarginAccountsOpenOrdersResponseInner;
4458
4459 #[test]
4460 fn margin_side_effect_default_is_auto_borrow_repay() {
4461 assert_eq!(
4462 MarginSideEffect::default(),
4463 MarginSideEffect::AutoBorrowRepay
4464 );
4465 }
4466
4467 #[test]
4468 fn margin_side_effect_wire_strings() {
4469 assert_eq!(
4470 MarginSideEffect::AutoBorrowRepay.as_binance_str(),
4471 "AUTO_BORROW_REPAY"
4472 );
4473 assert_eq!(
4474 MarginSideEffect::NoBorrow.as_binance_str(),
4475 "NO_SIDE_EFFECT"
4476 );
4477 }
4478
4479 #[test]
4480 fn margin_side_effect_serde_round_trip() {
4481 assert_eq!(
4483 serde_json::to_string(&MarginSideEffect::AutoBorrowRepay).unwrap(),
4484 r#""auto_borrow_repay""#
4485 );
4486 assert_eq!(
4487 serde_json::to_string(&MarginSideEffect::NoBorrow).unwrap(),
4488 r#""no_borrow""#
4489 );
4490 assert_eq!(
4491 serde_json::from_str::<MarginSideEffect>(r#""auto_borrow_repay""#).unwrap(),
4492 MarginSideEffect::AutoBorrowRepay
4493 );
4494 assert_eq!(
4495 serde_json::from_str::<MarginSideEffect>(r#""no_borrow""#).unwrap(),
4496 MarginSideEffect::NoBorrow
4497 );
4498 }
4499
4500 #[test]
4501 fn config_debug_redacts_secrets() {
4502 let config = BinanceMarginConfig::new(
4503 "my_api_key".to_string(),
4504 "my_secret_key".to_string(),
4505 false,
4506 false,
4507 Vec::new(),
4508 MarginSideEffect::default(),
4509 );
4510 let debug = format!("{config:?}");
4511 assert!(!debug.contains("my_api_key"));
4512 assert!(!debug.contains("my_secret_key"));
4513 assert!(debug.contains("***"));
4514 }
4515
4516 #[test]
4517 fn cross_margin_uses_common_case_defaults() {
4518 let config = BinanceMarginConfig::cross_margin("k".to_string(), "s".to_string());
4519 assert!(!config.testnet);
4520 assert!(!config.is_isolated);
4521 assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
4522 assert_eq!(config.api_key(), "k");
4523 }
4524
4525 #[test]
4526 fn config_deserializes_with_defaults() {
4527 let config: BinanceMarginConfig =
4529 serde_json::from_str(r#"{"api_key":"k","secret_key":"s","testnet":false}"#)
4530 .expect("deserialize");
4531 assert!(!config.is_isolated);
4532 assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
4533 }
4534
4535 #[test]
4536 fn config_deserializes_explicit_side_effect() {
4537 let config: BinanceMarginConfig = serde_json::from_str(
4538 r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true,"side_effect":"no_borrow"}"#,
4539 )
4540 .expect("deserialize");
4541 assert!(config.is_isolated);
4542 assert_eq!(config.side_effect, MarginSideEffect::NoBorrow);
4543 }
4544
4545 #[test]
4546 fn isolated_ctor_sets_symbols_and_flag() {
4547 let symbols = vec![
4548 InstrumentNameExchange::new("BTCUSDT"),
4549 InstrumentNameExchange::new("ETHUSDT"),
4550 ];
4551 let config =
4552 BinanceMarginConfig::isolated("k".to_string(), "s".to_string(), symbols.clone());
4553 assert!(config.is_isolated);
4554 assert_eq!(config.isolated_symbols, symbols);
4555 assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
4556 }
4557
4558 #[test]
4559 fn isolated_config_with_symbols_constructs() {
4560 let config = BinanceMarginConfig::isolated(
4562 "k".to_string(),
4563 "s".to_string(),
4564 vec![InstrumentNameExchange::new("BTCUSDT")],
4565 );
4566 let _client = BinanceMargin::new(config); }
4568
4569 #[test]
4570 #[should_panic(expected = "non-empty isolated_symbols")]
4571 fn isolated_empty_symbols_panics_at_new() {
4572 let config = BinanceMarginConfig::new(
4574 "k".to_string(),
4575 "s".to_string(),
4576 false,
4577 true,
4578 Vec::new(),
4579 MarginSideEffect::default(),
4580 );
4581 let _ = BinanceMargin::new(config);
4582 }
4583
4584 #[test]
4585 #[should_panic(expected = "non-empty isolated_symbols")]
4586 fn isolated_empty_symbols_panics_via_deserialize_path() {
4587 let config: BinanceMarginConfig = serde_json::from_str(
4590 r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true}"#,
4591 )
4592 .expect("deserialize");
4593 assert!(config.isolated_symbols.is_empty());
4594 let _ = BinanceMargin::new(config);
4595 }
4596
4597 #[test]
4598 fn isolated_flag_maps_mode() {
4599 assert_eq!(
4602 MarginAccountNewOrderIsIsolatedEnum::from_flag(true).as_str(),
4603 "TRUE"
4604 );
4605 assert_eq!(
4606 MarginAccountNewOrderIsIsolatedEnum::from_flag(false).as_str(),
4607 "FALSE"
4608 );
4609 assert_eq!(
4610 MarginAccountCancelOrderIsIsolatedEnum::from_flag(true).as_str(),
4611 "TRUE"
4612 );
4613 assert_eq!(
4614 MarginAccountCancelOrderIsIsolatedEnum::from_flag(false).as_str(),
4615 "FALSE"
4616 );
4617 assert_eq!(
4618 QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(true).as_str(),
4619 "TRUE"
4620 );
4621 assert_eq!(
4622 QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(false).as_str(),
4623 "FALSE"
4624 );
4625 assert_eq!(
4626 QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(true).as_str(),
4627 "TRUE"
4628 );
4629 assert_eq!(
4630 QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(false).as_str(),
4631 "FALSE"
4632 );
4633 }
4634
4635 #[test]
4640 fn cancel_params_cross_with_parseable_order_id() {
4641 let p = build_cancel_order_params(
4642 "BTCUSDT".to_string(),
4643 Some(&OrderId::new("12345")),
4644 &ClientOrderId::new("cid-1"),
4645 false,
4646 )
4647 .expect("build");
4648 assert_eq!(p.symbol, "BTCUSDT");
4649 assert_eq!(p.is_isolated.as_ref().map(|i| i.as_str()), Some("FALSE"));
4650 assert_eq!(p.order_id, Some(12345));
4651 assert_eq!(p.orig_client_order_id, None);
4652 }
4653
4654 #[test]
4655 fn cancel_params_isolated_is_config_driven() {
4656 let p = build_cancel_order_params(
4657 "BTCUSDT".to_string(),
4658 Some(&OrderId::new("12345")),
4659 &ClientOrderId::new("cid-1"),
4660 true,
4661 )
4662 .expect("build");
4663 assert_eq!(p.is_isolated.as_ref().map(|i| i.as_str()), Some("TRUE"));
4664 assert_eq!(p.order_id, Some(12345));
4665 }
4666
4667 #[test]
4668 fn cancel_params_non_i64_order_id_falls_back_to_cid() {
4669 let p = build_cancel_order_params(
4670 "BTCUSDT".to_string(),
4671 Some(&OrderId::new("not-an-i64")),
4672 &ClientOrderId::new("cid-1"),
4673 false,
4674 )
4675 .expect("build");
4676 assert_eq!(p.order_id, None);
4677 assert_eq!(p.orig_client_order_id.as_deref(), Some("cid-1"));
4678 }
4679
4680 #[test]
4681 fn cancel_params_absent_order_id_uses_cid() {
4682 let p = build_cancel_order_params(
4683 "BTCUSDT".to_string(),
4684 None,
4685 &ClientOrderId::new("cid-1"),
4686 false,
4687 )
4688 .expect("build");
4689 assert_eq!(p.order_id, None);
4690 assert_eq!(p.orig_client_order_id.as_deref(), Some("cid-1"));
4691 }
4692
4693 use crate::order::TrailingOffsetType;
4698
4699 fn params(
4701 side: Side,
4702 price: Option<Decimal>,
4703 kind: OrderKind,
4704 tif: TimeInForce,
4705 side_effect: MarginSideEffect,
4706 ) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
4707 build_new_order_params(
4708 "BTCUSDT".to_string(),
4709 side,
4710 price,
4711 Decimal::from(2),
4712 kind,
4713 tif,
4714 "cid-1".to_string(),
4715 side_effect,
4716 false, )
4718 }
4719
4720 fn gtc() -> TimeInForce {
4721 TimeInForce::GoodUntilCancelled { post_only: false }
4722 }
4723
4724 #[test]
4725 fn new_order_limit_maps_core_fields() {
4726 let p = params(
4727 Side::Buy,
4728 Some(Decimal::from(50_000)),
4729 OrderKind::Limit,
4730 gtc(),
4731 MarginSideEffect::AutoBorrowRepay,
4732 )
4733 .expect("build");
4734
4735 assert_eq!(p.symbol, "BTCUSDT");
4736 assert_eq!(p.side.as_str(), "BUY");
4737 assert_eq!(p.r#type.as_str(), "LIMIT");
4738 assert_eq!(p.quantity, Some(Decimal::from(2)));
4739 assert_eq!(p.price, Some(Decimal::from(50_000)));
4740 assert_eq!(p.stop_price, None);
4741 assert_eq!(p.time_in_force.as_ref().map(|t| t.as_str()), Some("GTC"));
4742 assert_eq!(p.new_client_order_id.as_deref(), Some("cid-1"));
4743 assert_eq!(
4745 p.new_order_resp_type.as_ref().map(|r| r.as_str()),
4746 Some("FULL")
4747 );
4748 }
4749
4750 #[test]
4751 fn new_order_is_isolated_is_config_driven() {
4752 let cross = build_new_order_params(
4754 "BTCUSDT".to_string(),
4755 Side::Sell,
4756 Some(Decimal::from(10)),
4757 Decimal::from(2),
4758 OrderKind::Limit,
4759 gtc(),
4760 "cid-1".to_string(),
4761 MarginSideEffect::AutoBorrowRepay,
4762 false,
4763 )
4764 .expect("build");
4765 assert_eq!(
4766 cross.is_isolated.as_ref().map(|i| i.as_str()),
4767 Some("FALSE")
4768 );
4769 assert_eq!(cross.side.as_str(), "SELL");
4770
4771 let isolated = build_new_order_params(
4773 "BTCUSDT".to_string(),
4774 Side::Sell,
4775 Some(Decimal::from(10)),
4776 Decimal::from(2),
4777 OrderKind::Limit,
4778 gtc(),
4779 "cid-1".to_string(),
4780 MarginSideEffect::AutoBorrowRepay,
4781 true,
4782 )
4783 .expect("build");
4784 assert_eq!(
4785 isolated.is_isolated.as_ref().map(|i| i.as_str()),
4786 Some("TRUE")
4787 );
4788 }
4789
4790 #[test]
4791 fn side_effect_and_auto_repay_gating() {
4792 let auto = params(
4794 Side::Buy,
4795 Some(Decimal::from(1)),
4796 OrderKind::Limit,
4797 gtc(),
4798 MarginSideEffect::AutoBorrowRepay,
4799 )
4800 .expect("build");
4801 assert_eq!(
4802 auto.side_effect_type.as_ref().map(|s| s.as_str()),
4803 Some("AUTO_BORROW_REPAY")
4804 );
4805 assert_eq!(auto.auto_repay_at_cancel, Some(true));
4806
4807 let no_borrow = params(
4809 Side::Buy,
4810 Some(Decimal::from(1)),
4811 OrderKind::Limit,
4812 gtc(),
4813 MarginSideEffect::NoBorrow,
4814 )
4815 .expect("build");
4816 assert_eq!(
4817 no_borrow.side_effect_type.as_ref().map(|s| s.as_str()),
4818 Some("NO_SIDE_EFFECT")
4819 );
4820 assert_eq!(no_borrow.auto_repay_at_cancel, None);
4821 }
4822
4823 #[test]
4824 fn new_order_market_has_no_price_or_tif() {
4825 let p = params(
4826 Side::Buy,
4827 None,
4828 OrderKind::Market,
4829 gtc(),
4830 MarginSideEffect::AutoBorrowRepay,
4831 )
4832 .expect("build");
4833 assert_eq!(p.r#type.as_str(), "MARKET");
4834 assert_eq!(p.price, None);
4835 assert_eq!(p.stop_price, None);
4836 assert!(p.time_in_force.is_none());
4837 }
4838
4839 #[test]
4840 fn post_only_limit_maps_to_limit_maker() {
4841 let p = params(
4842 Side::Buy,
4843 Some(Decimal::from(10)),
4844 OrderKind::Limit,
4845 TimeInForce::GoodUntilCancelled { post_only: true },
4846 MarginSideEffect::AutoBorrowRepay,
4847 )
4848 .expect("build");
4849 assert_eq!(p.r#type.as_str(), "LIMIT_MAKER");
4850 assert!(p.time_in_force.is_none());
4852 }
4853
4854 #[test]
4855 fn conditional_kinds_set_stop_price() {
4856 let trigger = Decimal::from(48_000);
4857
4858 let stop = params(
4859 Side::Sell,
4860 None,
4861 OrderKind::Stop {
4862 trigger_price: trigger,
4863 },
4864 gtc(),
4865 MarginSideEffect::AutoBorrowRepay,
4866 )
4867 .expect("build");
4868 assert_eq!(stop.r#type.as_str(), "STOP_LOSS");
4869 assert_eq!(stop.stop_price, Some(trigger));
4870 assert_eq!(stop.price, None);
4871
4872 let stop_limit = params(
4873 Side::Sell,
4874 Some(Decimal::from(47_900)),
4875 OrderKind::StopLimit {
4876 trigger_price: trigger,
4877 },
4878 gtc(),
4879 MarginSideEffect::AutoBorrowRepay,
4880 )
4881 .expect("build");
4882 assert_eq!(stop_limit.r#type.as_str(), "STOP_LOSS_LIMIT");
4883 assert_eq!(stop_limit.stop_price, Some(trigger));
4884 assert_eq!(stop_limit.price, Some(Decimal::from(47_900)));
4885 assert_eq!(
4886 stop_limit.time_in_force.as_ref().map(|t| t.as_str()),
4887 Some("GTC")
4888 );
4889
4890 let take_profit = params(
4891 Side::Sell,
4892 None,
4893 OrderKind::TakeProfit {
4894 trigger_price: trigger,
4895 },
4896 gtc(),
4897 MarginSideEffect::AutoBorrowRepay,
4898 )
4899 .expect("build");
4900 assert_eq!(take_profit.r#type.as_str(), "TAKE_PROFIT");
4901 assert_eq!(take_profit.stop_price, Some(trigger));
4902 }
4903
4904 #[test]
4907 fn a_trailing_stop_maps_to_stop_loss_with_trailing_delta() {
4908 let cases = [
4909 (Decimal::from(25), TrailingOffsetType::BasisPoints, 25),
4910 (Decimal::new(15, 1), TrailingOffsetType::Percentage, 150),
4911 ];
4912 for (offset, offset_type, trailing_delta) in cases {
4913 let p = params(
4914 Side::Sell,
4915 None,
4916 OrderKind::TrailingStop {
4917 offset,
4918 offset_type,
4919 activation_price: None,
4920 },
4921 gtc(),
4922 MarginSideEffect::NoBorrow,
4923 )
4924 .expect("build");
4925 assert_eq!(p.r#type.as_str(), "STOP_LOSS");
4926 assert_eq!(p.trailing_delta, Some(trailing_delta), "{offset_type:?}");
4927 assert_eq!(p.stop_price, None);
4928 assert_eq!(p.price, None);
4929 assert!(p.time_in_force.is_none());
4930 }
4931 }
4932
4933 #[test]
4937 fn trailing_orders_carry_their_activation_price_and_fixed_limit() {
4938 let activation = Some(Decimal::from(50_000));
4939 let trailing_stop = params(
4940 Side::Sell,
4941 None,
4942 OrderKind::TrailingStop {
4943 offset: Decimal::from(25),
4944 offset_type: TrailingOffsetType::BasisPoints,
4945 activation_price: activation,
4946 },
4947 gtc(),
4948 MarginSideEffect::NoBorrow,
4949 )
4950 .expect("build");
4951 assert_eq!(trailing_stop.r#type.as_str(), "STOP_LOSS");
4952 assert_eq!(trailing_stop.trailing_delta, Some(25));
4953 assert_eq!(trailing_stop.stop_price, activation);
4954 assert_eq!(trailing_stop.price, None);
4955
4956 for activation_price in [None, activation] {
4957 let fixed_limit = params(
4958 Side::Sell,
4959 Some(Decimal::from(49_000)),
4960 OrderKind::TrailingStopFixedLimit {
4961 offset: Decimal::new(15, 1),
4962 offset_type: TrailingOffsetType::Percentage,
4963 activation_price,
4964 },
4965 TimeInForce::ImmediateOrCancel,
4966 MarginSideEffect::NoBorrow,
4967 )
4968 .expect("build");
4969 assert_eq!(fixed_limit.r#type.as_str(), "STOP_LOSS_LIMIT");
4970 assert_eq!(fixed_limit.trailing_delta, Some(150));
4971 assert_eq!(fixed_limit.price, Some(Decimal::from(49_000)));
4972 assert_eq!(fixed_limit.stop_price, activation_price);
4973 assert_eq!(
4974 fixed_limit.time_in_force.as_ref().map(|t| t.as_str()),
4975 Some("IOC")
4976 );
4977 }
4978
4979 let absolute = params(
4980 Side::Sell,
4981 Some(Decimal::from(49_000)),
4982 OrderKind::TrailingStopFixedLimit {
4983 offset: Decimal::from(100),
4984 offset_type: TrailingOffsetType::Absolute,
4985 activation_price: None,
4986 },
4987 gtc(),
4988 MarginSideEffect::NoBorrow,
4989 );
4990 assert!(matches!(absolute, Err(BuildOrderError::Unsupported)));
4991
4992 let no_limit = params(
4994 Side::Sell,
4995 None,
4996 OrderKind::TrailingStopFixedLimit {
4997 offset: Decimal::from(25),
4998 offset_type: TrailingOffsetType::BasisPoints,
4999 activation_price: None,
5000 },
5001 gtc(),
5002 MarginSideEffect::NoBorrow,
5003 );
5004 assert!(matches!(
5005 no_limit,
5006 Err(BuildOrderError::Trailing(TrailingOrderError::LimitPrice(
5007 None
5008 )))
5009 ));
5010 let zero_activation = params(
5011 Side::Sell,
5012 None,
5013 OrderKind::TrailingStop {
5014 offset: Decimal::from(25),
5015 offset_type: TrailingOffsetType::BasisPoints,
5016 activation_price: Some(Decimal::ZERO),
5017 },
5018 gtc(),
5019 MarginSideEffect::NoBorrow,
5020 );
5021 assert!(matches!(
5022 zero_activation,
5023 Err(BuildOrderError::Trailing(
5024 TrailingOrderError::NonPositiveActivation(_)
5025 ))
5026 ));
5027 }
5028
5029 #[test]
5030 fn absolute_trailing_offsets_and_trailing_stop_limits_are_unsupported() {
5031 let absolute = params(
5032 Side::Sell,
5033 None,
5034 OrderKind::TrailingStop {
5035 offset: Decimal::from(100),
5036 offset_type: TrailingOffsetType::Absolute,
5037 activation_price: None,
5038 },
5039 gtc(),
5040 MarginSideEffect::AutoBorrowRepay,
5041 );
5042 assert!(matches!(absolute, Err(BuildOrderError::Unsupported)));
5043
5044 let trailing_stop_limit = params(
5045 Side::Sell,
5046 Some(Decimal::from(100)),
5047 OrderKind::TrailingStopLimit {
5048 offset: Decimal::from(100),
5049 offset_type: TrailingOffsetType::BasisPoints,
5050 stop_price: Decimal::from(95),
5051 limit_offset: Decimal::from(10),
5052 },
5053 gtc(),
5054 MarginSideEffect::AutoBorrowRepay,
5055 );
5056 assert!(matches!(
5057 trailing_stop_limit,
5058 Err(BuildOrderError::Unsupported)
5059 ));
5060 }
5061
5062 #[test]
5064 fn a_fractional_trailing_offset_is_refused() {
5065 let fractional = params(
5066 Side::Sell,
5067 None,
5068 OrderKind::TrailingStop {
5069 offset: Decimal::new(155, 3),
5070 offset_type: TrailingOffsetType::Percentage,
5071 activation_price: None,
5072 },
5073 gtc(),
5074 MarginSideEffect::AutoBorrowRepay,
5075 );
5076 assert!(matches!(
5077 fractional,
5078 Err(BuildOrderError::Trailing(TrailingOrderError::Delta(
5079 crate::client::binance::shared::TrailingDeltaError::Invalid(_)
5080 )))
5081 ));
5082 }
5083
5084 #[test]
5085 fn tif_variants_map_to_margin_enum() {
5086 for (tif, expected) in [
5087 (TimeInForce::ImmediateOrCancel, "IOC"),
5088 (TimeInForce::FillOrKill, "FOK"),
5089 (gtc(), "GTC"),
5090 ] {
5091 let p = params(
5092 Side::Buy,
5093 Some(Decimal::from(1)),
5094 OrderKind::Limit,
5095 tif,
5096 MarginSideEffect::AutoBorrowRepay,
5097 )
5098 .expect("build");
5099 assert_eq!(p.time_in_force.as_ref().map(|t| t.as_str()), Some(expected));
5100 }
5101 }
5102
5103 fn user_asset(
5108 asset: &str,
5109 free: &str,
5110 locked: &str,
5111 borrowed: &str,
5112 interest: &str,
5113 ) -> QueryCrossMarginAccountDetailsResponseUserAssetsInner {
5114 let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
5115 a.asset = Some(asset.to_string());
5116 a.free = Some(free.to_string());
5117 a.locked = Some(locked.to_string());
5118 a.borrowed = Some(borrowed.to_string());
5119 a.interest = Some(interest.to_string());
5120 a
5121 }
5122
5123 #[test]
5124 fn margin_balance_entry_maps_debt() {
5125 let ab =
5127 convert_margin_balance_entry(user_asset("USDT", "10", "2", "3", "0.5"), Utc::now())
5128 .expect("convert");
5129 assert_eq!(ab.asset.name().as_str(), "USDT");
5130 assert_eq!(ab.balance.total, Decimal::from(12));
5131 assert_eq!(ab.balance.free, Decimal::from(10));
5132 assert!(
5133 ab.balance.margin.is_some(),
5134 "REST snapshot must populate margin"
5135 );
5136 assert_eq!(ab.balance.net_asset(), Decimal::from(9));
5138 }
5139
5140 #[test]
5141 fn margin_balance_short_is_negative_net() {
5142 let ab = convert_margin_balance_entry(user_asset("BTC", "1", "0", "5", "0"), Utc::now())
5144 .expect("convert");
5145 assert_eq!(ab.balance.net_asset(), Decimal::from(-4));
5146 }
5147
5148 #[test]
5149 fn margin_balance_missing_debt_defaults_to_zero() {
5150 let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
5153 a.asset = Some("ETH".to_string());
5154 a.free = Some("4".to_string());
5155 a.locked = Some("0".to_string());
5156 let ab = convert_margin_balance_entry(a, Utc::now()).expect("convert");
5157 assert!(ab.balance.margin.is_some());
5158 assert_eq!(ab.balance.net_asset(), Decimal::from(4));
5159 }
5160
5161 #[test]
5162 fn margin_balance_missing_free_is_dropped() {
5163 let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
5165 a.asset = Some("USDT".to_string());
5166 a.locked = Some("0".to_string());
5167 assert!(convert_margin_balance_entry(a, Utc::now()).is_none());
5168 }
5169
5170 #[test]
5171 fn margin_balance_filtering() {
5172 let assets = vec![
5173 user_asset("USDT", "10", "0", "0", "0"),
5174 user_asset("BTC", "1", "0", "0", "0"),
5175 user_asset("ETH", "5", "0", "0", "0"),
5176 ];
5177 assert_eq!(
5179 filter_and_convert_margin_balances(assets.clone(), &[]).len(),
5180 3
5181 );
5182 let filtered = filter_and_convert_margin_balances(
5184 assets,
5185 &[AssetNameExchange::new("BTC"), AssetNameExchange::new("ETH")],
5186 );
5187 assert_eq!(filtered.len(), 2);
5188 assert!(filtered.iter().all(|b| b.asset.name().as_str() != "USDT"));
5189 }
5190
5191 #[test]
5192 fn margin_balance_filtering_large_slice_uses_hashset_branch() {
5193 let assets = vec![
5195 user_asset("BTC", "1", "0", "0", "0"),
5196 user_asset("ETH", "5", "0", "0", "0"),
5197 user_asset("DOGE", "9", "0", "0", "0"), ];
5199 let mut requested: Vec<AssetNameExchange> = (0..17)
5201 .map(|i| AssetNameExchange::new(format!("A{i:02}")))
5202 .collect();
5203 requested.push(AssetNameExchange::new("BTC"));
5204 requested.push(AssetNameExchange::new("ETH"));
5205 assert!(
5206 requested.len() > 16,
5207 "must exceed the linear-scan threshold"
5208 );
5209
5210 let filtered = filter_and_convert_margin_balances(assets, &requested);
5211 assert_eq!(filtered.len(), 2);
5212 assert!(filtered.iter().all(|b| b.asset.name().as_str() != "DOGE"));
5213 }
5214
5215 use binance_sdk::margin_trading::rest_api::{
5220 QueryIsolatedMarginAccountInfoResponseAssetsInnerBaseAsset as IsoBase,
5221 QueryIsolatedMarginAccountInfoResponseAssetsInnerQuoteAsset as IsoQuote,
5222 };
5223
5224 fn d(s: &str) -> Decimal {
5225 Decimal::from_str(s).unwrap()
5226 }
5227
5228 fn iso_base(asset: &str, free: &str, locked: &str, borrowed: &str, interest: &str) -> IsoBase {
5229 let mut b = IsoBase::new();
5230 b.asset = Some(asset.to_string());
5231 b.free = Some(free.to_string());
5232 b.locked = Some(locked.to_string());
5233 b.borrowed = Some(borrowed.to_string());
5234 b.interest = Some(interest.to_string());
5235 b
5236 }
5237
5238 fn iso_quote(
5239 asset: &str,
5240 free: &str,
5241 locked: &str,
5242 borrowed: &str,
5243 interest: &str,
5244 ) -> IsoQuote {
5245 let mut q = IsoQuote::new();
5246 q.asset = Some(asset.to_string());
5247 q.free = Some(free.to_string());
5248 q.locked = Some(locked.to_string());
5249 q.borrowed = Some(borrowed.to_string());
5250 q.interest = Some(interest.to_string());
5251 q
5252 }
5253
5254 fn iso_entry(
5255 symbol: &str,
5256 base: IsoBase,
5257 quote: IsoQuote,
5258 margin_level: Option<&str>,
5259 margin_ratio: Option<&str>,
5260 liquidate_price: Option<&str>,
5261 ) -> QueryIsolatedMarginAccountInfoResponseAssetsInner {
5262 let mut e = QueryIsolatedMarginAccountInfoResponseAssetsInner::new();
5263 e.symbol = Some(symbol.to_string());
5264 e.base_asset = Some(Box::new(base));
5265 e.quote_asset = Some(Box::new(quote));
5266 e.margin_level = margin_level.map(str::to_string);
5267 e.margin_ratio = margin_ratio.map(str::to_string);
5268 e.liquidate_price = liquidate_price.map(str::to_string);
5269 e
5270 }
5271
5272 fn isolated_client(symbols: &[&str]) -> BinanceMargin {
5273 let config = BinanceMarginConfig::isolated(
5274 "k".to_string(),
5275 "s".to_string(),
5276 symbols
5277 .iter()
5278 .map(|s| InstrumentNameExchange::new(*s))
5279 .collect(),
5280 );
5281 BinanceMargin::new(config)
5282 }
5283
5284 #[test]
5285 fn isolated_assets_map_base_quote_and_risk() {
5286 let entry = iso_entry(
5289 "BTCUSDT",
5290 iso_base("BTC", "0.5", "0.1", "0.2", "0.001"),
5291 iso_quote("USDT", "1000", "50", "300", "1.5"),
5292 Some("3.5"),
5293 Some("0.12"),
5294 Some("48000"),
5295 );
5296 let map = convert_isolated_margin_assets(vec![entry]);
5297 let state = map
5298 .get(&InstrumentNameExchange::new("BTCUSDT"))
5299 .expect("entry present");
5300
5301 assert_eq!(state.base.asset.name().as_str(), "BTC");
5302 assert_eq!(state.base.balance.total, d("0.6"));
5303 assert_eq!(state.base.balance.free, d("0.5"));
5304 assert_eq!(state.base.balance.net_asset(), d("0.4"));
5305
5306 assert_eq!(state.quote.asset.name().as_str(), "USDT");
5307 assert_eq!(state.quote.balance.total, d("1050"));
5308 assert_eq!(state.quote.balance.net_asset(), d("750"));
5309
5310 assert_eq!(state.risk.margin_level, Some(d("3.5")));
5311 assert_eq!(state.risk.margin_ratio, Some(d("0.12")));
5312 assert_eq!(state.risk.liquidation_price, Some(d("48000")));
5313 }
5314
5315 #[test]
5316 fn isolated_base_short_is_negative_net() {
5317 let entry = iso_entry(
5319 "BTCUSDT",
5320 iso_base("BTC", "0", "0", "1.5", "0.001"),
5321 iso_quote("USDT", "100", "0", "0", "0"),
5322 None,
5323 None,
5324 None,
5325 );
5326 let map = convert_isolated_margin_assets(vec![entry]);
5327 let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
5328 assert_eq!(state.base.balance.net_asset(), d("-1.5"));
5329 }
5330
5331 #[test]
5332 fn isolated_missing_debt_defaults_zero_and_risk_optional() {
5333 let mut base = IsoBase::new();
5336 base.asset = Some("BTC".to_string());
5337 base.free = Some("0.5".to_string());
5338 base.locked = Some("0".to_string());
5339 let mut quote = IsoQuote::new();
5340 quote.asset = Some("USDT".to_string());
5341 quote.free = Some("100".to_string());
5342 quote.locked = Some("0".to_string());
5343
5344 let map = convert_isolated_margin_assets(vec![iso_entry(
5345 "BTCUSDT", base, quote, None, None, None,
5346 )]);
5347 let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
5348 assert!(state.base.balance.margin.is_some());
5349 assert_eq!(state.base.balance.net_asset(), d("0.5"));
5350 assert_eq!(state.risk.margin_level, None);
5351 assert_eq!(state.risk.liquidation_price, None);
5352 }
5353
5354 #[test]
5355 fn isolated_unparseable_risk_is_none_but_entry_kept() {
5356 let entry = iso_entry(
5357 "BTCUSDT",
5358 iso_base("BTC", "1", "0", "0", "0"),
5359 iso_quote("USDT", "1", "0", "0", "0"),
5360 Some("not_a_number"),
5361 None,
5362 None,
5363 );
5364 let map = convert_isolated_margin_assets(vec![entry]);
5365 let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
5366 assert_eq!(state.risk.margin_level, None);
5367 }
5368
5369 #[test]
5370 fn isolated_missing_base_free_drops_entry() {
5371 let mut base = IsoBase::new();
5373 base.asset = Some("BTC".to_string());
5374 base.locked = Some("0".to_string()); let entry = iso_entry(
5376 "BTCUSDT",
5377 base,
5378 iso_quote("USDT", "100", "0", "0", "0"),
5379 None,
5380 None,
5381 None,
5382 );
5383 assert!(convert_isolated_margin_assets(vec![entry]).is_empty());
5384 }
5385
5386 #[test]
5387 fn isolated_missing_symbol_drops_entry() {
5388 let mut e = QueryIsolatedMarginAccountInfoResponseAssetsInner::new();
5389 e.base_asset = Some(Box::new(iso_base("BTC", "1", "0", "0", "0")));
5390 e.quote_asset = Some(Box::new(iso_quote("USDT", "1", "0", "0", "0")));
5391 assert!(convert_isolated_margin_assets(vec![e]).is_empty());
5393 }
5394
5395 #[test]
5396 fn chunk_symbols_batches_by_five() {
5397 let syms: Vec<InstrumentNameExchange> = (0..12)
5398 .map(|i| InstrumentNameExchange::new(format!("S{i:02}")))
5399 .collect();
5400 let chunks = chunk_symbols(&syms);
5401 assert_eq!(chunks.len(), 3, "12 symbols → 5 + 5 + 2");
5402 assert_eq!(chunks[0].split(',').count(), 5);
5403 assert_eq!(chunks[1].split(',').count(), 5);
5404 assert_eq!(chunks[2].split(',').count(), 2);
5405
5406 let three: Vec<_> = (0..3)
5408 .map(|i| InstrumentNameExchange::new(format!("S{i}")))
5409 .collect();
5410 assert_eq!(chunk_symbols(&three).len(), 1);
5411
5412 let five: Vec<_> = (0..5)
5414 .map(|i| InstrumentNameExchange::new(format!("S{i}")))
5415 .collect();
5416 let c = chunk_symbols(&five);
5417 assert_eq!(c.len(), 1);
5418 assert_eq!(c[0].split(',').count(), 5);
5419
5420 assert!(chunk_symbols(&[]).is_empty());
5422 }
5423
5424 #[test]
5425 fn effective_isolated_set_empty_returns_all_configured() {
5426 let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
5427 assert_eq!(
5428 client.effective_isolated_set(&[]),
5429 vec![
5430 InstrumentNameExchange::new("BTCUSDT"),
5431 InstrumentNameExchange::new("ETHUSDT"),
5432 ]
5433 );
5434 }
5435
5436 #[test]
5437 fn effective_isolated_set_intersects_requested() {
5438 let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
5439 assert_eq!(
5440 client.effective_isolated_set(&[InstrumentNameExchange::new("ETHUSDT")]),
5441 vec![InstrumentNameExchange::new("ETHUSDT")]
5442 );
5443 }
5444
5445 #[test]
5446 fn effective_isolated_set_skips_out_of_set() {
5447 let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
5448 assert_eq!(
5450 client.effective_isolated_set(&[
5451 InstrumentNameExchange::new("BTCUSDT"),
5452 InstrumentNameExchange::new("DOGEUSDT"),
5453 ]),
5454 vec![InstrumentNameExchange::new("BTCUSDT")]
5455 );
5456 }
5457
5458 #[tokio::test]
5459 async fn fetch_balances_isolated_returns_empty() {
5460 let client = isolated_client(&["BTCUSDT"]);
5463 assert!(client.fetch_balances(&[]).await.expect("ok").is_empty());
5464 }
5465
5466 fn open_order(order_id: Option<i64>, side: &str) -> QueryMarginAccountsOpenOrdersResponseInner {
5467 let mut o = QueryMarginAccountsOpenOrdersResponseInner::new();
5468 o.order_id = order_id;
5469 o.client_order_id = Some("cid-9".to_string());
5470 o.side = Some(side.to_string());
5471 o.price = Some("50000".to_string());
5472 o.orig_qty = Some("2".to_string());
5473 o.executed_qty = Some("0.5".to_string());
5474 o.r#type = Some("LIMIT".to_string());
5475 o.time_in_force = Some("GTC".to_string());
5476 o.time = Some(1_700_000_000_000);
5477 o.status = Some("PARTIALLY_FILLED".to_string());
5478 o
5479 }
5480
5481 #[test]
5482 fn margin_open_order_converts() {
5483 let inst = InstrumentNameExchange::new("BTCUSDT");
5484 let order = convert_open_order(
5485 &open_order(Some(42), "BUY"),
5486 ExchangeId::BinanceMargin,
5487 &inst,
5488 )
5489 .expect("convert");
5490 assert_eq!(order.key.exchange, ExchangeId::BinanceMargin);
5491 assert_eq!(
5492 order.state.id.assigned().map(|id| id.0.as_str()),
5493 Some("42")
5494 );
5495 assert_eq!(order.key.cid.0.as_str(), "cid-9");
5496 assert_eq!(order.side, Side::Buy);
5497 assert_eq!(order.price, Some(Decimal::from(50_000)));
5498 assert_eq!(order.quantity, Decimal::from(2));
5499 assert_eq!(order.state.filled_quantity, Decimal::new(5, 1));
5500 assert_eq!(order.kind, OrderKind::Limit);
5501 }
5502
5503 #[test]
5506 fn margin_open_order_that_is_not_live_is_dropped() {
5507 let inst = InstrumentNameExchange::new("BTCUSDT");
5508 let mut o = open_order(Some(42), "BUY");
5509 o.status = Some("CANCELED".to_string());
5510 assert!(convert_open_order(&o, ExchangeId::BinanceMargin, &inst).is_none());
5511 }
5512
5513 #[test]
5514 fn margin_open_order_missing_order_id_is_dropped() {
5515 let inst = InstrumentNameExchange::new("BTCUSDT");
5516 assert!(
5517 convert_open_order(&open_order(None, "BUY"), ExchangeId::BinanceMargin, &inst)
5518 .is_none()
5519 );
5520 }
5521
5522 #[test]
5528 fn margin_open_order_prefers_update_time_over_creation_time() {
5529 let inst = InstrumentNameExchange::new("BTCUSDT");
5530
5531 let mut updated = open_order(Some(42), "BUY");
5532 updated.update_time = Some(1_700_000_005_000);
5533 let order =
5534 convert_open_order(&updated, ExchangeId::BinanceMargin, &inst).expect("convert");
5535 assert_eq!(
5536 order.state.time_exchange,
5537 Utc.timestamp_millis_opt(1_700_000_005_000)
5538 .single()
5539 .unwrap(),
5540 "update_time must win over time"
5541 );
5542 }
5543
5544 #[test]
5547 fn margin_open_order_falls_back_to_creation_time_when_update_time_absent() {
5548 let inst = InstrumentNameExchange::new("BTCUSDT");
5549
5550 let base = open_order(Some(42), "BUY");
5551 assert!(base.update_time.is_none(), "fixture must omit update_time");
5552 let order = convert_open_order(&base, ExchangeId::BinanceMargin, &inst).expect("convert");
5553 assert_eq!(
5554 order.state.time_exchange,
5555 Utc.timestamp_millis_opt(1_700_000_000_000)
5556 .single()
5557 .unwrap(),
5558 "absent update_time falls back to time"
5559 );
5560 }
5561
5562 #[test]
5563 fn margin_open_order_owned_symbol_recovers_instrument() {
5564 let mut o = open_order(Some(42), "SELL");
5566 o.symbol = Some("ETHUSDT".to_string());
5567 let order =
5568 convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin).expect("convert");
5569 assert_eq!(order.key.instrument.name().as_str(), "ETHUSDT");
5570 assert_eq!(order.key.exchange, ExchangeId::BinanceMargin);
5571 assert_eq!(order.side, Side::Sell);
5572 }
5573
5574 #[test]
5575 fn margin_open_order_owned_symbol_missing_symbol_is_dropped() {
5576 let o = open_order(Some(42), "BUY");
5578 assert!(o.symbol.is_none());
5579 assert!(convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin).is_none());
5580 }
5581
5582 fn trade(id: Option<i64>, is_buyer: Option<bool>) -> QueryMarginAccountsTradeListResponseInner {
5583 let mut t = QueryMarginAccountsTradeListResponseInner::new();
5584 t.id = id;
5585 t.order_id = Some(7);
5586 t.is_buyer = is_buyer;
5587 t.price = Some("48000".to_string());
5588 t.qty = Some("0.25".to_string());
5589 t.commission = Some("0.001".to_string());
5590 t.commission_asset = Some("BNB".to_string());
5591 t.time = Some(1_700_000_000_000);
5592 t
5593 }
5594
5595 #[test]
5596 fn margin_trade_converts() {
5597 let inst = InstrumentNameExchange::new("BTCUSDT");
5598 let tr = convert_margin_trade(&trade(Some(11), Some(false)), &inst).expect("convert");
5599 assert_eq!(tr.id.0.as_str(), "11");
5600 assert_eq!(tr.order_id.0.as_str(), "7");
5601 assert_eq!(tr.side, Side::Sell);
5602 assert_eq!(tr.price, Decimal::from(48_000));
5603 assert_eq!(tr.quantity, Decimal::new(25, 2));
5604 assert_eq!(tr.fees.asset.name().as_str(), "BNB");
5606 assert_eq!(tr.fees.fees, Decimal::new(1, 3));
5607 }
5608
5609 #[test]
5610 fn margin_trade_missing_fields_are_dropped() {
5611 let inst = InstrumentNameExchange::new("BTCUSDT");
5612 assert!(convert_margin_trade(&trade(None, Some(true)), &inst).is_none());
5614 assert!(convert_margin_trade(&trade(Some(11), None), &inst).is_none());
5615 }
5616
5617 #[test]
5620 fn token_renew_after_applies_margin_and_floor() {
5621 let now_ms = Utc::now().timestamp_millis();
5622 let far = token_renew_after(now_ms + 86_400_000).as_secs();
5624 assert!(
5625 far > 80_000 && far <= 86_400,
5626 "expected ~24h-minus-margin, got {far}"
5627 );
5628 assert_eq!(
5630 token_renew_after(now_ms - 10_000).as_secs(),
5631 TOKEN_MIN_LIFETIME_SECS
5632 );
5633 assert_eq!(
5635 token_renew_after(now_ms + 60_000).as_secs(),
5636 TOKEN_MIN_LIFETIME_SECS
5637 );
5638 }
5639
5640 fn push(event: serde_json::Value) -> String {
5645 serde_json::json!({ "subscriptionId": 1, "event": event }).to_string()
5646 }
5647
5648 #[test]
5649 fn margin_ws_rpc_ack_yields_no_events() {
5650 let frame = serde_json::json!({ "id": "abc", "status": 200, "result": {} }).to_string();
5652 let mut buf = Vec::new();
5653 assert!(!convert_margin_user_data_events(&frame, &mut buf));
5654 assert!(buf.is_empty());
5655 }
5656
5657 #[test]
5660 fn margin_ws_unrecognised_frames_yield_no_events() {
5661 for frame in [
5662 r#"{"e":"eventStreamTerminated","E":1}"#,
5663 r#"{"subscriptionId":1,"event":{"E":1}}"#,
5664 ] {
5665 let mut buf = Vec::new();
5666 assert!(!convert_margin_user_data_events(frame, &mut buf), "{frame}");
5667 assert!(buf.is_empty(), "{frame}");
5668 }
5669 }
5670
5671 #[test]
5674 fn margin_ws_trailing_stop_report_decodes_as_a_trailing_stop() {
5675 for trailing_delta in [serde_json::json!(25), serde_json::json!("25")] {
5676 let frame = push(serde_json::json!({
5677 "e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "STOP_LOSS",
5678 "x": "NEW", "X": "NEW", "i": 12_345_i64, "c": "cid-1", "t": -1_i64,
5679 "p": "0.00000000", "P": "0.00000000", "d": trailing_delta, "D": 1_700_000_000_000_i64,
5680 "q": "2", "f": "GTC", "z": "0", "T": 1_700_000_000_000_i64,
5681 }));
5682 let mut buf = Vec::new();
5683 assert!(!convert_margin_user_data_events(&frame, &mut buf));
5684 let [event] = buf.as_slice() else {
5685 panic!("one snapshot expected for {frame}, got {buf:?}");
5686 };
5687 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
5688 panic!("expected an order snapshot, got {event:?}");
5689 };
5690 assert_eq!(
5691 snapshot.0.kind,
5692 OrderKind::TrailingStop {
5693 offset: Decimal::from(25),
5694 offset_type: TrailingOffsetType::BasisPoints,
5695 activation_price: None,
5696 }
5697 );
5698 assert_eq!(snapshot.0.price, None);
5699 }
5700 }
5701
5702 #[test]
5710 fn margin_ws_execution_report_trade_maps_to_trade_and_order_snapshot() {
5711 let frame = push(serde_json::json!({
5712 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
5713 "x": "TRADE", "X": "PARTIALLY_FILLED", "i": 12_345_i64, "c": "cid-1",
5714 "t": 99_i64, "l": "0.5", "L": "48000", "z": "0.5",
5715 "p": "48000", "q": "2", "f": "GTC",
5716 "n": "0.001", "N": "BNB", "T": 1_700_000_000_000_i64,
5717 }));
5718 let mut buf = Vec::new();
5719 assert!(!convert_margin_user_data_events(&frame, &mut buf));
5720 assert_eq!(
5721 buf.len(),
5722 2,
5723 "a fill must produce the execution and the order snapshot, got: {buf:?}"
5724 );
5725 assert_eq!(buf[0].exchange, ExchangeId::BinanceMargin);
5726 match &buf[0].kind {
5727 AccountEventKind::Trade(t) => {
5728 assert_eq!(t.instrument.name().as_str(), "BTCUSDT");
5729 assert_eq!(t.side, Side::Buy);
5730 assert_eq!(t.price, Decimal::from(48_000));
5731 assert_eq!(t.quantity, Decimal::new(5, 1));
5732 assert_eq!(t.fees.asset.name().as_str(), "BNB"); assert_eq!(t.fees.fees, Decimal::new(1, 3));
5734 assert_eq!(t.client_order_id, Some(ClientOrderId::new("cid-1")));
5735 }
5736 other => panic!("expected Trade, got {other:?}"),
5737 }
5738 assert_eq!(buf[1].exchange, ExchangeId::BinanceMargin);
5739 match &buf[1].kind {
5740 AccountEventKind::OrderSnapshot(snap) => {
5741 assert_eq!(
5742 snap.0.quantity,
5743 Decimal::from(2),
5744 "the order's total quantity"
5745 );
5746 match &snap.0.state {
5747 OrderState::Active(ActiveOrderState::Open(open)) => {
5748 assert_eq!(open.id.assigned().map(|id| id.0.as_str()), Some("12345"));
5749 assert_eq!(
5750 open.filled_quantity,
5751 Decimal::new(5, 1),
5752 "the snapshot must carry the order's cumulative filled quantity (z)"
5753 );
5754 assert_eq!(
5755 open.time_exchange.timestamp_millis(),
5756 1_700_000_000_000,
5757 "stamped with the execution's transaction time (T), which is what \
5758 gives the engine's staleness gate an ordering key"
5759 );
5760 }
5761 other => panic!("expected an Open order, got {other:?}"),
5762 }
5763 }
5764 other => panic!("expected OrderSnapshot, got {other:?}"),
5765 }
5766 }
5767
5768 #[test]
5772 fn margin_trade_for_an_order_no_longer_working_emits_no_snapshot() {
5773 for status in ["CANCELED", "EXPIRED", "REJECTED", "PENDING_CANCEL", "NEW"] {
5774 let frame = push(serde_json::json!({
5775 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
5776 "x": "TRADE", "X": status, "i": 12_345_i64, "c": "cid-1",
5777 "t": 99_i64, "l": "0.5", "L": "48000", "z": "0.5",
5778 "p": "48000", "q": "2", "f": "GTC", "T": 1_700_000_000_000_i64,
5779 }));
5780 let mut buf = Vec::new();
5781 assert!(!convert_margin_user_data_events(&frame, &mut buf));
5782 assert_eq!(
5783 buf.len(),
5784 1,
5785 "status {status} must produce the execution alone, got: {buf:?}"
5786 );
5787 assert!(
5788 matches!(buf[0].kind, AccountEventKind::Trade(_)),
5789 "the execution must still be reported for status {status}"
5790 );
5791 }
5792 }
5793
5794 #[test]
5795 fn margin_ws_new_report_maps_to_active_order_snapshot() {
5796 let frame = push(serde_json::json!({
5797 "e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "LIMIT",
5798 "x": "NEW", "X": "NEW", "i": 7_i64, "c": "cid-2",
5799 "p": "48000", "q": "1", "f": "GTC", "z": "0", "T": 1_700_000_000_000_i64,
5800 }));
5801 let mut buf = Vec::new();
5802 assert!(!convert_margin_user_data_events(&frame, &mut buf));
5803 assert_eq!(buf.len(), 1);
5804 match &buf[0].kind {
5805 AccountEventKind::OrderSnapshot(snap) => {
5806 assert_eq!(snap.0.side, Side::Sell);
5807 assert_eq!(snap.0.price, Some(Decimal::from(48_000)));
5808 assert_eq!(snap.0.quantity, Decimal::from(1));
5809 assert!(
5810 matches!(snap.0.state, OrderState::Active(_)),
5811 "NEW should be an active (resting) order"
5812 );
5813 }
5814 other => panic!("expected OrderSnapshot, got {other:?}"),
5815 }
5816 }
5817
5818 #[test]
5819 fn margin_ws_canceled_report_maps_to_order_cancelled() {
5820 let frame = push(serde_json::json!({
5821 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
5822 "x": "CANCELED", "X": "CANCELED", "i": 8_i64, "c": "cid-3",
5823 "z": "0", "T": 1_700_000_000_000_i64,
5824 }));
5825 let mut buf = Vec::new();
5826 assert!(!convert_margin_user_data_events(&frame, &mut buf));
5827 assert_eq!(buf.len(), 1);
5828 match &buf[0].kind {
5829 AccountEventKind::OrderCancelled(resp) => assert!(resp.state.is_ok()),
5830 other => panic!("expected OrderCancelled, got {other:?}"),
5831 }
5832 }
5833
5834 #[test]
5837 fn margin_ws_trade_without_a_client_order_id_reports_none() {
5838 let frame = push(serde_json::json!({
5839 "e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "MARKET",
5840 "x": "TRADE", "X": "FILLED", "i": 12_345_i64,
5841 "t": 99_i64, "l": "1", "L": "48000", "z": "1",
5842 "q": "1", "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
5843 }));
5844 let mut buf = Vec::new();
5845 assert!(!convert_margin_user_data_events(&frame, &mut buf));
5846 let AccountEventKind::Trade(trade) = &buf[0].kind else {
5847 panic!("expected Trade, got {:?}", buf[0].kind);
5848 };
5849 assert_eq!(trade.order_id.0.as_str(), "12345");
5850 assert_eq!(trade.client_order_id, None);
5851 }
5852
5853 #[test]
5854 fn margin_ws_execution_report_missing_symbol_is_dropped() {
5855 let frame = push(serde_json::json!({
5857 "e": "executionReport", "S": "BUY", "x": "TRADE", "i": 1_i64, "t": 1_i64,
5858 "l": "1", "L": "100", "T": 1_700_000_000_000_i64,
5859 }));
5860 let mut buf = Vec::new();
5861 assert!(!convert_margin_user_data_events(&frame, &mut buf));
5862 assert!(buf.is_empty());
5863 }
5864
5865 #[test]
5866 fn margin_ws_outbound_account_position_maps_to_balance_stream_updates() {
5867 let frame = push(serde_json::json!({
5868 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
5869 "B": [
5870 { "a": "USDT", "f": "100.0", "l": "5.0" },
5871 { "a": "BTC", "f": "0.5", "l": "0" },
5872 ],
5873 }));
5874 let mut buf = Vec::new();
5875 assert!(!convert_margin_user_data_events(&frame, &mut buf));
5876 assert_eq!(buf.len(), 2);
5877 for ev in &buf {
5879 assert!(matches!(ev.kind, AccountEventKind::BalanceStreamUpdate(_)));
5880 }
5881 }
5882
5883 #[test]
5884 fn margin_ws_stream_terminated_signals_reconnect() {
5885 let frame = push(serde_json::json!({ "e": "eventStreamTerminated" }));
5886 let mut buf = Vec::new();
5887 assert!(
5888 convert_margin_user_data_events(&frame, &mut buf),
5889 "eventStreamTerminated must signal reconnect"
5890 );
5891 assert!(buf.is_empty());
5892 }
5893
5894 #[test]
5895 fn margin_ws_liability_change_is_observable_only() {
5896 let mut buf = Vec::new();
5899 assert!(!convert_margin_user_data_events(
5900 &push(documented_liability_change("USER_LIABILITY_CHANGE")),
5901 &mut buf
5902 ));
5903 assert!(buf.is_empty());
5904 }
5905
5906 fn only_notice(buf: &[UnindexedAccountEvent]) -> &AccountNotice<InstrumentNameExchange> {
5908 match buf {
5909 [
5910 UnindexedAccountEvent {
5911 exchange: ExchangeId::BinanceMargin,
5912 kind: AccountEventKind::Notice(notice),
5913 },
5914 ] => notice,
5915 other => panic!("expected one BinanceMargin notice, got {other:?}"),
5916 }
5917 }
5918
5919 #[test]
5920 fn margin_ws_margin_level_change_is_forwarded_as_a_notice() {
5921 let mut buf = Vec::new();
5922 assert!(!convert_margin_user_data_events(
5923 &push(documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE")),
5924 &mut buf
5925 ));
5926 assert_eq!(
5927 only_notice(&buf),
5928 &AccountNotice::new(
5929 NoticeKind::MarginCall,
5930 None,
5931 Utc.timestamp_millis_opt(1_573_200_697_110).unwrap(),
5932 Some(Decimal::new(125, 2)),
5933 SmolStr::new("MARGIN_CALL"),
5934 ),
5935 "cross names no instrument"
5936 );
5937 }
5938
5939 #[test]
5940 fn margin_level_statuses_map_to_notice_kinds() {
5941 for (status, kind) in [
5942 ("MARGIN_CALL", NoticeKind::MarginCall),
5943 ("PRE_LIQUIDATION", NoticeKind::LiquidationWarning),
5944 ("FORCE_LIQUIDATION", NoticeKind::Liquidation),
5945 ("NORMAL", NoticeKind::MarginRestored),
5946 ("EXCESSIVE", NoticeKind::MarginRestored),
5947 ("SOME_FUTURE_STATUS", NoticeKind::Other),
5948 ] {
5949 let mut buf = Vec::new();
5950 let frame = push(serde_json::json!({
5951 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "1.5",
5952 "s": status,
5953 }));
5954 convert_margin_user_data_events(&frame, &mut buf);
5955 let notice = only_notice(&buf);
5956 assert_eq!(notice.kind, kind, "{status}");
5957 assert_eq!(notice.status, status, "the raw status is kept");
5958 }
5959 }
5960
5961 #[test]
5962 fn a_margin_level_change_missing_parts_is_still_delivered_where_it_can_be() {
5963 let mut buf = Vec::new();
5965 convert_margin_user_data_events(
5966 &push(serde_json::json!({
5967 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "n/a",
5968 "s": "FORCE_LIQUIDATION",
5969 })),
5970 &mut buf,
5971 );
5972 let notice = only_notice(&buf);
5973 assert_eq!(notice.kind, NoticeKind::Liquidation);
5974 assert_eq!(notice.margin_level, None);
5975
5976 let before = Utc::now();
5978 let mut buf = Vec::new();
5979 convert_margin_user_data_events(
5980 &push(serde_json::json!({ "e": "MARGIN_LEVEL_STATUS_CHANGE", "s": "MARGIN_CALL" })),
5981 &mut buf,
5982 );
5983 assert!(only_notice(&buf).time_exchange >= before);
5984
5985 let mut buf = Vec::new();
5987 convert_margin_user_data_events(
5988 &push(serde_json::json!({
5989 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "1.5",
5990 })),
5991 &mut buf,
5992 );
5993 assert!(buf.is_empty());
5994 }
5995
5996 fn documented_liability_change(e: &str) -> serde_json::Value {
5998 serde_json::json!({
5999 "e": e, "E": 1_573_200_697_110_i64, "a": "BTC", "t": "BORROW",
6000 "p": "1.03453430", "i": "0",
6001 })
6002 }
6003
6004 fn documented_margin_level_change(e: &str) -> serde_json::Value {
6006 serde_json::json!({
6007 "e": e, "E": 1_573_200_697_110_i64, "l": "1.25", "s": "MARGIN_CALL",
6008 })
6009 }
6010
6011 #[test]
6012 fn margin_ws_margin_specific_events_match_their_documented_names() {
6013 for (e, frame, expected) in [
6018 (
6019 "USER_LIABILITY_CHANGE",
6020 documented_liability_change("USER_LIABILITY_CHANGE"),
6021 "USER_LIABILITY_CHANGE (observable",
6022 ),
6023 (
6024 "MARGIN_LEVEL_STATUS_CHANGE",
6025 documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE"),
6026 "MARGIN_LEVEL_STATUS_CHANGE received",
6027 ),
6028 (
6030 "userLiabilityChange",
6031 documented_liability_change("userLiabilityChange"),
6032 "USER_LIABILITY_CHANGE (observable",
6033 ),
6034 (
6035 "marginLevelStatusChange",
6036 documented_margin_level_change("marginLevelStatusChange"),
6037 "MARGIN_LEVEL_STATUS_CHANGE received",
6038 ),
6039 ] {
6040 let mut buf = Vec::new();
6041 let (terminated, messages) =
6042 log_messages(|| convert_margin_user_data_events(&push(frame), &mut buf));
6043 assert!(!terminated, "{e}");
6044 assert_eq!(
6046 buf.len(),
6047 usize::from(expected.starts_with("MARGIN")),
6048 "{e}"
6049 );
6050 assert!(
6051 messages.iter().any(|m| m.contains(expected)),
6052 "{e}: expected its own arm's log, got {messages:?}"
6053 );
6054 assert!(
6055 !messages.iter().any(|m| m.contains("undeserializable")
6056 || m.contains("unhandled")
6057 || m.contains("does not handle")),
6058 "{e}: {messages:?}"
6059 );
6060 }
6061 }
6062
6063 fn log_messages<T>(f: impl FnOnce() -> T) -> (T, Vec<String>) {
6067 use std::sync::{Arc, Mutex};
6068 use tracing_subscriber::layer::SubscriberExt;
6069
6070 #[derive(Default)]
6071 struct Messages(Arc<Mutex<Vec<String>>>);
6072
6073 struct Message<'a>(&'a mut String);
6074
6075 impl tracing::field::Visit for Message<'_> {
6076 fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
6077 if field.name() == "message" {
6078 *self.0 = format!("{value:?}");
6079 }
6080 }
6081 }
6082
6083 impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Messages {
6084 fn on_event(
6085 &self,
6086 event: &tracing::Event<'_>,
6087 _: tracing_subscriber::layer::Context<'_, S>,
6088 ) {
6089 let mut message = String::new();
6090 event.record(&mut Message(&mut message));
6091 self.0.lock().unwrap().push(message);
6092 }
6093 }
6094
6095 let layer = Messages::default();
6096 let messages = Arc::clone(&layer.0);
6097 let value =
6099 tracing::subscriber::with_default(tracing_subscriber::registry().with(layer), f);
6100 let messages = messages.lock().unwrap().clone();
6101 (value, messages)
6102 }
6103
6104 #[test]
6105 fn margin_ws_unhandled_event_types_are_ignored_without_teardown() {
6106 for (e, expected) in [
6109 ("listStatus", "ignoring listStatus"),
6110 ("someFutureEvent", "of a type this client does not handle"),
6112 ] {
6113 let mut buf = Vec::new();
6114 let frame = push(serde_json::json!({ "e": e, "E": 1_700_000_000_000_i64 }));
6115 let (terminated, messages) =
6116 log_messages(|| convert_margin_user_data_events(&frame, &mut buf));
6117 assert!(!terminated, "{e}");
6118 assert!(buf.is_empty(), "{e}");
6119 assert!(
6120 messages.iter().any(|m| m.contains(expected)),
6121 "{e}: {messages:?}"
6122 );
6123 }
6124 }
6125
6126 #[test]
6129 fn margin_trade_event_dedup_by_key() {
6130 let frame = push(serde_json::json!({
6133 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6134 "x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 4_242_i64,
6135 "l": "1", "L": "100", "z": "1", "p": "100", "q": "1", "f": "GTC",
6136 "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
6137 }));
6138 let mut buf = Vec::new();
6139 convert_margin_user_data_events(&frame, &mut buf);
6140 let event = buf.first().expect("a Trade event");
6143 assert!(
6144 matches!(event.kind, AccountEventKind::Trade(_)),
6145 "the execution is the first of the two events a fill produces"
6146 );
6147
6148 let cache = new_dedup_cache();
6149 let first = dedup_key_from_event(event).expect("Trade events have a dedup key");
6151 assert!(!is_duplicate(&cache, first), "first sighting is fresh");
6152 let second = dedup_key_from_event(event).expect("Trade events have a dedup key");
6153 assert!(
6154 is_duplicate(&cache, second),
6155 "second sighting is a duplicate"
6156 );
6157 }
6158
6159 fn drive_ws_pipeline(frames: &[&str]) -> Vec<UnindexedAccountEvent> {
6167 let cache = new_dedup_cache();
6168 let mut delivered = Vec::new();
6169 let mut buf = Vec::new();
6170 for frame in frames {
6171 let _ = convert_margin_user_data_events(frame, &mut buf);
6172 for ev in buf.drain(..) {
6173 if let Some(key) = dedup_key_from_event(&ev)
6174 && is_duplicate(&cache, key)
6175 {
6176 continue;
6177 }
6178 delivered.push(ev);
6179 }
6180 }
6181 delivered
6182 }
6183
6184 fn ws_new(cumulative_filled: &str) -> String {
6185 push(serde_json::json!({
6186 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6187 "x": "NEW", "X": "NEW", "i": 12_345_i64, "c": "client-1",
6188 "q": "2", "p": "100", "f": "GTC", "z": cumulative_filled,
6189 "T": 1_700_000_000_000_i64,
6190 }))
6191 }
6192
6193 fn ws_partial_fill(cumulative_filled: &str) -> String {
6194 push(serde_json::json!({
6195 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6196 "x": "TRADE", "X": "PARTIALLY_FILLED", "i": 12_345_i64, "c": "client-1",
6197 "q": "2", "p": "100", "f": "GTC", "z": cumulative_filled,
6198 "l": "1", "L": "100", "t": 555_i64, "n": "0.1", "N": "USDT",
6199 "T": 1_700_000_001_000_i64,
6200 }))
6201 }
6202
6203 fn open_filled_quantities(events: &[UnindexedAccountEvent]) -> Vec<Decimal> {
6204 events
6205 .iter()
6206 .filter_map(|ev| match &ev.kind {
6207 AccountEventKind::OrderSnapshot(snap) => match &snap.0.state {
6208 OrderState::Active(ActiveOrderState::Open(open)) => Some(open.filled_quantity),
6209 _ => None,
6210 },
6211 _ => None,
6212 })
6213 .collect()
6214 }
6215
6216 #[test]
6224 fn margin_ws_fill_snapshot_survives_the_dedup_gate() {
6225 let delivered = drive_ws_pipeline(&[&ws_new("0"), &ws_partial_fill("1")]);
6226
6227 assert_eq!(
6228 delivered.len(),
6229 3,
6230 "expected the ack snapshot, then the fill's execution and its snapshot, got: {delivered:?}"
6231 );
6232 assert_eq!(
6233 open_filled_quantities(&delivered),
6234 vec![Decimal::ZERO, Decimal::ONE],
6235 "the fill's snapshot must reach the consumer carrying the advanced filled quantity"
6236 );
6237 }
6238
6239 #[test]
6242 fn margin_ws_replayed_frame_is_still_deduplicated() {
6243 let delivered = drive_ws_pipeline(&[
6244 &ws_new("0"),
6245 &ws_partial_fill("1"),
6246 &ws_new("0"),
6247 &ws_partial_fill("1"),
6248 ]);
6249
6250 assert_eq!(
6251 delivered.len(),
6252 3,
6253 "re-delivering both frames must add nothing, got: {delivered:?}"
6254 );
6255 }
6256
6257 #[test]
6263 fn margin_ws_an_ack_and_its_fill_reporting_one_state_deliver_that_state_once() {
6264 let delivered = drive_ws_pipeline(&[&ws_new("1"), &ws_partial_fill("1")]);
6265
6266 assert_eq!(
6267 delivered.len(),
6268 2,
6269 "the ack snapshot and the execution, with the fill's duplicate snapshot collapsed, \
6270 got: {delivered:?}"
6271 );
6272 assert!(
6273 delivered
6274 .iter()
6275 .any(|ev| matches!(ev.kind, AccountEventKind::Trade(_))),
6276 "the execution must reach the consumer regardless"
6277 );
6278 assert_eq!(
6279 open_filled_quantities(&delivered),
6280 vec![Decimal::ONE],
6281 "one snapshot, carrying the state both reports agree on"
6282 );
6283 }
6284
6285 fn push_with_sub(subscription_id: i64, event: serde_json::Value) -> String {
6290 serde_json::json!({ "subscriptionId": subscription_id, "event": event }).to_string()
6291 }
6292
6293 fn isolated_handler(
6296 sub_map: Arc<Mutex<HashMap<i64, InstrumentNameExchange>>>,
6297 base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
6298 ) -> impl FnMut(OutboundAccountPosition, Option<i64>, &mut Vec<UnindexedAccountEvent>) {
6299 move |position, subscription_id, buf| {
6300 route_isolated_account_position(position, subscription_id, &sub_map, &base_quote, buf);
6301 }
6302 }
6303
6304 fn btcusdt() -> InstrumentNameExchange {
6305 InstrumentNameExchange::new("BTCUSDT")
6306 }
6307
6308 fn token(expiration_time_ms: i64) -> UserListenToken {
6309 UserListenToken {
6310 token: "t".to_string(),
6311 expiration_time_ms,
6312 }
6313 }
6314
6315 #[test]
6316 fn listen_token_query_cross_vs_isolated() {
6317 assert!(build_listen_token_query(None).is_empty());
6319
6320 let q = build_listen_token_query(Some(&btcusdt()));
6322 assert_eq!(q.get("isIsolated").and_then(|v| v.as_str()), Some("TRUE"));
6323 assert_eq!(q.get("symbol").and_then(|v| v.as_str()), Some("BTCUSDT"));
6324 assert_eq!(q.len(), 2);
6325 }
6326
6327 #[test]
6328 fn earliest_token_expiry_picks_min() {
6329 let tokens = vec![
6330 (btcusdt(), token(3_000)),
6331 (InstrumentNameExchange::new("ETHUSDT"), token(1_000)),
6332 (InstrumentNameExchange::new("BNBUSDT"), token(2_000)),
6333 ];
6334 assert_eq!(earliest_token_expiry_ms(&tokens), 1_000);
6335 assert_eq!(earliest_token_expiry_ms(&[]), i64::MAX);
6337 }
6338
6339 #[test]
6340 fn base_quote_map_extracts_base_and_quote() {
6341 let entries = vec![
6342 iso_entry(
6343 "BTCUSDT",
6344 iso_base("BTC", "0", "0", "0", "0"),
6345 iso_quote("USDT", "0", "0", "0", "0"),
6346 None,
6347 None,
6348 None,
6349 ),
6350 iso_entry(
6351 "ETHBTC",
6352 iso_base("ETH", "0", "0", "0", "0"),
6353 iso_quote("BTC", "0", "0", "0", "0"),
6354 None,
6355 None,
6356 None,
6357 ),
6358 ];
6359 let map = build_base_quote_map(&entries);
6360 assert_eq!(
6361 map.get(&btcusdt())
6362 .map(|(b, q)| (b.name().as_str(), q.name().as_str())),
6363 Some(("BTC", "USDT"))
6364 );
6365 assert_eq!(
6368 map.get(&InstrumentNameExchange::new("ETHBTC"))
6369 .map(|(b, q)| (b.name().as_str(), q.name().as_str())),
6370 Some(("ETH", "BTC"))
6371 );
6372 }
6373
6374 #[test]
6375 fn isolated_margin_level_change_names_its_subscriptions_pair_or_none() {
6376 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6377 let notice_on = |subscription_id: i64| {
6378 let mut buf = Vec::new();
6379 let frame = push_with_sub(
6380 subscription_id,
6381 documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE"),
6382 );
6383 assert!(!convert_margin_user_data_events_with(
6384 &frame,
6385 &mut buf,
6386 &mut cross_account_position_handler,
6387 &mut |id| isolated_notice_instrument(id, &sub_map),
6388 ));
6389 only_notice(&buf).instrument.clone()
6390 };
6391
6392 assert_eq!(notice_on(7), Some(btcusdt()));
6393 assert_eq!(notice_on(99), None);
6395 assert_eq!(isolated_notice_instrument(None, &sub_map), None);
6397 }
6398
6399 #[test]
6400 fn the_two_streams_key_one_notice_alike() {
6401 use crate::client::binance::margin_risk::{RiskFrame, convert_risk_frame};
6403
6404 let change = documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE");
6405 let mut from_user_data = Vec::new();
6406 convert_margin_user_data_events(&push(change.clone()), &mut from_user_data);
6407 let mut from_risk = Vec::new();
6408 let risk_frame = serde_json::json!({ "stream": "listen-key", "data": change });
6409 assert_eq!(
6410 convert_risk_frame(&risk_frame.to_string(), "listen-key", &mut from_risk),
6411 RiskFrame::Continue
6412 );
6413
6414 let dedup = new_dedup_cache();
6415 let key = |events: &[UnindexedAccountEvent]| dedup_key_from_event(&events[0]).unwrap();
6416 assert!(!is_duplicate(&dedup, key(&from_user_data)));
6417 assert!(
6418 is_duplicate(&dedup, key(&from_risk)),
6419 "the risk data stream's copy is dropped"
6420 );
6421 }
6422
6423 #[test]
6424 fn a_notice_from_both_streams_is_delivered_once() {
6425 let dedup = new_dedup_cache();
6429 let notice = |time_ms: i64| {
6430 let mut buf = Vec::new();
6431 convert_margin_user_data_events(
6432 &push(serde_json::json!({
6433 "e": "MARGIN_LEVEL_STATUS_CHANGE", "E": time_ms, "l": "1.1",
6434 "s": "MARGIN_CALL",
6435 })),
6436 &mut buf,
6437 );
6438 buf.pop().unwrap()
6439 };
6440 let is_new = |event: &UnindexedAccountEvent| {
6441 !is_duplicate(&dedup, dedup_key_from_event(event).unwrap())
6442 };
6443
6444 assert!(is_new(¬ice(1_700_000_000_000)));
6445 assert!(!is_new(¬ice(1_700_000_000_000)), "second copy dropped");
6446 assert!(is_new(¬ice(1_700_000_060_000)), "a later notice is new");
6447 }
6448
6449 #[test]
6450 fn isolated_outbound_position_routes_to_instrument_balance_update() {
6451 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6452 let base_quote = Arc::new(HashMap::from([(
6453 btcusdt(),
6454 (
6455 AssetNameExchange::new("BTC"),
6456 AssetNameExchange::new("USDT"),
6457 ),
6458 )]));
6459 let mut handler = isolated_handler(sub_map, base_quote);
6460
6461 let frame = push_with_sub(
6463 7,
6464 serde_json::json!({
6465 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6466 "B": [
6467 { "a": "USDT", "f": "1000.0", "l": "50.0" },
6468 { "a": "BTC", "f": "0.5", "l": "0.1" },
6469 ],
6470 }),
6471 );
6472 let mut buf = Vec::new();
6473 assert!(!convert_margin_user_data_events_with(
6474 &frame,
6475 &mut buf,
6476 &mut handler,
6477 &mut cross_notice_instrument,
6478 ));
6479 assert_eq!(buf.len(), 1, "one InstrumentBalanceUpdate for the pair");
6480 match &buf[0].kind {
6481 AccountEventKind::InstrumentBalanceUpdate(ibu) => {
6482 assert_eq!(ibu.instrument.name().as_str(), "BTCUSDT");
6483 assert_eq!(ibu.base.asset.name().as_str(), "BTC");
6484 assert_eq!(ibu.base.update.free, Decimal::new(5, 1));
6485 assert_eq!(ibu.base.update.locked, Decimal::new(1, 1));
6486 assert_eq!(ibu.quote.asset.name().as_str(), "USDT");
6487 assert_eq!(ibu.quote.update.free, Decimal::from(1000));
6488 assert_eq!(ibu.quote.update.locked, Decimal::from(50));
6489 }
6490 other => panic!("expected InstrumentBalanceUpdate, got {other:?}"),
6491 }
6492 assert!(!matches!(
6494 buf[0].kind,
6495 AccountEventKind::BalanceStreamUpdate(_)
6496 ));
6497 }
6498
6499 #[test]
6500 fn isolated_outbound_position_unknown_subscription_dropped() {
6501 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6503 let base_quote = Arc::new(HashMap::from([(
6504 btcusdt(),
6505 (
6506 AssetNameExchange::new("BTC"),
6507 AssetNameExchange::new("USDT"),
6508 ),
6509 )]));
6510 let mut handler = isolated_handler(sub_map, base_quote);
6511
6512 let frame = push_with_sub(
6513 99,
6514 serde_json::json!({
6515 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6516 "B": [{ "a": "BTC", "f": "1", "l": "0" }, { "a": "USDT", "f": "1", "l": "0" }],
6517 }),
6518 );
6519 let mut buf = Vec::new();
6520 assert!(!convert_margin_user_data_events_with(
6521 &frame,
6522 &mut buf,
6523 &mut handler,
6524 &mut cross_notice_instrument,
6525 ));
6526 assert!(buf.is_empty(), "unmapped subscriptionId frame is dropped");
6527 }
6528
6529 #[test]
6530 fn isolated_outbound_position_missing_side_dropped() {
6531 let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
6534 let base_quote = Arc::new(HashMap::from([(
6535 btcusdt(),
6536 (
6537 AssetNameExchange::new("BTC"),
6538 AssetNameExchange::new("USDT"),
6539 ),
6540 )]));
6541 let mut handler = isolated_handler(sub_map, base_quote);
6542
6543 let frame = push_with_sub(
6544 7,
6545 serde_json::json!({
6546 "e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
6547 "B": [{ "a": "BTC", "f": "0.5", "l": "0" }],
6548 }),
6549 );
6550 let mut buf = Vec::new();
6551 assert!(!convert_margin_user_data_events_with(
6552 &frame,
6553 &mut buf,
6554 &mut handler,
6555 &mut cross_notice_instrument,
6556 ));
6557 assert!(buf.is_empty(), "missing quote side → frame dropped");
6558 }
6559
6560 #[test]
6561 fn isolated_execution_report_routes_by_inner_symbol_independent_of_map() {
6562 let sub_map = Arc::new(Mutex::new(HashMap::<i64, InstrumentNameExchange>::new()));
6566 let base_quote = Arc::new(HashMap::new());
6567 let mut handler = isolated_handler(sub_map, base_quote);
6568
6569 let frame = push_with_sub(
6570 42,
6571 serde_json::json!({
6572 "e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
6573 "x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 5_i64,
6574 "l": "1", "L": "100", "z": "1", "p": "100", "q": "1", "f": "GTC",
6575 "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
6576 }),
6577 );
6578 let mut buf = Vec::new();
6579 assert!(!convert_margin_user_data_events_with(
6580 &frame,
6581 &mut buf,
6582 &mut handler,
6583 &mut cross_notice_instrument,
6584 ));
6585 assert_eq!(buf.len(), 2, "the execution and its order snapshot");
6586 match &buf[0].kind {
6587 AccountEventKind::Trade(t) => assert_eq!(t.instrument.name().as_str(), "BTCUSDT"),
6588 other => panic!("expected Trade, got {other:?}"),
6589 }
6590 match &buf[1].kind {
6591 AccountEventKind::OrderSnapshot(snap) => {
6592 assert_eq!(snap.0.key.instrument.name().as_str(), "BTCUSDT");
6593 }
6594 other => panic!("expected OrderSnapshot, got {other:?}"),
6595 }
6596 }
6597
6598 struct MarginMyTradesVenue {
6609 trades: Vec<(i64, i64, i64)>,
6611 end_exclusive: bool,
6612 }
6613
6614 impl wiremock::Respond for MarginMyTradesVenue {
6615 fn respond(&self, request: &wiremock::Request) -> wiremock::ResponseTemplate {
6616 const DAY_MS: i64 = 24 * 60 * 60 * 1000;
6617 let query: std::collections::HashMap<String, String> =
6618 request.url.query_pairs().into_owned().collect();
6619 let param = |name: &str| query.get(name).map(|v| v.parse::<i64>().unwrap());
6620 let limit = usize::try_from(param("limit").unwrap_or(500)).unwrap();
6621 let order = param("orderId");
6622
6623 let mut selected: Vec<&(i64, i64, i64)> = self
6624 .trades
6625 .iter()
6626 .filter(|(_, _, order_id)| order.is_none_or(|order| *order_id == order))
6627 .collect();
6628 if let Some(from_id) = param("fromId") {
6629 selected.retain(|(id, _, _)| *id >= from_id);
6630 } else if order.is_none() {
6631 let start = param("startTime").expect("a query by time sends startTime");
6632 let end = match param("endTime") {
6633 Some(end) if end - start >= DAY_MS => {
6634 return wiremock::ResponseTemplate::new(400).set_body_json(
6635 serde_json::json!({"code": -1127, "msg": "More than 24 hours between startTime and endTime."}),
6636 );
6637 }
6638 Some(end) if self.end_exclusive => end - 1,
6639 Some(end) => end,
6640 None => start + DAY_MS - 1,
6641 };
6642 selected.retain(|(_, time, _)| (start..=end).contains(time));
6643 }
6644
6645 let page: Vec<serde_json::Value> = selected
6646 .into_iter()
6647 .take(limit)
6648 .map(|(id, time, order_id)| {
6649 serde_json::json!({
6650 "id": id, "symbol": "BTCUSDT", "time": time, "orderId": order_id,
6651 "price": "50000", "qty": "0.01", "commission": "0",
6652 "commissionAsset": "USDT", "isBuyer": true, "isMaker": false,
6653 "isBestMatch": true, "isIsolated": true,
6654 })
6655 })
6656 .collect();
6657 wiremock::ResponseTemplate::new(200).set_body_json(page)
6658 }
6659 }
6660
6661 fn since(start: i64) -> MyTradesFrom {
6663 MyTradesFrom::Span {
6664 start,
6665 end: i64::MAX,
6666 }
6667 }
6668
6669 async fn read_margin_trades(
6672 venue: MarginMyTradesVenue,
6673 from: MyTradesFrom,
6674 ) -> (Vec<i64>, Vec<std::collections::HashMap<String, String>>) {
6675 let server = wiremock::MockServer::start().await;
6676 wiremock::Mock::given(wiremock::matchers::method("GET"))
6677 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
6678 .respond_with(venue)
6679 .mount(&server)
6680 .await;
6681 let rest = Arc::new(MarginTradingRestApi::from_config(
6682 ConfigurationRestApi::builder()
6683 .api_key("key")
6684 .api_secret("secret")
6685 .base_path(server.uri())
6686 .build()
6687 .unwrap(),
6688 ));
6689
6690 let read = paginate_margin_my_trades(
6691 &rest,
6692 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
6693 &InstrumentNameExchange::new("BTCUSDT"),
6694 from,
6695 true,
6696 RequestKind::Query,
6697 )
6698 .await
6699 .unwrap();
6700
6701 let requests = server
6702 .received_requests()
6703 .await
6704 .unwrap()
6705 .iter()
6706 .map(|request| request.url.query_pairs().into_owned().collect())
6707 .collect();
6708 (read.into_iter().map(|t| t.id.unwrap()).collect(), requests)
6709 }
6710
6711 #[tokio::test]
6714 async fn recovered_margin_fills_carry_their_orders_client_order_ids() {
6715 recovered_margin_fills_carry_client_order_ids(true).await;
6716 recovered_margin_fills_carry_client_order_ids(false).await;
6717 }
6718
6719 async fn recovered_margin_fills_carry_client_order_ids(is_isolated: bool) {
6720 let disconnect = Utc::now() - chrono::Duration::minutes(10);
6721 let reconnect = Utc::now() - chrono::Duration::minutes(5);
6722 let fill_time = (disconnect + chrono::Duration::minutes(1)).timestamp_millis();
6723 let server = wiremock::MockServer::start().await;
6724 wiremock::Mock::given(wiremock::matchers::method("GET"))
6725 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
6726 .respond_with(MarginMyTradesVenue {
6727 trades: vec![(1, fill_time, 1), (2, fill_time, 2)],
6728 end_exclusive: false,
6729 })
6730 .mount(&server)
6731 .await;
6732 wiremock::Mock::given(wiremock::matchers::method("GET"))
6733 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
6734 .and(wiremock::matchers::query_param("orderId", "1"))
6735 .and(wiremock::matchers::query_param(
6736 "isIsolated",
6737 if is_isolated { "TRUE" } else { "FALSE" },
6738 ))
6739 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
6740 serde_json::json!({"symbol": "BTCUSDT", "orderId": 1, "clientOrderId": "repay-1"}),
6741 ))
6742 .mount(&server)
6743 .await;
6744 wiremock::Mock::given(wiremock::matchers::method("GET"))
6745 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
6746 .and(wiremock::matchers::query_param("orderId", "2"))
6747 .respond_with(
6748 wiremock::ResponseTemplate::new(400).set_body_json(
6749 serde_json::json!({"code": -2013, "msg": "Order does not exist."}),
6750 ),
6751 )
6752 .mount(&server)
6753 .await;
6754 let rest = Arc::new(MarginTradingRestApi::from_config(
6755 ConfigurationRestApi::builder()
6756 .api_key("key")
6757 .api_secret("secret")
6758 .base_path(server.uri())
6759 .build()
6760 .unwrap(),
6761 ));
6762 let (tx, mut rx) = mpsc::unbounded_channel();
6763 let mut unrecovered = UnrecoveredFills::default();
6764 unrecovered.open(
6765 &[InstrumentNameExchange::new("BTCUSDT")],
6766 disconnect,
6767 reconnect,
6768 );
6769
6770 recover_margin_fills(
6771 &rest,
6772 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
6773 &mut unrecovered,
6774 &tx,
6775 &new_dedup_cache(),
6776 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
6777 is_isolated,
6778 )
6779 .await;
6780
6781 let mut cids: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
6782 .filter_map(|event| match event.kind {
6783 AccountEventKind::Trade(trade) => Some((
6784 trade.id.0.to_string(),
6785 trade.client_order_id,
6786 trade.order_filled_quantity.is_some(),
6787 )),
6788 _ => None,
6789 })
6790 .collect();
6791 cids.sort();
6792 assert_eq!(
6794 cids,
6795 [
6796 ("1".to_string(), Some(ClientOrderId::new("repay-1")), true),
6797 ("2".to_string(), None, true),
6798 ],
6799 "is_isolated: {is_isolated}"
6800 );
6801 let order_reads = server
6802 .received_requests()
6803 .await
6804 .unwrap()
6805 .iter()
6806 .filter(|request| request.url.path() == "/sapi/v1/margin/order")
6807 .count();
6808 assert_eq!(order_reads, 2, "one read per order");
6809 }
6810
6811 fn venue(trades: Vec<(i64, i64)>) -> MarginMyTradesVenue {
6813 MarginMyTradesVenue {
6814 trades: trades
6815 .into_iter()
6816 .map(|(id, time)| (id, time, id))
6817 .collect(),
6818 end_exclusive: false,
6819 }
6820 }
6821
6822 const HOUR_MS: i64 = 60 * 60 * 1000;
6823
6824 #[tokio::test]
6827 async fn margin_trades_read_a_lone_fill_beyond_the_first_day() {
6828 let now = Utc::now().timestamp_millis();
6829 let (ids, _) = read_margin_trades(
6830 venue(vec![(7, now - 72 * HOUR_MS)]),
6831 since(now - 120 * HOUR_MS),
6832 )
6833 .await;
6834
6835 assert_eq!(ids, vec![7]);
6836 }
6837
6838 #[tokio::test]
6841 async fn margin_trades_read_fills_across_several_windows_once_each() {
6842 let now = Utc::now().timestamp_millis();
6843 let trades = vec![
6844 (1, now - 119 * HOUR_MS),
6845 (2, now - 72 * HOUR_MS),
6846 (3, now - HOUR_MS),
6847 ];
6848 let (ids, _) = read_margin_trades(venue(trades), since(now - 144 * HOUR_MS)).await;
6849
6850 assert_eq!(ids, vec![1, 2, 3]);
6851 }
6852
6853 #[tokio::test]
6856 async fn margin_trades_read_a_trade_on_a_window_boundary_once() {
6857 for end_exclusive in [false, true] {
6858 let now = Utc::now().timestamp_millis();
6859 let from = now - 48 * HOUR_MS;
6860 let venue = MarginMyTradesVenue {
6861 end_exclusive,
6862 ..venue(vec![(5, from + MARGIN_MY_TRADES_WINDOW_MS)])
6863 };
6864 let (ids, _) = read_margin_trades(venue, since(from)).await;
6865
6866 assert_eq!(ids, vec![5], "end_exclusive: {end_exclusive}");
6867 }
6868 }
6869
6870 #[tokio::test]
6872 async fn margin_trades_page_on_by_id_past_a_full_window() {
6873 let now = Utc::now().timestamp_millis();
6874 let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
6875 let trades = (1..=count)
6876 .map(|id| (id, now - 48 * HOUR_MS + id))
6877 .collect();
6878 let (ids, _) = read_margin_trades(venue(trades), since(now - 50 * HOUR_MS)).await;
6879
6880 assert_eq!(ids, (1..=count).collect::<Vec<_>>());
6881 }
6882
6883 #[tokio::test]
6886 async fn margin_trades_page_on_past_a_full_final_window() {
6887 let now = Utc::now().timestamp_millis();
6888 let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
6889 let trades = (1..=count).map(|id| (id, now - HOUR_MS + id)).collect();
6890 let (ids, requests) = read_margin_trades(venue(trades), since(now - 2 * HOUR_MS)).await;
6891
6892 assert_eq!(ids, (1..=count).collect::<Vec<_>>());
6893 assert_eq!(requests.len(), 2);
6894 }
6895
6896 #[tokio::test]
6899 async fn margin_trades_quiet_lookback_sends_one_request_per_window() {
6900 let now = Utc::now().timestamp_millis();
6901 let (ids, requests) =
6902 read_margin_trades(venue(Vec::new()), since(now - 72 * HOUR_MS)).await;
6903
6904 assert!(ids.is_empty());
6905 assert_eq!(requests.len(), 4);
6907 }
6908
6909 #[tokio::test]
6912 async fn margin_trades_read_a_short_gap_in_one_request() {
6913 let now = Utc::now().timestamp_millis();
6914 let (ids, requests) =
6915 read_margin_trades(venue(vec![(9, now - HOUR_MS)]), since(now - 2 * HOUR_MS)).await;
6916
6917 assert_eq!(ids, vec![9]);
6918 assert_eq!(requests.len(), 1);
6919 }
6920
6921 #[tokio::test]
6923 async fn margin_trades_from_the_future_send_no_request() {
6924 let now = Utc::now().timestamp_millis();
6925 let (ids, requests) = read_margin_trades(venue(Vec::new()), since(now + HOUR_MS)).await;
6926
6927 assert!(ids.is_empty());
6928 assert!(requests.is_empty());
6929 }
6930
6931 #[tokio::test]
6934 async fn margin_trades_by_order_page_by_id_without_time_bounds() {
6935 let now = Utc::now().timestamp_millis();
6936 let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
6937 let trades = (1..=2 * count)
6939 .map(|id| {
6940 (
6941 id,
6942 now - 48 * HOUR_MS + id,
6943 if id % 2 == 0 { 42 } else { 7 },
6944 )
6945 })
6946 .collect();
6947 let venue = MarginMyTradesVenue {
6948 trades,
6949 end_exclusive: false,
6950 };
6951 let (ids, requests) = read_margin_trades(venue, MyTradesFrom::Order(42)).await;
6952
6953 assert_eq!(ids, (1..=count).map(|n| 2 * n).collect::<Vec<_>>());
6954 assert_eq!(requests.len(), 2);
6955 assert!(
6956 requests.iter().all(
6957 |query| query.get("orderId").map(String::as_str) == Some("42")
6958 && !query.contains_key("startTime")
6959 && !query.contains_key("endTime")
6960 )
6961 );
6962 }
6963
6964 async fn margin_trades_error(response: wiremock::ResponseTemplate) -> UnindexedClientError {
6967 let server = wiremock::MockServer::start().await;
6968 wiremock::Mock::given(wiremock::matchers::method("GET"))
6969 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
6970 .respond_with(response)
6971 .mount(&server)
6972 .await;
6973 let rest = Arc::new(MarginTradingRestApi::from_config(
6974 ConfigurationRestApi::builder()
6975 .api_key("key")
6976 .api_secret("secret")
6977 .base_path(server.uri())
6978 .build()
6979 .unwrap(),
6980 ));
6981
6982 paginate_margin_my_trades(
6983 &rest,
6984 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
6985 &InstrumentNameExchange::new("BTCUSDT"),
6986 since(Utc::now().timestamp_millis() - HOUR_MS),
6987 true,
6988 RequestKind::Query,
6989 )
6990 .await
6991 .unwrap_err()
6992 }
6993
6994 #[tokio::test]
6997 async fn margin_trades_report_a_venue_rejection_as_a_non_transient_request_rejection() {
6998 let err = margin_trades_error(wiremock::ResponseTemplate::new(400).set_body_json(
6999 serde_json::json!({"code": -1127, "msg": "More than 24 hours between startTime and endTime."}),
7000 ))
7001 .await;
7002
7003 let UnindexedClientError::Api(ApiError::RequestRejected(msg)) = &err else {
7004 panic!("expected RequestRejected, got {err:?}");
7005 };
7006 assert!(msg.contains("-1127"), "code missing from {msg:?}");
7007 assert!(!err.is_transient());
7008 }
7009
7010 #[tokio::test]
7012 async fn margin_trades_report_an_undecodable_response_as_internal() {
7013 let err = margin_trades_error(
7014 wiremock::ResponseTemplate::new(200)
7015 .set_body_json(serde_json::json!({"unexpected": 1})),
7016 )
7017 .await;
7018
7019 assert!(
7020 matches!(err, UnindexedClientError::Internal(_)),
7021 "got {err:?}"
7022 );
7023 assert!(!err.is_transient());
7024 }
7025
7026 #[tokio::test]
7029 async fn sapi_used_ip_weight_near_the_limit_pauses_queries() {
7030 for (used, pauses) in [("10799", false), ("10800", true), ("not a number", false)] {
7031 let server = wiremock::MockServer::start().await;
7032 wiremock::Mock::given(wiremock::matchers::method("GET"))
7033 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
7034 .respond_with(
7035 wiremock::ResponseTemplate::new(200)
7036 .insert_header("x-sapi-used-ip-weight-1m", used)
7037 .set_body_json(serde_json::json!([])),
7038 )
7039 .mount(&server)
7040 .await;
7041 let rest = Arc::new(MarginTradingRestApi::from_config(
7042 ConfigurationRestApi::builder()
7043 .api_key("key")
7044 .api_secret("secret")
7045 .base_path(server.uri())
7046 .build()
7047 .unwrap(),
7048 ));
7049 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7050
7051 fetch_margin_all_open_orders(rest, Arc::clone(&tracker), false)
7052 .await
7053 .unwrap();
7054
7055 let query_waits = tokio::time::timeout(
7056 Duration::from_millis(50),
7057 tracker.wait_if_blocked(RequestKind::Query),
7058 )
7059 .await
7060 .is_err();
7061 assert_eq!(query_waits, pauses, "used IP weight {used}");
7062 tokio::time::timeout(
7063 Duration::from_millis(50),
7064 tracker.wait_if_blocked(RequestKind::Order),
7065 )
7066 .await
7067 .expect("orders never wait on the pause");
7068 }
7069 }
7070
7071 #[tokio::test]
7073 async fn margin_fill_recovery_does_not_wait_for_the_weight_pause() {
7074 let server = wiremock::MockServer::start().await;
7075 wiremock::Mock::given(wiremock::matchers::method("GET"))
7076 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7077 .respond_with(
7079 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([{
7080 "symbol": "BTCUSDT",
7081 "id": 1,
7082 "orderId": 7,
7083 "price": "100",
7084 "qty": "1",
7085 "commission": "0",
7086 "commissionAsset": "USDT",
7087 "time": Utc::now().timestamp_millis(),
7088 "isBuyer": true,
7089 "isMaker": false,
7090 }])),
7091 )
7092 .expect(2)
7093 .mount(&server)
7094 .await;
7095 let rest = Arc::new(MarginTradingRestApi::from_config(
7096 ConfigurationRestApi::builder()
7097 .api_key("key")
7098 .api_secret("secret")
7099 .base_path(server.uri())
7100 .build()
7101 .unwrap(),
7102 ));
7103 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7104 tracker.throttle(Duration::from_secs(60));
7105 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
7106 let mut unrecovered = UnrecoveredFills::default();
7107 unrecovered.open(
7108 &[InstrumentNameExchange::new("BTCUSDT")],
7109 Utc::now() - chrono::Duration::minutes(1),
7110 Utc::now(),
7111 );
7112
7113 tokio::time::timeout(
7115 Duration::from_millis(800),
7116 recover_margin_fills(
7117 &rest,
7118 &tracker,
7119 &mut unrecovered,
7120 &tx,
7121 &new_dedup_cache(),
7122 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7123 false,
7124 ),
7125 )
7126 .await
7127 .expect("recovery does not wait for the pause");
7128 }
7129
7130 #[tokio::test]
7133 async fn margin_trades_read_over_a_span_stop_at_its_end() {
7134 let now = Utc::now().timestamp_millis();
7135 let trades = vec![
7136 (1, now - 50 * HOUR_MS),
7137 (2, now - 49 * HOUR_MS),
7138 (3, now - 10 * HOUR_MS),
7139 ];
7140 let (ids, _) = read_margin_trades(
7141 venue(trades),
7142 MyTradesFrom::Span {
7143 start: now - 60 * HOUR_MS,
7144 end: now - 20 * HOUR_MS,
7145 },
7146 )
7147 .await;
7148
7149 assert_eq!(ids, vec![1, 2]);
7150 }
7151
7152 #[tokio::test]
7155 async fn a_margin_gap_whose_every_read_fails_is_reported_when_given_up() {
7156 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7157 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7158 let server = wiremock::MockServer::start().await;
7159 wiremock::Mock::given(wiremock::matchers::method("GET"))
7160 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7161 .respond_with(
7162 wiremock::ResponseTemplate::new(400)
7163 .set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
7164 )
7165 .mount(&server)
7166 .await;
7167 let rest = Arc::new(MarginTradingRestApi::from_config(
7168 ConfigurationRestApi::builder()
7169 .api_key("key")
7170 .api_secret("secret")
7171 .base_path(server.uri())
7172 .build()
7173 .expect("valid config"),
7174 ));
7175 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7176 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
7177 let dedup = new_dedup_cache();
7178 let btc = InstrumentNameExchange::new("BTCUSDT");
7179 let mut unrecovered = UnrecoveredFills::default();
7180 unrecovered.open(std::slice::from_ref(&btc), disconnect, reconnect);
7181
7182 for _ in 0..=crate::client::order_recovery::MAX_GAP_RETRIES {
7183 recover_margin_fills(
7184 &rest,
7185 &tracker,
7186 &mut unrecovered,
7187 &tx,
7188 &dedup,
7189 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7190 false,
7191 )
7192 .await;
7193 unrecovered.make_due();
7194 }
7195
7196 assert!(unrecovered.is_empty(), "the gap is given up");
7197 let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7198 let [event] = forwarded.as_slice() else {
7199 panic!("only the give-up is sent: {forwarded:?}");
7200 };
7201 let AccountEventKind::FillRecoveryGaveUp(gave_up) = &event.kind else {
7202 panic!("expected FillRecoveryGaveUp, got {event:?}");
7203 };
7204 assert_eq!(event.exchange, ExchangeId::BinanceMargin);
7205 assert_eq!(
7206 gave_up.scope,
7207 crate::FillRecoveryScope::Instruments(vec![btc])
7208 );
7209 assert_eq!(
7210 gave_up.start.timestamp_millis(),
7211 disconnect.timestamp_millis()
7212 );
7213 assert_eq!(
7214 gave_up.attempts,
7215 crate::client::order_recovery::MAX_GAP_RETRIES + 1
7216 );
7217 assert!(
7218 matches!(&gave_up.reason, FillRecoveryFailure::Request(reason) if reason.contains("rejected")),
7219 "the last read's error: {:?}",
7220 gave_up.reason
7221 );
7222 }
7223
7224 #[tokio::test]
7227 async fn a_failed_margin_recovery_keeps_its_closed_gap_for_the_retry() {
7228 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7229 let reconnect = Utc::now() - chrono::Duration::minutes(5);
7230 let server = wiremock::MockServer::start().await;
7231 wiremock::Mock::given(wiremock::matchers::method("GET"))
7232 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7233 .respond_with(
7234 wiremock::ResponseTemplate::new(400)
7235 .set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
7236 )
7237 .up_to_n_times(1)
7238 .with_priority(1)
7239 .mount(&server)
7240 .await;
7241 wiremock::Mock::given(wiremock::matchers::method("GET"))
7242 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7243 .respond_with(venue(vec![
7244 (
7245 1,
7246 (disconnect + chrono::Duration::minutes(1)).timestamp_millis(),
7247 ),
7248 (
7249 2,
7250 (Utc::now() - chrono::Duration::minutes(1)).timestamp_millis(),
7251 ),
7252 ]))
7253 .mount(&server)
7254 .await;
7255 let rest = Arc::new(MarginTradingRestApi::from_config(
7256 ConfigurationRestApi::builder()
7257 .api_key("key")
7258 .api_secret("secret")
7259 .base_path(server.uri())
7260 .build()
7261 .unwrap(),
7262 ));
7263 let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
7264 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
7265 let dedup = new_dedup_cache();
7266 let mut unrecovered = UnrecoveredFills::default();
7267 unrecovered.open(
7268 &[InstrumentNameExchange::new("BTCUSDT")],
7269 disconnect,
7270 reconnect,
7271 );
7272
7273 recover_margin_fills(
7274 &rest,
7275 &tracker,
7276 &mut unrecovered,
7277 &tx,
7278 &dedup,
7279 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7280 false,
7281 )
7282 .await;
7283 assert!(!unrecovered.is_empty(), "the failed gap is kept");
7284 assert!(rx.try_recv().is_err(), "nothing forwarded");
7285
7286 unrecovered.make_due();
7287 recover_margin_fills(
7288 &rest,
7289 &tracker,
7290 &mut unrecovered,
7291 &tx,
7292 &dedup,
7293 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7294 false,
7295 )
7296 .await;
7297 assert!(unrecovered.is_empty(), "the retry recovered it");
7298 let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7299 assert_eq!(
7300 forwarded.len(),
7301 1,
7302 "only the fill inside the gap: {forwarded:?}"
7303 );
7304 }
7305
7306 #[tokio::test]
7309 async fn a_timed_out_margin_recovery_charges_only_the_gaps_it_started() {
7310 let server = wiremock::MockServer::start().await;
7311 wiremock::Mock::given(wiremock::matchers::method("GET"))
7312 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7313 .respond_with(
7314 wiremock::ResponseTemplate::new(200)
7315 .set_body_json(serde_json::json!([]))
7316 .set_delay(Duration::from_secs(3_600)),
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 .timeout(3_600_000_u64)
7326 .retries(0_u32)
7327 .build()
7328 .unwrap(),
7329 ));
7330 let instruments: Vec<_> = (0..9)
7331 .map(|i| InstrumentNameExchange::new(format!("SYM{i}USDT")))
7332 .collect();
7333 let mut unrecovered = UnrecoveredFills::default();
7334 unrecovered.open(
7335 &instruments,
7336 Utc::now() - chrono::Duration::minutes(10),
7337 Utc::now(),
7338 );
7339 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
7340 tokio::time::pause();
7342
7343 recover_margin_fills(
7344 &rest,
7345 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7346 &mut unrecovered,
7347 &tx,
7348 &new_dedup_cache(),
7349 &KnownLiveOrders::shared(ExchangeId::BinanceMargin),
7350 false,
7351 )
7352 .await;
7353
7354 let now = tokio::time::Instant::now();
7355 assert_eq!(
7356 unrecovered.due(now).len(),
7357 1,
7358 "the gap never started stays due"
7359 );
7360 assert_eq!(
7361 unrecovered
7362 .due(now + Duration::from_secs(crate::client::order_recovery::GAP_RETRY_BASE_SECS))
7363 .len(),
7364 9,
7365 "the eight started wait for their first retry"
7366 );
7367 }
7368
7369 fn margin_order_row(cid: &str, status: &str) -> serde_json::Value {
7375 serde_json::json!({
7376 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid, "price": "100",
7377 "origQty": "2", "executedQty": "1", "cummulativeQuoteQty": "105", "status": status,
7378 "timeInForce": "GTC", "type": "LIMIT", "side": "BUY", "isIsolated": false,
7379 "time": 1_700_000_000_000_i64, "updateTime": 1_700_000_060_000_i64,
7380 })
7381 }
7382
7383 fn margin_key(instrument: &str, cid: &str) -> UnindexedOrderKey {
7384 OrderKey::new(
7385 ExchangeId::BinanceMargin,
7386 InstrumentNameExchange::new(instrument),
7387 StrategyId::new("strategy"),
7388 ClientOrderId::new(cid),
7389 )
7390 }
7391
7392 #[test]
7393 fn a_margin_order_row_says_how_the_order_ended() {
7394 use crate::order::state::InactiveOrderState;
7395 use binance_sdk::margin_trading::rest_api::QueryMarginAccountsOrderResponse;
7396 let key = margin_key("BTCUSDT", "a");
7397 let ended = |row: serde_json::Value| {
7398 let row: QueryMarginAccountsOrderResponse = serde_json::from_value(row).unwrap();
7399 convert_ended_order(&row, ExchangeId::BinanceMargin, &key, None)
7400 .map(|order| order.state)
7401 };
7402
7403 let mut filled = margin_order_row("a", "FILLED");
7404 filled["executedQty"] = "2".into();
7405 filled["cummulativeQuoteQty"] = "210".into();
7406 let Some(InactiveOrderState::FullyFilled(filled)) = ended(filled) else {
7407 panic!("FILLED is fully filled");
7408 };
7409 assert_eq!(filled.avg_price, Some(Decimal::from(105)));
7410 assert!(matches!(
7411 ended(margin_order_row("a", "CANCELED")),
7412 Some(InactiveOrderState::Cancelled(cancelled)) if cancelled.filled_quantity == Some(Decimal::ONE)
7413 ));
7414 assert_eq!(ended(margin_order_row("a", "NEW")), None);
7415
7416 let mut unknown_quote = margin_order_row("a", "FILLED");
7418 unknown_quote["executedQty"] = "2".into();
7419 unknown_quote["cummulativeQuoteQty"] = "-1".into();
7420 let Some(InactiveOrderState::FullyFilled(filled)) = ended(unknown_quote) else {
7421 panic!("still fully filled");
7422 };
7423 assert_eq!(filled.avg_price, None, "no negative average price");
7424 }
7425
7426 fn margin_rest_at(server: &wiremock::MockServer) -> Arc<RestApi> {
7427 Arc::new(MarginTradingRestApi::from_config(
7428 ConfigurationRestApi::builder()
7429 .api_key("key")
7430 .api_secret("secret")
7431 .base_path(server.uri())
7432 .build()
7433 .unwrap(),
7434 ))
7435 }
7436
7437 pub(super) fn margin_client_at(
7439 server: &wiremock::MockServer,
7440 is_isolated: bool,
7441 ) -> BinanceMargin {
7442 let config = if is_isolated {
7443 BinanceMarginConfig::isolated(
7444 "key".into(),
7445 "secret".into(),
7446 vec![InstrumentNameExchange::new("BTCUSDT")],
7447 )
7448 } else {
7449 BinanceMarginConfig::cross_margin("key".into(), "secret".into())
7450 };
7451 let mut client = <BinanceMargin as ExecutionClient>::new(config);
7452 client.rest = margin_rest_at(server);
7453 client
7454 }
7455
7456 async fn mount_margin_order(
7458 server: &wiremock::MockServer,
7459 cid: &str,
7460 response: wiremock::ResponseTemplate,
7461 ) {
7462 wiremock::Mock::given(wiremock::matchers::method("GET"))
7463 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7464 .and(wiremock::matchers::query_param("origClientOrderId", cid))
7465 .respond_with(response)
7466 .mount(server)
7467 .await;
7468 }
7469
7470 fn margin_venue_error(code: i64) -> wiremock::ResponseTemplate {
7471 wiremock::ResponseTemplate::new(400)
7472 .set_body_json(serde_json::json!({"code": code, "msg": "refused"}))
7473 }
7474
7475 fn margin_row_response(cid: &str, status: &str) -> wiremock::ResponseTemplate {
7476 wiremock::ResponseTemplate::new(200).set_body_json(margin_order_row(cid, status))
7477 }
7478
7479 async fn margin_requests(server: &wiremock::MockServer) -> Vec<(String, Option<String>)> {
7481 server
7482 .received_requests()
7483 .await
7484 .unwrap()
7485 .iter()
7486 .map(|request| {
7487 let isolated = request
7488 .url
7489 .query_pairs()
7490 .find(|(key, _)| key == "isIsolated")
7491 .map(|(_, value)| value.into_owned());
7492 (request.url.path().to_owned(), isolated)
7493 })
7494 .collect()
7495 }
7496
7497 #[tokio::test]
7498 async fn margin_fetch_ended_orders_asks_the_configured_account_and_omits_the_rest() {
7499 use crate::order::state::InactiveOrderState;
7500 for (is_isolated, flag) in [(false, "FALSE"), (true, "TRUE")] {
7501 let server = wiremock::MockServer::start().await;
7502 mount_margin_order(&server, "filled", margin_row_response("filled", "FILLED")).await;
7503 mount_margin_order(&server, "live", margin_row_response("live", "NEW")).await;
7504 mount_margin_order(&server, "unknown", margin_venue_error(-2013)).await;
7505 mount_margin_order(&server, "no-symbol", margin_venue_error(-1121)).await;
7506 let client = margin_client_at(&server, is_isolated);
7507 let keys = ["filled", "live", "unknown"].map(|cid| margin_key("BTCUSDT", cid));
7508
7509 let ended = client
7510 .fetch_ended_orders(
7511 &[keys.to_vec(), vec![margin_key("NOSUCH", "no-symbol")]].concat(),
7512 )
7513 .await
7514 .unwrap();
7515
7516 let [order] = ended.as_slice() else {
7517 panic!("one order ended: {ended:?}");
7518 };
7519 assert_eq!(order.key, keys[0], "the strategy asked survives");
7520 assert!(matches!(order.state, InactiveOrderState::FullyFilled(_)));
7521 let requests = margin_requests(&server).await;
7522 assert_eq!(requests.len(), 4);
7523 for (path, isolated) in requests {
7524 assert_eq!(path, "/sapi/v1/margin/order");
7525 assert_eq!(
7526 isolated.as_deref(),
7527 Some(flag),
7528 "is_isolated = {is_isolated}"
7529 );
7530 }
7531 }
7532 }
7533
7534 #[tokio::test]
7535 async fn margin_fetch_ended_orders_fails_when_any_lookup_fails() {
7536 let server = wiremock::MockServer::start().await;
7537 mount_margin_order(
7538 &server,
7539 "cancelled",
7540 margin_row_response("cancelled", "CANCELED"),
7541 )
7542 .await;
7543 mount_margin_order(&server, "refused", margin_venue_error(-1100)).await;
7544 let client = margin_client_at(&server, false);
7545
7546 let result = client
7547 .fetch_ended_orders(&[
7548 margin_key("BTCUSDT", "cancelled"),
7549 margin_key("BTCUSDT", "refused"),
7550 ])
7551 .await;
7552
7553 assert!(
7554 matches!(
7555 result,
7556 Err(UnindexedClientError::Api(ApiError::RequestRejected(_)))
7557 ),
7558 "{result:?}"
7559 );
7560 }
7561
7562 fn held_margin_orders() -> SharedKnownLiveOrders {
7564 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
7565 for (n, cid) in ["live", "gone", "unknown"].into_iter().enumerate() {
7566 known.lock().live(
7567 &margin_key("BTCUSDT", cid),
7568 Decimal::TWO,
7569 OrderKind::Limit,
7570 &Open::new(
7571 VenueOrderId::Assigned(OrderId::new(n.to_string())),
7572 Utc::now(),
7573 Decimal::ZERO,
7574 ),
7575 );
7576 }
7577 known
7578 }
7579
7580 #[tokio::test]
7585 async fn an_ended_margin_trailing_stop_is_reported_with_the_kind_it_was_held_with() {
7586 let server = wiremock::MockServer::start().await;
7587 wiremock::Mock::given(wiremock::matchers::method("GET"))
7588 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
7589 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
7590 .mount(&server)
7591 .await;
7592 let mut row = margin_order_row("trailing", "CANCELED");
7593 row["type"] = "STOP_LOSS".into();
7594 row["price"] = "0.00000000".into();
7595 row["stopPrice"] = "0.00000000".into();
7596 mount_margin_order(
7597 &server,
7598 "trailing",
7599 wiremock::ResponseTemplate::new(200).set_body_json(row),
7600 )
7601 .await;
7602 let trailing_stop = OrderKind::TrailingStop {
7603 offset: Decimal::from(25),
7604 offset_type: TrailingOffsetType::BasisPoints,
7605 activation_price: None,
7606 };
7607 let hold = |known: &SharedKnownLiveOrders| {
7608 known.lock().live(
7609 &margin_key("BTCUSDT", "trailing"),
7610 Decimal::TWO,
7611 trailing_stop,
7612 &Open::new(
7613 VenueOrderId::Assigned(OrderId::new("7")),
7614 Utc::now(),
7615 Decimal::ZERO,
7616 ),
7617 );
7618 };
7619
7620 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
7621 hold(&known);
7622 let mut unchecked = UncheckedOrders::default();
7623 unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
7624 let (tx, mut rx) = mpsc::unbounded_channel();
7625 recover_margin_ended_orders(
7626 &margin_rest_at(&server),
7627 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7628 &known,
7629 &mut unchecked,
7630 &UnrecoveredFills::default(),
7631 &tx,
7632 false,
7633 )
7634 .await;
7635 let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7636 let [event] = sent.as_slice() else {
7637 panic!("the trailing stop ended: {sent:?}");
7638 };
7639 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
7640 panic!("an order snapshot: {event:?}");
7641 };
7642 assert_eq!(snapshot.0.kind, trailing_stop);
7643 assert!(matches!(
7644 snapshot.0.state,
7645 OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(_))
7646 ));
7647 assert!(!known.lock().contains(&ClientOrderId::new("trailing")));
7648
7649 let client = margin_client_at(&server, false);
7650 hold(&client.known_live);
7651 let ended = client
7652 .fetch_ended_orders(&[margin_key("BTCUSDT", "trailing")])
7653 .await
7654 .expect("lookup");
7655 let [order] = ended.as_slice() else {
7656 panic!("the trailing stop ended: {ended:?}");
7657 };
7658 assert_eq!(order.kind, trailing_stop);
7659 }
7660
7661 #[tokio::test]
7662 async fn a_margin_reconnect_reports_how_an_unlisted_order_ended_in_the_configured_account() {
7663 for (is_isolated, flag) in [(false, "FALSE"), (true, "TRUE")] {
7664 let server = wiremock::MockServer::start().await;
7665 wiremock::Mock::given(wiremock::matchers::method("GET"))
7666 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
7667 .and(wiremock::matchers::query_param("symbol", "BTCUSDT"))
7668 .respond_with(
7669 wiremock::ResponseTemplate::new(200)
7670 .set_body_json(serde_json::json!([margin_order_row("live", "NEW")])),
7671 )
7672 .mount(&server)
7673 .await;
7674 mount_margin_order(&server, "gone", margin_row_response("gone", "CANCELED")).await;
7675 mount_margin_order(&server, "unknown", margin_venue_error(-2013)).await;
7676 let known = held_margin_orders();
7677 let mut unchecked = UncheckedOrders::default();
7678 unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
7679 let (tx, mut rx) = mpsc::unbounded_channel();
7680
7681 recover_margin_ended_orders(
7682 &margin_rest_at(&server),
7683 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
7684 &known,
7685 &mut unchecked,
7686 &UnrecoveredFills::default(),
7687 &tx,
7688 is_isolated,
7689 )
7690 .await;
7691
7692 let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
7693 let [event] = sent.as_slice() else {
7694 panic!("one order ended: {sent:?}");
7695 };
7696 assert_eq!(event.exchange, ExchangeId::BinanceMargin);
7697 let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
7698 panic!("an order snapshot: {event:?}");
7699 };
7700 assert_eq!(snapshot.0.key.cid, ClientOrderId::new("gone"));
7701 assert_eq!(snapshot.0.key.strategy, StrategyId::unknown());
7702 assert!(matches!(
7703 snapshot.0.state,
7704 OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(_))
7705 ));
7706 let mut requests = margin_requests(&server).await;
7707 requests.sort();
7708 let expected = |path: &str| (path.to_owned(), Some(flag.to_owned()));
7709 assert_eq!(
7710 requests,
7711 [
7712 expected("/sapi/v1/margin/openOrders"),
7713 expected("/sapi/v1/margin/order"),
7714 expected("/sapi/v1/margin/order"),
7715 ],
7716 "one listing, and a lookup of each unlisted order only"
7717 );
7718 let known = known.lock();
7719 assert!(known.contains(&ClientOrderId::new("live")));
7720 assert!(!known.contains(&ClientOrderId::new("gone")));
7721 assert!(
7722 !known.contains(&ClientOrderId::new("unknown")),
7723 "asking again cannot help"
7724 );
7725 assert!(unchecked.is_empty());
7726 }
7727 }
7728
7729 fn margin_open_request<'a>(
7730 instrument: &'a InstrumentNameExchange,
7731 cid: &str,
7732 ) -> OrderRequestOpen<ExchangeId, &'a InstrumentNameExchange> {
7733 OrderRequestOpen {
7734 key: OrderKey::new(
7735 ExchangeId::BinanceMargin,
7736 instrument,
7737 StrategyId::new("strategy"),
7738 ClientOrderId::new(cid),
7739 ),
7740 state: crate::order::request::RequestOpen {
7741 side: Side::Buy,
7742 price: Some(Decimal::from(100)),
7743 quantity: Decimal::TWO,
7744 kind: OrderKind::Limit,
7745 time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
7746 position_id: None,
7747 reduce_only: false,
7748 market: None,
7749 },
7750 }
7751 }
7752
7753 #[tokio::test]
7756 async fn a_margin_trailing_stop_is_sent_with_trailing_delta() {
7757 let server = wiremock::MockServer::start().await;
7758 let sent_as_trailing_stop = |request: &wiremock::Request| {
7759 let mut body_as_query = request.url.clone();
7761 body_as_query.set_query(Some(&String::from_utf8_lossy(&request.body)));
7762 let params: Vec<(String, String)> = request
7763 .url
7764 .query_pairs()
7765 .chain(body_as_query.query_pairs())
7766 .map(|(k, v)| (k.into_owned(), v.into_owned()))
7767 .collect();
7768 let has = |key: &str, value: &str| params.iter().any(|(k, v)| k == key && v == value);
7769 has("type", "STOP_LOSS")
7770 && has("trailingDelta", "25")
7771 && has("sideEffectType", "AUTO_BORROW_REPAY")
7772 && !params.iter().any(|(k, _)| k == "stopPrice" || k == "price")
7773 };
7774 wiremock::Mock::given(wiremock::matchers::method("POST"))
7775 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7776 .and(sent_as_trailing_stop)
7777 .respond_with(
7778 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
7779 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "trailing",
7780 "transactTime": 1_700_000_000_000_i64, "price": "0", "origQty": "2",
7781 "executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
7782 "timeInForce": "GTC", "type": "STOP_LOSS", "side": "SELL",
7783 })),
7784 )
7785 .expect(1)
7786 .mount(&server)
7787 .await;
7788 let client = margin_client_at(&server, false);
7789 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
7790 let trailing_request = |cid, offset| {
7791 let mut request = margin_open_request(&btcusdt, cid);
7792 request.state.side = Side::Sell;
7793 request.state.price = None;
7794 request.state.kind = OrderKind::TrailingStop {
7795 offset,
7796 offset_type: TrailingOffsetType::BasisPoints,
7797 activation_price: None,
7798 };
7799 request
7800 };
7801
7802 let placed = client
7803 .open_order(trailing_request("trailing", Decimal::from(25)))
7804 .await;
7805 assert!(
7806 matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
7807 "{placed:?}"
7808 );
7809
7810 let refused = client
7811 .open_order(trailing_request("fractional", Decimal::new(25, 1)))
7812 .await;
7813 assert!(
7814 matches!(
7815 refused.state,
7816 OrderState::Inactive(crate::order::state::InactiveOrderState::OpenFailed(
7817 OrderError::InvalidPrecision(_)
7818 ))
7819 ),
7820 "{refused:?}"
7821 );
7822 }
7823
7824 #[tokio::test]
7825 async fn placing_listing_and_cancelling_margin_orders_feed_the_known_live_set() {
7826 let server = wiremock::MockServer::start().await;
7827 wiremock::Mock::given(wiremock::matchers::method("POST"))
7828 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7829 .respond_with(
7830 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
7831 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "placed",
7832 "transactTime": 1_700_000_000_000_i64, "price": "100", "origQty": "2",
7833 "executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
7834 "timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
7835 })),
7836 )
7837 .mount(&server)
7838 .await;
7839 wiremock::Mock::given(wiremock::matchers::method("DELETE"))
7840 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7841 .respond_with(
7842 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
7843 "symbol": "BTCUSDT", "orderId": "7", "origClientOrderId": "placed",
7844 "clientOrderId": "cancel", "price": "100", "origQty": "2", "executedQty": "0",
7845 "cummulativeQuoteQty": "0", "status": "CANCELED", "timeInForce": "GTC",
7846 "type": "LIMIT", "side": "BUY",
7847 })),
7848 )
7849 .mount(&server)
7850 .await;
7851 wiremock::Mock::given(wiremock::matchers::method("GET"))
7852 .and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
7853 .respond_with(
7854 wiremock::ResponseTemplate::new(200)
7855 .set_body_json(serde_json::json!([margin_order_row("listed", "NEW")])),
7856 )
7857 .mount(&server)
7858 .await;
7859 let client = margin_client_at(&server, false);
7860 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
7861
7862 let placed = client
7863 .open_order(margin_open_request(&btcusdt, "placed"))
7864 .await;
7865 assert!(
7866 matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
7867 "{placed:?}"
7868 );
7869 assert!(
7870 client
7871 .known_live
7872 .lock()
7873 .contains(&ClientOrderId::new("placed"))
7874 );
7875
7876 client.fetch_open_orders(&[]).await.unwrap();
7877 assert!(
7878 client
7879 .known_live
7880 .lock()
7881 .contains(&ClientOrderId::new("listed"))
7882 );
7883
7884 let cancelled = client
7885 .cancel_order(OrderRequestCancel {
7886 key: OrderKey::new(
7887 ExchangeId::BinanceMargin,
7888 &btcusdt,
7889 StrategyId::new("strategy"),
7890 ClientOrderId::new("placed"),
7891 ),
7892 state: crate::order::request::RequestCancel {
7893 id: Some(VenueOrderId::Assigned(OrderId::new("7"))),
7894 },
7895 })
7896 .await;
7897 assert!(cancelled.state.is_ok(), "{cancelled:?}");
7898 assert!(
7899 !client
7900 .known_live
7901 .lock()
7902 .contains(&ClientOrderId::new("placed"))
7903 );
7904 }
7905
7906 #[tokio::test]
7908 async fn a_cancel_response_without_executed_qty_reports_the_fill_unknown() {
7909 let server = wiremock::MockServer::start().await;
7910 wiremock::Mock::given(wiremock::matchers::method("DELETE"))
7911 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
7912 .respond_with(
7913 wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
7914 "symbol": "BTCUSDT", "orderId": "7", "origClientOrderId": "placed",
7915 "clientOrderId": "cancel", "price": "100", "origQty": "2",
7916 "status": "CANCELED", "timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
7917 })),
7918 )
7919 .mount(&server)
7920 .await;
7921 let client = margin_client_at(&server, false);
7922
7923 let cancelled = client
7924 .cancel_order(OrderRequestCancel {
7925 key: OrderKey::new(
7926 ExchangeId::BinanceMargin,
7927 &InstrumentNameExchange::new("BTCUSDT"),
7928 StrategyId::new("strategy"),
7929 ClientOrderId::new("placed"),
7930 ),
7931 state: crate::order::request::RequestCancel {
7932 id: Some(VenueOrderId::Assigned(OrderId::new("7"))),
7933 },
7934 })
7935 .await;
7936 let Ok(cancelled) = cancelled.state else {
7937 panic!("expected Ok, got {cancelled:?}");
7938 };
7939 assert_eq!(cancelled.filled_quantity, None);
7940 }
7941
7942 #[test]
7943 fn a_reconnect_opens_a_gap_on_each_stream_instrument_and_checks_those_with_orders_held() {
7944 let [btc, eth, xrp] = ["BTCUSDT", "ETHUSDT", "XRPUSDT"].map(InstrumentNameExchange::new);
7945 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
7946 for (instrument, cid) in [(&btc, "btc"), (&xrp, "xrp")] {
7947 known.lock().live(
7948 &margin_key(instrument.name(), cid),
7949 Decimal::TWO,
7950 OrderKind::Limit,
7951 &Open::new(
7952 VenueOrderId::Assigned(OrderId::new(cid)),
7953 Utc::now(),
7954 Decimal::ZERO,
7955 ),
7956 );
7957 }
7958 let mut unrecovered = UnrecoveredFills::default();
7959 let mut unchecked = UncheckedOrders::default();
7960
7961 open_reconnect_recovery(
7962 &mut unrecovered,
7963 &mut unchecked,
7964 &known,
7965 &[btc.clone(), eth.clone()],
7966 Utc::now() - chrono::Duration::minutes(1),
7967 );
7968
7969 assert!(unrecovered.covers(&btc) && unrecovered.covers(ð));
7970 assert!(!unrecovered.covers(&xrp), "not the stream's");
7971 assert!(unchecked.contains(&btc));
7972 assert!(!unchecked.contains(ð), "no order held");
7973 assert!(!unchecked.contains(&xrp), "its fills are not recovered");
7974 }
7975
7976 #[tokio::test]
7979 async fn a_recovered_margin_fill_that_completes_a_held_order_ends_it() {
7980 let disconnect = Utc::now() - chrono::Duration::minutes(10);
7981 let server = wiremock::MockServer::start().await;
7982 wiremock::Mock::given(wiremock::matchers::method("GET"))
7983 .and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
7984 .respond_with(venue(vec![
7985 (
7986 1,
7987 (disconnect + chrono::Duration::minutes(1)).timestamp_millis(),
7988 ),
7989 (
7990 2,
7991 (disconnect + chrono::Duration::minutes(2)).timestamp_millis(),
7992 ),
7993 ]))
7994 .mount(&server)
7995 .await;
7996 let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
7999 for (cid, order_id, quantity) in [("one", "1", "0.01"), ("two", "2", "0.02")] {
8000 known.lock().live(
8001 &margin_key("BTCUSDT", cid),
8002 Decimal::from_str(quantity).unwrap(),
8003 OrderKind::Limit,
8004 &Open::new(
8005 VenueOrderId::Assigned(OrderId::new(order_id)),
8006 Utc::now(),
8007 Decimal::ZERO,
8008 ),
8009 );
8010 }
8011 let mut unrecovered = UnrecoveredFills::default();
8012 unrecovered.open(
8013 &[InstrumentNameExchange::new("BTCUSDT")],
8014 disconnect,
8015 Utc::now(),
8016 );
8017 let (tx, mut rx) = mpsc::unbounded_channel();
8018
8019 recover_margin_fills(
8020 &margin_rest_at(&server),
8021 &Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
8022 &mut unrecovered,
8023 &tx,
8024 &new_dedup_cache(),
8025 &known,
8026 false,
8027 )
8028 .await;
8029
8030 assert_eq!(
8031 std::iter::from_fn(|| rx.try_recv().ok()).count(),
8032 2,
8033 "both fills"
8034 );
8035 let known = known.lock();
8036 assert!(!known.contains(&ClientOrderId::new("one")), "completed");
8037 assert!(known.contains(&ClientOrderId::new("two")), "part-filled");
8038 }
8039
8040 #[tokio::test]
8043 async fn a_margin_order_that_expired_on_placement_is_not_reported_or_held_open() {
8044 use rust_decimal_macros::dec;
8045
8046 for (cid, filled, quote) in [("unfilled", "0", "0"), ("partly", "1", "100")] {
8047 let server = wiremock::MockServer::start().await;
8048 wiremock::Mock::given(wiremock::matchers::method("POST"))
8049 .and(wiremock::matchers::path("/sapi/v1/margin/order"))
8050 .respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
8051 serde_json::json!({
8052 "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid,
8053 "transactTime": 1_700_000_000_000_i64, "price": "100", "origQty": "2",
8054 "executedQty": filled, "cummulativeQuoteQty": quote,
8055 "status": "EXPIRED", "timeInForce": "IOC", "type": "LIMIT", "side": "BUY",
8056 }),
8057 ))
8058 .mount(&server)
8059 .await;
8060 let client = margin_client_at(&server, false);
8061 let btcusdt = InstrumentNameExchange::new("BTCUSDT");
8062
8063 let placed = client.open_order(margin_open_request(&btcusdt, cid)).await;
8064
8065 let OrderState::Inactive(crate::order::state::InactiveOrderState::Expired(expired)) =
8066 &placed.state
8067 else {
8068 panic!("{cid}: expired, not open: {placed:?}");
8069 };
8070 assert_eq!(expired.filled_quantity, Decimal::from_str(filled).ok());
8071 let late = Open::new(
8072 VenueOrderId::Assigned(OrderId::new("7")),
8073 Utc::now(),
8074 Decimal::ZERO,
8075 );
8076 let mut known = client.known_live.lock();
8077 known.live(&placed.key, dec!(2), OrderKind::Limit, &late);
8078 assert!(
8079 !known.contains(&ClientOrderId::new(cid)),
8080 "{cid}: not held, even after a late live report"
8081 );
8082 }
8083 }
8084
8085 async fn mount_borrow_terms(
8088 server: &wiremock::MockServer,
8089 pair: serde_json::Value,
8090 asset: serde_json::Value,
8091 rate: serde_json::Value,
8092 max_borrow: serde_json::Value,
8093 ) {
8094 use wiremock::{
8095 Mock, ResponseTemplate,
8096 matchers::{method, path},
8097 };
8098 for (route, body) in [
8099 ("/sapi/v1/margin/allPairs", pair.clone()),
8100 ("/sapi/v1/margin/isolated/allPairs", pair),
8101 ("/sapi/v1/margin/allAssets", asset),
8102 ("/sapi/v1/margin/next-hourly-interest-rate", rate),
8103 ("/sapi/v1/margin/maxBorrowable", max_borrow),
8104 ] {
8105 Mock::given(method("GET"))
8106 .and(path(route))
8107 .respond_with(ResponseTemplate::new(200).set_body_json(body))
8108 .mount(server)
8109 .await;
8110 }
8111 }
8112
8113 fn btcusdt_pair(is_margin_trade: bool, is_sell_allowed: bool) -> serde_json::Value {
8114 serde_json::json!([{
8115 "base": "BTC", "quote": "USDT", "symbol": "BTCUSDT",
8116 "isMarginTrade": is_margin_trade, "isSellAllowed": is_sell_allowed, "isBuyAllowed": true,
8117 }])
8118 }
8119
8120 fn btc_asset(is_borrowable: bool) -> serde_json::Value {
8121 serde_json::json!([{ "assetName": "BTC", "isBorrowable": is_borrowable }])
8122 }
8123
8124 fn btc_rate(rate: &str) -> serde_json::Value {
8125 serde_json::json!([{ "asset": "BTC", "nextHourlyInterestRate": rate }])
8126 }
8127
8128 async fn query_value(server: &wiremock::MockServer, route: &str, key: &str) -> Option<String> {
8130 server
8131 .received_requests()
8132 .await
8133 .unwrap()
8134 .iter()
8135 .find(|request| request.url.path() == route)
8136 .and_then(|request| {
8137 request
8138 .url
8139 .query_pairs()
8140 .find(|(name, _)| name == key)
8141 .map(|(_, value)| value.into_owned())
8142 })
8143 }
8144
8145 #[tokio::test]
8146 async fn margin_shortability_reads_the_pair_the_asset_and_the_annualised_rate() {
8147 use crate::client::ShortabilityClient;
8148 use rust_decimal_macros::dec;
8149 for (is_isolated, pairs_route, flag) in [
8150 (false, "/sapi/v1/margin/allPairs", "FALSE"),
8151 (true, "/sapi/v1/margin/isolated/allPairs", "TRUE"),
8152 ] {
8153 let server = wiremock::MockServer::start().await;
8154 mount_borrow_terms(
8155 &server,
8156 btcusdt_pair(true, true),
8157 btc_asset(true),
8158 btc_rate("0.00000571"),
8159 serde_json::json!({}),
8160 )
8161 .await;
8162
8163 let shortability = margin_client_at(&server, is_isolated)
8164 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8165 .await
8166 .unwrap();
8167
8168 assert_eq!(
8169 shortability,
8170 Shortability::new()
8171 .with_shortable(true)
8172 .with_fee_rate(dec!(0.00000571) * dec!(8760)),
8173 "isolated: {is_isolated}"
8174 );
8175 assert_eq!(
8176 query_value(&server, pairs_route, "symbol").await.as_deref(),
8177 Some("BTCUSDT"),
8178 "the pair list is asked for the one symbol"
8179 );
8180 assert_eq!(
8181 query_value(&server, "/sapi/v1/margin/allAssets", "asset")
8182 .await
8183 .as_deref(),
8184 Some("BTC"),
8185 "the base asset is the one borrowed"
8186 );
8187 let rate = "/sapi/v1/margin/next-hourly-interest-rate";
8188 assert_eq!(
8189 query_value(&server, rate, "assets").await.as_deref(),
8190 Some("BTC")
8191 );
8192 assert_eq!(
8193 query_value(&server, rate, "isIsolated").await.as_deref(),
8194 Some(flag)
8195 );
8196 }
8197 }
8198
8199 #[tokio::test]
8200 async fn margin_shortability_is_refused_by_any_flag_and_unknown_without_them() {
8201 use crate::client::ShortabilityClient;
8202 let cases = [
8203 (btcusdt_pair(false, true), btc_asset(true), Some(false)),
8204 (btcusdt_pair(true, false), btc_asset(true), Some(false)),
8205 (btcusdt_pair(true, true), btc_asset(false), Some(false)),
8206 (btcusdt_pair(true, true), serde_json::json!([]), None),
8208 ];
8209 for (pair, asset, shortable) in cases {
8210 let server = wiremock::MockServer::start().await;
8211 mount_borrow_terms(
8212 &server,
8213 pair.clone(),
8214 asset.clone(),
8215 serde_json::json!([]),
8216 serde_json::json!({}),
8217 )
8218 .await;
8219
8220 let shortability = margin_client_at(&server, false)
8221 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8222 .await
8223 .unwrap();
8224
8225 assert_eq!(shortability.shortable, shortable, "{pair} {asset}");
8226 assert_eq!(
8227 shortability.fee_rate, None,
8228 "no rate returned is an unknown fee"
8229 );
8230 assert_eq!(shortability.easy_to_borrow, None);
8231 assert_eq!(shortability.available, None);
8232 }
8233 }
8234
8235 #[tokio::test]
8236 async fn margin_shortability_of_an_unlisted_pair_is_an_invalid_instrument() {
8237 use crate::client::ShortabilityClient;
8238 let server = wiremock::MockServer::start().await;
8239 mount_borrow_terms(
8240 &server,
8241 serde_json::json!([]),
8242 btc_asset(true),
8243 btc_rate("0.00000571"),
8244 serde_json::json!({}),
8245 )
8246 .await;
8247
8248 let result = margin_client_at(&server, false)
8249 .fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
8250 .await;
8251
8252 assert!(
8253 matches!(
8254 result,
8255 Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(ref instrument, _)))
8256 if instrument.name().as_str() == "BTCUSDT"
8257 ),
8258 "{result:?}"
8259 );
8260 }
8261
8262 #[tokio::test]
8263 async fn margin_borrow_capacity_asks_for_the_asset_the_side_borrows() {
8264 use crate::client::BorrowCapacityClient;
8265 use rust_decimal_macros::dec;
8266 for (is_isolated, side, asset, isolated_symbol) in [
8267 (false, Side::Sell, "BTC", None),
8268 (false, Side::Buy, "USDT", None),
8269 (true, Side::Sell, "BTC", Some("BTCUSDT")),
8270 ] {
8271 let server = wiremock::MockServer::start().await;
8272 mount_borrow_terms(
8273 &server,
8274 btcusdt_pair(true, true),
8275 btc_asset(true),
8276 btc_rate("0.00000571"),
8277 serde_json::json!({ "amount": "1.69248805", "borrowLimit": "60" }),
8278 )
8279 .await;
8280
8281 let capacity = margin_client_at(&server, is_isolated)
8282 .fetch_borrow_capacity(&InstrumentNameExchange::new("BTCUSDT"), side)
8283 .await
8284 .unwrap();
8285
8286 assert_eq!(
8287 capacity,
8288 BorrowCapacity::new(AssetNameExchange::new(asset))
8289 .with_borrowable_now(dec!(1.69248805))
8290 .with_account_limit(dec!(60)),
8291 "{side:?} isolated: {is_isolated}"
8292 );
8293 let max_borrow = "/sapi/v1/margin/maxBorrowable";
8294 assert_eq!(
8295 query_value(&server, max_borrow, "asset").await.as_deref(),
8296 Some(asset)
8297 );
8298 assert_eq!(
8299 query_value(&server, max_borrow, "isolatedSymbol")
8300 .await
8301 .as_deref(),
8302 isolated_symbol,
8303 "only isolated margin names the pair's account"
8304 );
8305 }
8306 }
8307
8308 #[tokio::test]
8309 async fn margin_borrow_capacity_keeps_an_omitted_amount_unknown_and_fails_a_malformed_one() {
8310 use crate::client::BorrowCapacityClient;
8311 use rust_decimal_macros::dec;
8312 let instrument = InstrumentNameExchange::new("BTCUSDT");
8313
8314 let server = wiremock::MockServer::start().await;
8315 mount_borrow_terms(
8316 &server,
8317 btcusdt_pair(true, true),
8318 btc_asset(true),
8319 btc_rate("0"),
8320 serde_json::json!({ "borrowLimit": "60" }),
8321 )
8322 .await;
8323 let capacity = margin_client_at(&server, false)
8324 .fetch_borrow_capacity(&instrument, Side::Sell)
8325 .await
8326 .unwrap();
8327 assert_eq!(capacity.borrowable_now, None);
8328 assert_eq!(capacity.account_limit, Some(dec!(60)));
8329
8330 let server = wiremock::MockServer::start().await;
8331 mount_borrow_terms(
8332 &server,
8333 btcusdt_pair(true, true),
8334 btc_asset(true),
8335 btc_rate("0"),
8336 serde_json::json!({ "amount": "lots", "borrowLimit": "60" }),
8337 )
8338 .await;
8339 let result = margin_client_at(&server, false)
8340 .fetch_borrow_capacity(&instrument, Side::Sell)
8341 .await;
8342 assert!(
8343 matches!(result, Err(UnindexedClientError::Internal(ref message)) if message.contains("amount")),
8344 "{result:?}"
8345 );
8346 }
8347
8348 #[tokio::test]
8349 async fn margin_borrow_capacity_without_an_isolated_account_names_the_pair() {
8350 use crate::client::BorrowCapacityClient;
8351 use wiremock::{
8352 Mock,
8353 matchers::{method, path},
8354 };
8355 let server = wiremock::MockServer::start().await;
8356 Mock::given(method("GET"))
8357 .and(path("/sapi/v1/margin/maxBorrowable"))
8358 .respond_with(
8359 wiremock::ResponseTemplate::new(400).set_body_json(serde_json::json!({
8360 "code": -11001, "msg": "Isolated margin account does not exist."
8361 })),
8362 )
8363 .mount(&server)
8364 .await;
8365 mount_borrow_terms(
8366 &server,
8367 btcusdt_pair(true, true),
8368 btc_asset(true),
8369 btc_rate("0"),
8370 serde_json::json!({}),
8371 )
8372 .await;
8373
8374 let err = margin_client_at(&server, true)
8375 .fetch_borrow_capacity(&InstrumentNameExchange::new("BTCUSDT"), Side::Sell)
8376 .await
8377 .unwrap_err();
8378
8379 assert_eq!(
8380 err.to_string(),
8381 "API: request rejected: BTCUSDT: -11001 Isolated margin account does not exist."
8382 );
8383 assert!(!err.is_transient());
8384 }
8385}