use super::shared::{
AbortOnDropStream, BINANCE_MAX_TRADES, BinanceOrderType, BinanceTimeInForce,
CONNECT_TIMEOUT_SECS, ExponentialBackoff, FILL_RECOVERY_TIMEOUT_SECS, HEARTBEAT_TIMEOUT_SECS,
RateLimitTracker, SIGNAL_RECOVERY_LOOKBACK_MS, SharedDedupCache, classify_order_kind_tif,
classify_rest_order_error, connectivity_error, dedup_key_from_event, is_duplicate,
new_dedup_cache, parse_order_kind, parse_side, parse_time_in_force, rest_call_with_retry,
};
use crate::{
AccountEventKind, AccountSnapshot, InstrumentAccountSnapshot, InstrumentBalanceUpdate,
IsolatedInstrumentState, IsolatedMarginRisk, UnindexedAccountEvent, UnindexedAccountSnapshot,
balance::{AssetBalance, AssetBalanceUpdate, Balance, BalanceUpdate},
client::ExecutionClient,
emit_stream_terminated,
error::{
ApiError, ConnectivityError, OrderError, StreamTerminationReason, UnindexedClientError,
UnindexedOrderError,
},
order::{
Order, OrderKey, OrderKind, TimeInForce,
id::{ClientOrderId, OrderId, StrategyId},
request::{OrderRequestCancel, OrderRequestOpen, UnindexedOrderResponseCancel},
state::{Cancelled, Filled, Open, OrderState, UnindexedOrderState},
},
trade::{AssetFees, Trade, TradeId},
};
use binance_sdk::{
common::{
config::{ConfigurationRestApi, ConfigurationWebsocketApi},
constants::{MARGIN_TRADING_REST_API_PROD_URL, SPOT_WS_API_PROD_URL},
models::WebsocketEvent,
websocket::{
SendWebsocketMessageResult, Subscription, WebsocketApi as WsApiBase,
WebsocketMessageSendOptions,
},
},
margin_trading::{
MarginTradingRestApi,
rest_api::{
MarginAccountCancelOrderParams, MarginAccountNewOrderNewOrderRespTypeEnum,
MarginAccountNewOrderParams, MarginAccountNewOrderSideEnum,
MarginAccountNewOrderTimeInForceEnum, QueryCrossMarginAccountDetailsParams,
QueryCrossMarginAccountDetailsResponseUserAssetsInner,
QueryIsolatedMarginAccountInfoParams,
QueryIsolatedMarginAccountInfoResponseAssetsInner, QueryMarginAccountsOpenOrdersParams,
QueryMarginAccountsOpenOrdersResponseInner, QueryMarginAccountsTradeListParams,
QueryMarginAccountsTradeListResponseInner, RestApi,
},
websocket_streams::{
Executionreport, MarginLevelStatusChange, Outboundaccountposition, UserLiabilityChange,
},
},
};
use chrono::{DateTime, TimeZone, Utc};
use futures::stream::BoxStream;
use rust_decimal::Decimal;
use rustrade_instrument::{
Side, asset::name::AssetNameExchange, exchange::ExchangeId,
instrument::name::InstrumentNameExchange,
};
use serde::{Deserialize, Serialize};
use smol_str::{SmolStr, format_smolstr};
use std::{
collections::{BTreeMap, HashMap},
str::FromStr,
sync::{
Arc, Mutex,
atomic::{AtomicBool, Ordering},
},
time::Duration,
};
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, error, info, trace, warn};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MarginSideEffect {
#[default]
AutoBorrowRepay,
NoBorrow,
}
impl MarginSideEffect {
pub fn as_binance_str(self) -> &'static str {
match self {
MarginSideEffect::AutoBorrowRepay => "AUTO_BORROW_REPAY",
MarginSideEffect::NoBorrow => "NO_SIDE_EFFECT",
}
}
}
#[derive(Clone, Deserialize)]
pub struct BinanceMarginConfig {
api_key: String,
secret_key: String,
pub testnet: bool,
#[serde(default)]
pub is_isolated: bool,
#[serde(default)]
pub isolated_symbols: Vec<InstrumentNameExchange>,
#[serde(default)]
pub side_effect: MarginSideEffect,
}
impl std::fmt::Debug for BinanceMarginConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BinanceMarginConfig")
.field("api_key", &"***")
.field("secret_key", &"***")
.field("testnet", &self.testnet)
.field("is_isolated", &self.is_isolated)
.field("isolated_symbols", &self.isolated_symbols)
.field("side_effect", &self.side_effect)
.finish()
}
}
impl BinanceMarginConfig {
pub fn new(
api_key: String,
secret_key: String,
testnet: bool,
is_isolated: bool,
isolated_symbols: Vec<InstrumentNameExchange>,
side_effect: MarginSideEffect,
) -> Self {
Self {
api_key,
secret_key,
testnet,
is_isolated,
isolated_symbols,
side_effect,
}
}
pub fn cross_margin(api_key: String, secret_key: String) -> Self {
Self::new(
api_key,
secret_key,
false,
false,
Vec::new(),
MarginSideEffect::default(),
)
}
pub fn isolated(
api_key: String,
secret_key: String,
symbols: Vec<InstrumentNameExchange>,
) -> Self {
Self::new(
api_key,
secret_key,
false,
true,
symbols,
MarginSideEffect::default(),
)
}
pub fn api_key(&self) -> &str {
&self.api_key
}
}
#[derive(Clone)]
pub struct BinanceMargin {
config: Arc<BinanceMarginConfig>,
rest: Arc<RestApi>,
rest_config: Arc<ConfigurationRestApi>,
ws_config: ConfigurationWebsocketApi,
rate_limiter: Arc<RateLimitTracker>,
}
impl std::fmt::Debug for BinanceMargin {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BinanceMargin")
.field("testnet", &self.config.testnet)
.field("is_isolated", &self.config.is_isolated)
.field("side_effect", &self.config.side_effect)
.finish_non_exhaustive()
}
}
impl BinanceMargin {
#[allow(clippy::expect_used)] fn build_rest_config(config: &BinanceMarginConfig) -> ConfigurationRestApi {
ConfigurationRestApi::builder()
.api_key(config.api_key.clone())
.api_secret(config.secret_key.clone())
.base_path(MARGIN_TRADING_REST_API_PROD_URL)
.build()
.expect("failed to build Binance margin REST configuration")
}
#[allow(clippy::expect_used)] fn build_ws_config(config: &BinanceMarginConfig) -> ConfigurationWebsocketApi {
ConfigurationWebsocketApi::builder()
.api_key(config.api_key.clone())
.api_secret(config.secret_key.clone())
.ws_url(SPOT_WS_API_PROD_URL)
.build()
.expect("failed to build Binance margin WebSocket configuration")
}
fn effective_isolated_set(
&self,
instruments: &[InstrumentNameExchange],
) -> Vec<InstrumentNameExchange> {
if instruments.is_empty() {
return self.config.isolated_symbols.clone();
}
instruments
.iter()
.filter(|inst| {
let in_set = self.config.isolated_symbols.contains(*inst);
if !in_set {
warn!(
instrument = %inst.name(),
"BinanceMargin isolated: requested instrument not in configured isolated_symbols — skipping"
);
}
in_set
})
.cloned()
.collect()
}
async fn isolated_account_snapshot(
&self,
instruments: &[InstrumentNameExchange],
) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
use futures::{StreamExt as _, TryStreamExt as _};
let effective = self.effective_isolated_set(instruments);
if effective.is_empty() {
return Ok(AccountSnapshot::new(
ExchangeId::BinanceMargin,
Vec::new(),
Vec::new(),
));
}
let mut balances_by_symbol = convert_isolated_margin_assets(
fetch_isolated_margin_account_info(
self.rest.clone(),
self.rate_limiter.clone(),
effective.clone(),
)
.await?,
);
let instrument_snapshots: Vec<_> =
futures::stream::iter(effective.into_iter().map(|instrument| {
fetch_margin_open_orders_for_instrument(
self.rest.clone(),
self.rate_limiter.clone(),
instrument,
true,
)
}))
.buffer_unordered(8)
.map(|result| {
let (inst, orders) = result?;
let wrapped = orders.into_iter().map(active_order_snapshot).collect();
let isolated = balances_by_symbol.remove(&inst);
if isolated.is_none() {
warn!(
instrument = %inst.name(),
"BinanceMargin isolated: no isolated balance entry returned for instrument — snapshot will carry isolated: None"
);
}
Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
inst, wrapped, None, isolated,
))
})
.try_collect()
.await?;
Ok(AccountSnapshot::new(
ExchangeId::BinanceMargin,
Vec::new(),
instrument_snapshots,
))
}
}
impl ExecutionClient for BinanceMargin {
const EXCHANGE: ExchangeId = ExchangeId::BinanceMargin;
type Config = BinanceMarginConfig;
type AccountStream = BoxStream<'static, UnindexedAccountEvent>;
fn new(config: Self::Config) -> Self {
if config.testnet {
warn!(
"BinanceMarginConfig.testnet = true is ignored: Binance margin has no testnet; \
using production endpoints"
);
}
assert!(
!(config.is_isolated && config.isolated_symbols.is_empty()),
"BinanceMarginConfig: is_isolated = true requires a non-empty isolated_symbols"
);
let rest_config = Self::build_rest_config(&config);
let rest = Arc::new(MarginTradingRestApi::production(rest_config.clone()));
let rest_config = Arc::new(rest_config);
let ws_config = Self::build_ws_config(&config);
Self {
config: Arc::new(config),
rest,
rest_config,
ws_config,
rate_limiter: Arc::new(RateLimitTracker::new()),
}
}
async fn open_order(
&self,
request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
) -> Option<Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState>> {
let instrument = request.key.instrument.clone();
let side = request.state.side;
let price = request.state.price;
let quantity = request.state.quantity;
let kind = request.state.kind;
let time_in_force = request.state.time_in_force;
let cid = request.key.cid.clone();
let order_key = OrderKey::new(
ExchangeId::BinanceMargin,
instrument.clone(),
request.key.strategy.clone(),
cid.clone(),
);
let inactive = |state: UnindexedOrderState| {
Some(Order {
key: order_key.clone(),
side,
price,
quantity,
kind,
time_in_force,
state,
})
};
let params = match build_new_order_params(
instrument.name().to_string(),
side,
price,
quantity,
kind,
time_in_force,
cid.0.to_string(),
self.config.side_effect,
self.config.is_isolated,
) {
Ok(params) => params,
Err(BuildOrderError::Unsupported) => {
return inactive(OrderState::inactive(OrderError::UnsupportedOrderType(
format!(
"BinanceMargin does not support OrderKind::{kind:?} with {time_in_force:?}"
),
)));
}
Err(BuildOrderError::Build(msg)) => {
error!(%msg, "BinanceMargin failed to build new order params");
return inactive(OrderState::inactive(OrderError::Rejected(
ApiError::OrderRejected(msg),
)));
}
};
let response = match rest_call_with_retry(&self.rest, &self.rate_limiter, |rest| {
let params = params.clone();
Box::pin(async move { rest.margin_account_new_order(params).await })
})
.await
{
Ok(response) => response,
Err(e) => {
return inactive(OrderState::inactive(classify_rest_order_error(
&e,
&instrument,
)));
}
};
let data = match response.data().await {
Ok(data) => data,
Err(e) => {
return inactive(OrderState::inactive(OrderError::Rejected(
ApiError::OrderRejected(e.to_string()),
)));
}
};
let time_exchange = data
.transact_time
.and_then(|ms| Utc.timestamp_millis_opt(ms).single())
.unwrap_or_else(Utc::now);
let exchange_order_id = match data.order_id {
Some(id) => OrderId(format_smolstr!("{id}")),
None => {
error!("BinanceMargin open_order response missing orderId");
return inactive(OrderState::inactive(OrderError::Rejected(
ApiError::OrderRejected("open_order response missing orderId".into()),
)));
}
};
let filled_qty = match data.executed_qty.as_deref() {
Some(q) => Decimal::from_str(q).unwrap_or_else(|_| {
warn!(
executed_qty = q,
"BinanceMargin: failed to parse executedQty; treating as zero"
);
Decimal::ZERO
}),
None => {
warn!("BinanceMargin: executedQty missing in response; treating as zero");
Decimal::ZERO
}
};
let state = if filled_qty >= quantity {
let avg_price = margin_avg_price(data.cummulative_quote_qty.as_deref(), filled_qty);
OrderState::fully_filled(Filled::new(
exchange_order_id,
time_exchange,
filled_qty,
avg_price,
))
} else {
OrderState::active(Open::new(exchange_order_id, time_exchange, filled_qty))
};
Some(Order {
key: order_key,
side,
price,
quantity,
kind,
time_in_force,
state,
})
}
async fn cancel_order(
&self,
request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
) -> Option<UnindexedOrderResponseCancel> {
let instrument = request.key.instrument.clone();
let key = OrderKey {
exchange: request.key.exchange,
instrument: instrument.clone(),
strategy: request.key.strategy.clone(),
cid: request.key.cid.clone(),
};
let params = match build_cancel_order_params(
instrument.name().to_string(),
request.state.id.as_ref(),
&request.key.cid,
self.config.is_isolated,
) {
Ok(p) => p,
Err(e) => {
error!(%e, "BinanceMargin failed to build cancel order params");
return Some(UnindexedOrderResponseCancel {
key,
state: Err(OrderError::Rejected(ApiError::OrderRejected(e))),
});
}
};
let response = match rest_call_with_retry(&self.rest, &self.rate_limiter, |rest| {
let params = params.clone();
Box::pin(async move { rest.margin_account_cancel_order(params).await })
})
.await
{
Ok(response) => response,
Err(e) => {
return Some(UnindexedOrderResponseCancel {
key,
state: Err(classify_rest_order_error(&e, &instrument)),
});
}
};
let data = match response.data().await {
Ok(data) => data,
Err(e) => {
return Some(UnindexedOrderResponseCancel {
key,
state: Err(OrderError::Rejected(ApiError::OrderRejected(e.to_string()))),
});
}
};
let time_exchange = Utc::now();
let exchange_order_id = match data.order_id {
Some(id) => OrderId(SmolStr::new(id)),
None => {
error!("BinanceMargin cancel response missing orderId");
return Some(UnindexedOrderResponseCancel {
key,
state: Err(OrderError::Rejected(ApiError::OrderRejected(
"cancel response missing orderId".into(),
))),
});
}
};
let filled_qty = match data.executed_qty.as_deref() {
Some(q) => Decimal::from_str(q).unwrap_or_else(|_| {
warn!(
executed_qty = q,
"BinanceMargin: failed to parse executedQty; treating as zero"
);
Decimal::ZERO
}),
None => {
warn!("BinanceMargin: executedQty missing in response; treating as zero");
Decimal::ZERO
}
};
Some(UnindexedOrderResponseCancel {
key,
state: Ok(Cancelled::new(exchange_order_id, time_exchange, filled_qty)),
})
}
async fn account_snapshot(
&self,
assets: &[AssetNameExchange],
instruments: &[InstrumentNameExchange],
) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
if self.config.is_isolated {
return self.isolated_account_snapshot(instruments).await;
}
let response = rest_call_with_retry(&self.rest, &self.rate_limiter, |rest| {
Box::pin(async move {
let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
rest.query_cross_margin_account_details(params).await
})
})
.await
.map_err(connectivity_error)?;
let account = response
.data()
.await
.map_err(|e| connectivity_error(e.into()))?;
let balances =
filter_and_convert_margin_balances(account.user_assets.unwrap_or_default(), assets);
use futures::{StreamExt as _, TryStreamExt as _};
let instrument_snapshots: Vec<_> =
futures::stream::iter(instruments.iter().cloned().map(|instrument| {
fetch_margin_open_orders_for_instrument(
self.rest.clone(),
self.rate_limiter.clone(),
instrument,
false,
)
}))
.buffer_unordered(8)
.map(|result| {
let (inst, orders) = result?;
let wrapped = orders.into_iter().map(active_order_snapshot).collect();
Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
inst, wrapped, None, None,
))
})
.try_collect()
.await?;
Ok(AccountSnapshot::new(
ExchangeId::BinanceMargin,
balances,
instrument_snapshots,
))
}
async fn fetch_balances(
&self,
assets: &[AssetNameExchange],
) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
if self.config.is_isolated {
return Ok(Vec::new());
}
let response = rest_call_with_retry(&self.rest, &self.rate_limiter, |rest| {
Box::pin(async move {
let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
rest.query_cross_margin_account_details(params).await
})
})
.await
.map_err(connectivity_error)?;
let account = response
.data()
.await
.map_err(|e| connectivity_error(e.into()))?;
Ok(filter_and_convert_margin_balances(
account.user_assets.unwrap_or_default(),
assets,
))
}
async fn fetch_open_orders(
&self,
instruments: &[InstrumentNameExchange],
) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
use futures::{StreamExt as _, TryStreamExt as _};
if self.config.is_isolated {
let effective = self.effective_isolated_set(instruments);
let capacity = effective.len();
return futures::stream::iter(effective.into_iter().map(|instrument| {
fetch_margin_open_orders_for_instrument(
self.rest.clone(),
self.rate_limiter.clone(),
instrument,
true,
)
}))
.buffer_unordered(8)
.try_fold(
Vec::with_capacity(capacity),
|mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>,
(_, orders)| async move {
acc.extend(orders);
Ok(acc)
},
)
.await;
}
if instruments.is_empty() {
return fetch_margin_all_open_orders(
self.rest.clone(),
self.rate_limiter.clone(),
false,
)
.await;
}
futures::stream::iter(instruments.iter().cloned().map(|instrument| {
fetch_margin_open_orders_for_instrument(
self.rest.clone(),
self.rate_limiter.clone(),
instrument,
false,
)
}))
.buffer_unordered(8)
.try_fold(
Vec::with_capacity(instruments.len()),
|mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, (_, orders)| async move {
acc.extend(orders);
Ok(acc)
},
)
.await
}
async fn fetch_trades(
&self,
time_since: DateTime<Utc>,
instruments: &[InstrumentNameExchange],
) -> Result<Vec<Trade<AssetNameExchange, InstrumentNameExchange>>, UnindexedClientError> {
use futures::StreamExt as _;
let effective: Vec<InstrumentNameExchange> = if self.config.is_isolated {
self.effective_isolated_set(instruments)
} else {
instruments.to_vec()
};
if effective.is_empty() {
debug!(
is_isolated = self.config.is_isolated,
"BinanceMargin fetch_trades: empty effective instrument set — returning empty result"
);
return Ok(Vec::new());
}
let start_time_ms = time_since.timestamp_millis();
let is_isolated = self.config.is_isolated;
let mut all_trades = Vec::new();
let mut stream = futures::stream::iter(effective.into_iter().map(|inst| {
let rest = self.rest.clone();
let rate_limiter = self.rate_limiter.clone();
async move {
let pages = paginate_margin_my_trades(
&rest,
&rate_limiter,
&inst,
start_time_ms,
is_isolated,
)
.await?;
Ok::<_, UnindexedClientError>((inst, pages))
}
}))
.buffer_unordered(8);
while let Some(result) = stream.next().await {
let (instrument, trades_data) = result?;
for t in trades_data {
if let Some(trade) = convert_margin_trade(&t, &instrument) {
all_trades.push(trade);
}
}
}
Ok(all_trades)
}
async fn account_stream(
&self,
_assets: &[AssetNameExchange],
instruments: &[InstrumentNameExchange],
) -> Result<Self::AccountStream, UnindexedClientError> {
let (tx, rx) = mpsc::unbounded_channel::<UnindexedAccountEvent>();
let dedup = new_dedup_cache();
let rest = self.rest.clone();
let rest_config = self.rest_config.clone();
let ws_config = self.ws_config.clone();
let rate_limiter = self.rate_limiter.clone();
let cm_handle = if self.config.is_isolated {
let symbols = self.config.isolated_symbols.clone();
debug_assert!(
symbols.iter().all(|i| i.name().len() <= 23),
"instrument name exceeds SmolStr inline capacity: {:?}",
symbols.iter().find(|i| i.name().len() > 23)
);
let assets = fetch_isolated_margin_account_info(
rest.clone(),
rate_limiter.clone(),
symbols.clone(),
)
.await?;
let base_quote = Arc::new(build_base_quote_map(&assets));
for sym in &symbols {
if !base_quote.contains_key(sym) {
warn!(
symbol = %sym.name(),
"BinanceMargin isolated: account info returned no base/quote for configured \
symbol — its live per-pair balance frames will be dropped (fills/orders \
unaffected; balances available via account_snapshot)"
);
}
}
let tokens = acquire_all_isolated_tokens(&rest_config, &rate_limiter, &symbols).await?;
let live =
isolated_connect_and_subscribe(&ws_config, &tx, &dedup, &base_quote, &tokens)
.await
.map_err(|e| {
UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
})?;
tokio::spawn(isolated_connection_manager(
tx,
dedup,
ws_config,
rest_config,
rest,
rate_limiter,
symbols,
base_quote,
Some(live),
))
} else {
let instruments = instruments.to_vec();
debug_assert!(
instruments.iter().all(|i| i.name().len() <= 23),
"instrument name exceeds SmolStr inline capacity: {:?}",
instruments.iter().find(|i| i.name().len() > 23)
);
let initial_token =
acquire_user_listen_token(&rest_config, &rate_limiter, None).await?;
let initial_ws = connect_margin_ws(&ws_config).await.map_err(|e| {
UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
})?;
tokio::spawn(margin_connection_manager(
tx,
dedup,
ws_config,
rest_config,
rest,
rate_limiter,
instruments,
Some((initial_ws, initial_token)),
))
};
let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
let guarded_stream = AbortOnDropStream::new(rx_stream, cm_handle);
Ok(futures::StreamExt::boxed(guarded_stream))
}
}
const TOKEN_RENEW_MARGIN_SECS: i64 = 300; const TOKEN_MIN_LIFETIME_SECS: u64 = 60;
struct UserListenToken {
token: String,
expiration_time_ms: i64,
}
fn build_listen_token_query(
symbol: Option<&InstrumentNameExchange>,
) -> BTreeMap<String, serde_json::Value> {
let mut query = BTreeMap::new();
if let Some(sym) = symbol {
query.insert(
"isIsolated".to_string(),
serde_json::Value::String("TRUE".to_string()),
);
query.insert(
"symbol".to_string(),
serde_json::Value::String(sym.name().to_string()),
);
}
query
}
#[derive(Deserialize)]
struct UserListenTokenResponse {
#[serde(alias = "listenToken")]
token: String,
#[serde(rename = "expirationTime")]
expiration_time: i64,
}
async fn acquire_user_listen_token(
rest_config: &Arc<ConfigurationRestApi>,
rate_limiter: &RateLimitTracker,
symbol: Option<&InstrumentNameExchange>,
) -> Result<UserListenToken, UnindexedClientError> {
let query = build_listen_token_query(symbol);
let response = rest_call_with_retry(rest_config, rate_limiter, |cfg| {
let query = query.clone();
Box::pin(async move {
binance_sdk::common::utils::send_request::<UserListenTokenResponse>(
&cfg,
"/sapi/v1/userListenToken",
reqwest::Method::POST,
query,
BTreeMap::new(),
None,
true, )
.await
})
})
.await
.map_err(connectivity_error)?;
let data = response
.data()
.await
.map_err(|e| connectivity_error(e.into()))?;
Ok(UserListenToken {
token: data.token,
expiration_time_ms: data.expiration_time,
})
}
fn token_renew_after(expiration_time_ms: i64) -> Duration {
let now_ms = Utc::now().timestamp_millis();
let remaining_secs = expiration_time_ms
.saturating_sub(now_ms)
.saturating_div(1_000)
.saturating_sub(TOKEN_RENEW_MARGIN_SECS);
let secs = u64::try_from(remaining_secs).unwrap_or(0);
if secs < TOKEN_MIN_LIFETIME_SECS {
warn!(
computed_secs = secs,
floor_secs = TOKEN_MIN_LIFETIME_SECS,
"BinanceMargin token renewal wait clamped to floor (near-expiry or odd expirationTime)"
);
}
Duration::from_secs(secs.max(TOKEN_MIN_LIFETIME_SECS))
}
async fn connect_margin_ws(
ws_config: &ConfigurationWebsocketApi,
) -> anyhow::Result<Arc<WsApiBase>> {
let ws = WsApiBase::new(ws_config.clone(), vec![]);
tokio::time::timeout(
Duration::from_secs(CONNECT_TIMEOUT_SECS),
ws.clone().connect(),
)
.await
.map_err(|_| {
anyhow::anyhow!("BinanceMargin WS connect timed out after {CONNECT_TIMEOUT_SECS}s")
})??;
Ok(ws)
}
async fn subscribe_listen_token(ws: &Arc<WsApiBase>, token: &str) -> anyhow::Result<()> {
let mut payload = BTreeMap::new();
payload.insert(
"listenToken".to_string(),
serde_json::Value::String(token.to_string()),
);
ws.send_message::<serde_json::Value>(
"userDataStream.subscribe.listenToken",
payload,
WebsocketMessageSendOptions::new(),
)
.await
.map_err(|e| anyhow::anyhow!("userDataStream.subscribe.listenToken failed: {e}"))?;
Ok(())
}
async fn subscribe_listen_token_capture(
ws: &Arc<WsApiBase>,
token: &str,
) -> anyhow::Result<Option<i64>> {
#[derive(Deserialize)]
struct SubscribeAck {
#[serde(rename = "subscriptionId", default)]
subscription_id: Option<i64>,
}
let mut payload = BTreeMap::new();
payload.insert(
"listenToken".to_string(),
serde_json::Value::String(token.to_string()),
);
let result = ws
.send_message::<SubscribeAck>(
"userDataStream.subscribe.listenToken",
payload,
WebsocketMessageSendOptions::new(),
)
.await
.map_err(|e| anyhow::anyhow!("userDataStream.subscribe.listenToken failed: {e}"))?;
let ack = match result {
SendWebsocketMessageResult::Single(resp) => resp.data()?,
SendWebsocketMessageResult::Multiple(responses) => responses
.into_iter()
.next()
.ok_or_else(|| anyhow::anyhow!("empty subscribe ack"))?
.data()?,
};
Ok(ack.subscription_id)
}
fn cross_account_position_handler(
position: Outboundaccountposition,
_subscription_id: Option<i64>,
buf: &mut Vec<UnindexedAccountEvent>,
) {
convert_margin_account_position(position, buf);
}
#[cfg(test)]
fn convert_margin_user_data_events(frame: &str, buf: &mut Vec<UnindexedAccountEvent>) -> bool {
convert_margin_user_data_events_with(frame, buf, &mut cross_account_position_handler)
}
fn convert_margin_user_data_events_with(
frame: &str,
buf: &mut Vec<UnindexedAccountEvent>,
handle_position: &mut impl FnMut(
Outboundaccountposition,
Option<i64>,
&mut Vec<UnindexedAccountEvent>,
),
) -> bool {
use serde_json::value::RawValue;
#[derive(Deserialize)]
struct Envelope<'a> {
#[serde(borrow, default)]
id: Option<&'a RawValue>,
#[serde(borrow, default)]
event: Option<&'a RawValue>,
#[serde(rename = "subscriptionId", default)]
subscription_id: Option<i64>,
}
#[derive(Deserialize)]
struct EventTag<'a> {
#[serde(borrow, default)]
e: Option<&'a str>,
}
let envelope = match serde_json::from_str::<Envelope<'_>>(frame) {
Ok(env) => env,
Err(e) => {
trace!(error = %e, "BinanceMargin WS: skipped unparseable frame");
return false;
}
};
if envelope.id.is_some() {
return false;
}
let Some(event) = envelope.event else {
trace!("BinanceMargin WS: ignoring frame without `event` or `id`");
return false;
};
let subscription_id = envelope.subscription_id;
let event_raw = event.get();
let event_type = serde_json::from_str::<EventTag<'_>>(event_raw)
.ok()
.and_then(|tag| tag.e)
.unwrap_or_default();
match event_type {
"executionReport" => {
match serde_json::from_str::<Executionreport>(event_raw) {
Ok(report) => {
if let Some(ev) = convert_margin_execution_report(report) {
buf.push(ev);
}
}
Err(e) => {
warn!(error = %e, "BinanceMargin: undeserializable executionReport, dropping")
}
}
false
}
"outboundAccountPosition" => {
match serde_json::from_str::<Outboundaccountposition>(event_raw) {
Ok(position) => handle_position(position, subscription_id, buf),
Err(e) => {
warn!(error = %e, "BinanceMargin: undeserializable outboundAccountPosition, dropping")
}
}
false
}
"balanceUpdate" => {
false
}
"userLiabilityChange" => {
match serde_json::from_str::<UserLiabilityChange>(event_raw) {
Ok(c) => {
if tracing::enabled!(tracing::Level::INFO) {
info!(
asset = c.a.as_deref().unwrap_or("?"),
kind = c.t.as_deref().unwrap_or("?"),
principal = c.p.as_deref().unwrap_or("?"),
interest = c.i.as_deref().unwrap_or("?"),
"BinanceMargin userLiabilityChange (observable; not applied to balance state)"
);
}
}
Err(e) => warn!(
error = %e,
"BinanceMargin: undeserializable userLiabilityChange, dropping"
),
}
false
}
"marginLevelStatusChange" => {
if tracing::enabled!(tracing::Level::WARN) {
match serde_json::from_str::<MarginLevelStatusChange>(event_raw) {
Ok(c) => warn!(
margin_level = c.l.as_deref().unwrap_or("?"),
status = c.s.as_deref().unwrap_or("?"),
"BinanceMargin marginLevelStatusChange (liquidation risk; observable, no policy)"
),
Err(e) => warn!(
error = %e,
"BinanceMargin: undeserializable marginLevelStatusChange, dropping"
),
}
}
false
}
"eventStreamTerminated" => {
warn!("BinanceMargin user data stream terminated by exchange, signalling reconnect");
true
}
other => {
trace!(
event_type = other,
"BinanceMargin ignoring unhandled user data event"
);
false
}
}
}
#[allow(clippy::cognitive_complexity)] fn convert_margin_execution_report(report: Executionreport) -> Option<UnindexedAccountEvent> {
let exec_type = match report.x.as_deref() {
Some(t) => t,
None => {
warn!("BinanceMargin executionReport missing execution type (x), dropping");
return None;
}
};
let symbol = match report.s.as_deref() {
Some(s) => InstrumentNameExchange::new(s),
None => {
warn!("BinanceMargin executionReport missing symbol (s), dropping");
return None;
}
};
let order_id = match report.i {
Some(id) => OrderId(format_smolstr!("{id}")),
None => {
warn!(%symbol, "BinanceMargin executionReport missing orderId (i), dropping");
return None;
}
};
let cid = match report.c.as_deref() {
Some(c) => ClientOrderId::new(c),
None => ClientOrderId::new(order_id.0.as_str()),
};
let time_exchange = match report
.t_uppercase
.and_then(|ms| Utc.timestamp_millis_opt(ms).single())
{
Some(t) => t,
None => {
warn!(%symbol, "BinanceMargin executionReport missing/unparseable transaction time (T), using now");
Utc::now()
}
};
match exec_type {
"NEW" => convert_margin_new_order(&report, symbol, cid, order_id, time_exchange),
"TRADE" => {
let trade_id = match report.t {
Some(id) => TradeId(format_smolstr!("{id}")),
None => {
warn!(%symbol, "BinanceMargin TRADE event missing trade ID (t), dropping");
return None;
}
};
let side = match report.s_uppercase.as_deref().and_then(parse_side) {
Some(s) => s,
None => {
warn!(%symbol, "BinanceMargin TRADE event missing/unknown side (S), dropping");
return None;
}
};
let last_price = match report.l_uppercase.as_deref() {
Some(s) => match Decimal::from_str(s) {
Ok(v) => v,
Err(e) => {
warn!(%symbol, error = %e, raw = s, "BinanceMargin TRADE event unparseable last price (L), dropping fill");
return None;
}
},
None => {
warn!(%symbol, "BinanceMargin TRADE event missing last price (L), dropping fill");
return None;
}
};
let last_qty = match report.l.as_deref() {
Some(s) => match Decimal::from_str(s) {
Ok(v) => v,
Err(e) => {
warn!(%symbol, error = %e, raw = s, "BinanceMargin TRADE event unparseable last qty (l), dropping fill");
return None;
}
},
None => {
warn!(%symbol, "BinanceMargin TRADE event missing last qty (l), dropping fill");
return None;
}
};
let commission = match Decimal::from_str(report.n.as_deref().unwrap_or("0")) {
Ok(v) => v,
Err(e) => {
warn!(%symbol, error = %e, "BinanceMargin TRADE event unparseable commission (n), defaulting to 0");
Decimal::ZERO
}
};
let fee_asset = report
.n_uppercase
.as_deref()
.map(AssetNameExchange::from)
.unwrap_or_else(|| AssetNameExchange::from("UNKNOWN"));
let trade = Trade::new(
trade_id,
order_id,
symbol,
StrategyId::unknown(), time_exchange,
side,
last_price,
last_qty,
AssetFees::new(fee_asset, commission, None),
);
Some(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::Trade(trade),
))
}
"CANCELED" | "EXPIRED" | "EXPIRED_IN_MATCH" => {
let filled_qty = report
.z
.as_deref()
.and_then(|s| Decimal::from_str(s).ok())
.unwrap_or(Decimal::ZERO);
let response = UnindexedOrderResponseCancel {
key: OrderKey::new(
ExchangeId::BinanceMargin,
symbol,
StrategyId::unknown(),
cid,
),
state: Ok(Cancelled::new(order_id, time_exchange, filled_qty)),
};
Some(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::OrderCancelled(response),
))
}
"REJECTED" => {
let reject_reason = report.r.unwrap_or_else(|| "unknown".to_string());
warn!(%symbol, %order_id, reason = %reject_reason, "BinanceMargin order REJECTED by matching engine");
let response = UnindexedOrderResponseCancel {
key: OrderKey::new(
ExchangeId::BinanceMargin,
symbol,
StrategyId::unknown(),
cid,
),
state: Err(UnindexedOrderError::Rejected(ApiError::OrderRejected(
reject_reason,
))),
};
Some(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::OrderCancelled(response),
))
}
"REPLACE" => {
let filled_qty = report
.z
.as_deref()
.and_then(|s| Decimal::from_str(s).ok())
.unwrap_or(Decimal::ZERO);
let response = UnindexedOrderResponseCancel {
key: OrderKey::new(
ExchangeId::BinanceMargin,
symbol,
StrategyId::unknown(),
cid,
),
state: Ok(Cancelled::new(order_id, time_exchange, filled_qty)),
};
Some(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::OrderCancelled(response),
))
}
_ => {
trace!(exec_type, "BinanceMargin ignoring execution type");
None
}
}
}
fn convert_margin_new_order(
report: &Executionreport,
symbol: InstrumentNameExchange,
cid: ClientOrderId,
order_id: OrderId,
time_exchange: DateTime<Utc>,
) -> Option<UnindexedAccountEvent> {
let side = match report.s_uppercase.as_deref().and_then(parse_side) {
Some(s) => s,
None => {
warn!(%symbol, "BinanceMargin NEW event missing/unknown side (S), dropping");
return None;
}
};
let kind = parse_order_kind(report.o.as_deref().unwrap_or("LIMIT"))?;
let price: Option<Decimal> = match (report.p.as_deref(), &kind) {
(Some(p), _) => match Decimal::from_str(p) {
Ok(v) if !v.is_zero() => Some(v),
Ok(_) => {
if matches!(
kind,
OrderKind::Limit
| OrderKind::StopLimit { .. }
| OrderKind::TakeProfitLimit { .. }
| OrderKind::TrailingStopLimit { .. }
) {
trace!(%symbol, %kind, "BinanceMargin NEW event has zero price (p) on limit-type order, treating as no limit price");
}
None
}
Err(e) => {
warn!(%symbol, price = p, error = %e, "BinanceMargin NEW event unparseable price (p), dropping");
return None;
}
},
(
None,
OrderKind::Market
| OrderKind::Stop { .. }
| OrderKind::TakeProfit { .. }
| OrderKind::TrailingStop { .. },
) => None,
(
None,
OrderKind::Limit
| OrderKind::StopLimit { .. }
| OrderKind::TakeProfitLimit { .. }
| OrderKind::TrailingStopLimit { .. },
) => {
warn!(%symbol, "BinanceMargin NEW limit-type order missing price (p), dropping");
return None;
}
};
let quantity = match report.q.as_deref() {
Some(q) => match Decimal::from_str(q) {
Ok(v) => v,
Err(e) => {
warn!(%symbol, qty = q, error = %e, "BinanceMargin NEW event unparseable quantity (q), dropping");
return None;
}
},
None => {
warn!(%symbol, "BinanceMargin NEW order missing quantity (q), dropping");
return None;
}
};
let time_in_force = parse_time_in_force(report.f.as_deref().unwrap_or("GTC"));
let filled_qty = report
.z
.as_deref()
.and_then(|s| Decimal::from_str(s).ok())
.unwrap_or(Decimal::ZERO);
let order = Order {
key: OrderKey::new(
ExchangeId::BinanceMargin,
symbol,
StrategyId::unknown(),
cid,
),
side,
price,
quantity,
kind,
time_in_force,
state: OrderState::active(Open::new(order_id, time_exchange, filled_qty)),
};
Some(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::OrderSnapshot(rustrade_integration::collection::snapshot::Snapshot::new(
order,
)),
))
}
fn convert_margin_account_position(
position: Outboundaccountposition,
buf: &mut Vec<UnindexedAccountEvent>,
) {
let time_exchange = position
.u
.and_then(|ms| Utc.timestamp_millis_opt(ms).single())
.unwrap_or_else(Utc::now);
for b in position.b_uppercase.unwrap_or_default() {
let asset = match b.a {
Some(a) => AssetNameExchange::new(a),
None => {
warn!("BinanceMargin account position entry missing asset name");
continue;
}
};
let free = match b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%asset, "BinanceMargin account position missing/unparseable 'free' field");
continue;
}
};
let locked = match b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%asset, "BinanceMargin account position missing/unparseable 'locked' field");
continue;
}
};
let update =
AssetBalanceUpdate::new(asset, BalanceUpdate::new(free, locked), time_exchange);
buf.push(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::BalanceStreamUpdate(
rustrade_integration::collection::snapshot::Snapshot::new(update),
),
));
}
}
fn route_isolated_account_position(
position: Outboundaccountposition,
subscription_id: Option<i64>,
sub_map: &Mutex<HashMap<i64, InstrumentNameExchange>>,
base_quote: &HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>,
buf: &mut Vec<UnindexedAccountEvent>,
) {
let Some(sub_id) = subscription_id else {
warn!(
"BinanceMargin isolated: outboundAccountPosition without subscriptionId — dropping \
(per-instrument balance routing requires it)"
);
return;
};
let instrument = {
let map = sub_map.lock().unwrap_or_else(|p| p.into_inner());
match map.get(&sub_id) {
Some(inst) => inst.clone(),
None => {
warn!(
subscription_id = sub_id,
"BinanceMargin isolated: outboundAccountPosition for unmapped subscriptionId — dropping"
);
return;
}
}
};
let Some((base_asset, quote_asset)) = base_quote.get(&instrument) else {
warn!(
instrument = %instrument.name(),
"BinanceMargin isolated: no base/quote known for instrument — dropping balance frame"
);
return;
};
let time_exchange = position
.u
.and_then(|ms| Utc.timestamp_millis_opt(ms).single())
.unwrap_or_else(Utc::now);
let mut base_update: Option<AssetBalanceUpdate<AssetNameExchange>> = None;
let mut quote_update: Option<AssetBalanceUpdate<AssetNameExchange>> = None;
for b in position.b_uppercase.unwrap_or_default() {
let Some(asset) = b.a.as_deref() else {
warn!(instrument = %instrument.name(), "BinanceMargin isolated account position entry missing asset name");
continue;
};
let side = if asset == base_asset.name().as_str() {
&mut base_update
} else if asset == quote_asset.name().as_str() {
&mut quote_update
} else {
continue;
};
let (Some(free), Some(locked)) = (
b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()),
b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()),
) else {
warn!(%asset, instrument = %instrument.name(), "BinanceMargin isolated account position missing/unparseable free/locked");
continue;
};
*side = Some(AssetBalanceUpdate::new(
AssetNameExchange::new(asset),
BalanceUpdate::new(free, locked),
time_exchange,
));
}
let (Some(base), Some(quote)) = (base_update, quote_update) else {
warn!(
instrument = %instrument.name(),
"BinanceMargin isolated: outboundAccountPosition missing base or quote side — dropping (no partial InstrumentBalanceUpdate)"
);
return;
};
buf.push(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::InstrumentBalanceUpdate(InstrumentBalanceUpdate::new(
instrument, base, quote,
)),
));
}
fn register_user_data_listener(
ws: &Arc<WsApiBase>,
tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: SharedDedupCache,
heartbeat_flag: Arc<AtomicBool>,
signal_tx: oneshot::Sender<()>,
mut handle_position: impl FnMut(
Outboundaccountposition,
Option<i64>,
&mut Vec<UnindexedAccountEvent>,
) + Send
+ 'static,
) -> Subscription {
let mut signal_tx_opt = Some(signal_tx);
let mut event_tx = Some(tx);
let mut event_buf = Vec::with_capacity(32);
ws.common.events.subscribe(move |event| {
let Some(ref sender) = event_tx else { return };
match event {
WebsocketEvent::Message(json_str) => {
heartbeat_flag.store(true, Ordering::Release);
let terminated = convert_margin_user_data_events_with(
&json_str,
&mut event_buf,
&mut handle_position,
);
for ev in event_buf.drain(..) {
if let Some(key) = dedup_key_from_event(&ev)
&& is_duplicate(&dedup, key)
{
trace!("BinanceMargin dedup: skipping duplicate event");
continue;
}
if sender.send(ev).is_err() {
warn!("BinanceMargin account_stream receiver dropped, suppressing sends");
event_tx.take();
if let Some(s) = signal_tx_opt.take() {
let _ = s.send(());
}
return;
}
}
if terminated {
event_tx.take();
if let Some(s) = signal_tx_opt.take() {
let _ = s.send(());
}
}
}
WebsocketEvent::Ping | WebsocketEvent::Pong => {
heartbeat_flag.store(true, Ordering::Release);
}
WebsocketEvent::Error(e) => {
warn!(%e, "BinanceMargin WebSocket error, will attempt reconnect");
event_tx.take();
if let Some(s) = signal_tx_opt.take() {
let _ = s.send(());
}
}
WebsocketEvent::Close(code, reason) => {
warn!(code, %reason, "BinanceMargin WebSocket closed");
event_tx.take();
if let Some(s) = signal_tx_opt.take() {
let _ = s.send(());
}
}
_ => {
trace!("BinanceMargin ignoring unhandled WebsocketEvent variant");
}
}
})
}
async fn recover_margin_fills(
rest: &Arc<RestApi>,
rate_limiter: &Arc<RateLimitTracker>,
instruments: &[InstrumentNameExchange],
disconnect_time: DateTime<Utc>,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: &SharedDedupCache,
is_isolated: bool,
) {
use futures::StreamExt as _;
if instruments.is_empty() {
debug!("BinanceMargin recover_fills: empty instruments — no fills recovered");
return;
}
info!(
since = %disconnect_time,
instruments = instruments.len(),
"BinanceMargin recovering fills after reconnect"
);
let start_time_ms = disconnect_time.timestamp_millis();
let mut recovered = 0u32;
let mut duplicates = 0u32;
let mut failed_instruments = 0u32;
let mut stream = futures::stream::iter(instruments.iter().cloned().map(|inst| {
let rest = rest.clone();
let rl = rate_limiter.clone();
async move {
match paginate_margin_my_trades(&rest, &rl, &inst, start_time_ms, is_isolated).await {
Ok(pages) => Some(
pages
.iter()
.filter_map(|t| convert_margin_trade(t, &inst))
.collect::<Vec<_>>(),
),
Err(e) => {
warn!(%e, %inst, "BinanceMargin fill recovery: REST request failed");
None
}
}
}
}))
.buffer_unordered(8);
while let Some(result) = stream.next().await {
let trades = match result {
Some(t) => t,
None => {
failed_instruments += 1;
continue;
}
};
for trade in trades {
let event = UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::Trade(trade),
);
if let Some(key) = dedup_key_from_event(&event)
&& is_duplicate(dedup, key)
{
duplicates += 1;
continue;
}
if tx.send(event).is_err() {
debug!("BinanceMargin fill recovery: consumer dropped during recovery");
return;
}
recovered += 1;
}
}
if failed_instruments > 0 {
error!(
recovered,
duplicates,
failed_instruments,
"BinanceMargin fill recovery complete with failures — some fills may be permanently missed"
);
} else {
info!(
recovered,
duplicates, "BinanceMargin fill recovery complete"
);
}
}
#[allow(
clippy::cognitive_complexity,
reason = "inherent reconnect-loop complexity (token + connect + subscribe + callback + recovery \
+ monitor + cleanup + backoff); mirrors spot's `connection_manager`, not worth splitting further"
)]
async fn margin_connection_manager(
tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: SharedDedupCache,
ws_config: ConfigurationWebsocketApi,
rest_config: Arc<ConfigurationRestApi>,
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
instruments: Vec<InstrumentNameExchange>,
initial: Option<(Arc<WsApiBase>, UserListenToken)>,
) {
enum DisconnectReason {
Signal,
HeartbeatTimeout,
TokenRefresh,
ConsumerDropped,
}
let mut backoff = ExponentialBackoff::new();
let mut disconnect_time: Option<DateTime<Utc>> = None;
let (mut current_ws, mut current_token) = match initial {
Some((ws, token)) => (Some(ws), Some(token)),
None => (None, None),
};
loop {
let token = match current_token.take() {
Some(t) => t,
None => match acquire_user_listen_token(&rest_config, &rate_limiter, None).await {
Ok(t) => t,
Err(e) => {
error!(%e, "BinanceMargin userListenToken acquisition failed");
if !backoff.wait().await {
error!("BinanceMargin max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
},
};
let ws = match current_ws.take() {
Some(ws) => ws,
None => match connect_margin_ws(&ws_config).await {
Ok(ws) => ws,
Err(e) => {
error!(%e, "BinanceMargin WS connect failed");
if !backoff.wait().await {
error!("BinanceMargin max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
},
};
let (signal_tx, signal_rx) = oneshot::channel::<()>();
let heartbeat_flag = Arc::new(AtomicBool::new(true));
let subscription = register_user_data_listener(
&ws,
tx.clone(),
dedup.clone(),
heartbeat_flag.clone(),
signal_tx,
cross_account_position_handler,
);
if let Err(e) = subscribe_listen_token(&ws, &token.token).await {
warn!(%e, "BinanceMargin user-data subscribe failed, cleaning up and retrying");
subscription.unsubscribe();
if let Err(de) = ws.disconnect().await {
warn!(%de, "BinanceMargin failed to disconnect after subscribe failure");
}
if !backoff.wait().await {
error!("BinanceMargin max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
info!("BinanceMargin account_stream connected and subscribed");
backoff.reset();
let token_deadline =
tokio::time::Instant::now() + token_renew_after(token.expiration_time_ms);
if let Some(dt) = disconnect_time.take()
&& tokio::time::timeout(
Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS),
recover_margin_fills(&rest, &rate_limiter, &instruments, dt, &tx, &dedup, false),
)
.await
.is_err()
{
warn!(
timeout_secs = FILL_RECOVERY_TIMEOUT_SECS,
"BinanceMargin fill recovery timed out — remaining instruments not queried"
);
}
let reason = {
let mut signal_rx = signal_rx;
loop {
tokio::select! {
biased;
_ = tx.closed() => {
debug!("BinanceMargin account_stream consumer dropped, terminating");
break DisconnectReason::ConsumerDropped;
}
_ = &mut signal_rx => {
warn!("BinanceMargin WS disconnected, will attempt reconnect");
break DisconnectReason::Signal;
}
() = tokio::time::sleep_until(token_deadline) => {
info!("BinanceMargin userListenToken nearing expiry, renewing");
break DisconnectReason::TokenRefresh;
}
() = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
if heartbeat_flag.swap(false, Ordering::AcqRel) {
continue;
}
warn!("BinanceMargin heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
break DisconnectReason::HeartbeatTimeout;
}
}
}
};
let should_reconnect = !matches!(reason, DisconnectReason::ConsumerDropped);
if should_reconnect {
disconnect_time = Some(match reason {
DisconnectReason::HeartbeatTimeout => {
Utc::now()
- chrono::Duration::seconds(HEARTBEAT_TIMEOUT_SECS as i64)
- chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
}
DisconnectReason::TokenRefresh => {
Utc::now()
- chrono::Duration::seconds(CONNECT_TIMEOUT_SECS as i64)
- chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
}
_ => Utc::now() - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS),
});
}
subscription.unsubscribe();
if let Err(e) = ws.disconnect().await {
warn!(%e, "BinanceMargin failed to disconnect WebSocket");
}
if !should_reconnect || tx.is_closed() {
debug!("BinanceMargin connection manager exiting");
break;
}
if !matches!(reason, DisconnectReason::TokenRefresh) && !backoff.wait().await {
error!("BinanceMargin max reconnect attempts exhausted, stream terminating");
let last_error = match reason {
DisconnectReason::HeartbeatTimeout => {
format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
}
_ => "WebSocket disconnected (server close/error)".to_string(),
};
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error,
},
);
break;
}
}
}
fn build_base_quote_map(
assets: &[QueryIsolatedMarginAccountInfoResponseAssetsInner],
) -> HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)> {
let mut map = HashMap::with_capacity(assets.len());
for entry in assets {
let (Some(symbol), Some(base), Some(quote)) = (
entry.symbol.as_deref(),
entry.base_asset.as_deref().and_then(|b| b.asset.as_deref()),
entry
.quote_asset
.as_deref()
.and_then(|q| q.asset.as_deref()),
) else {
warn!(
"BinanceMargin isolated: account-info entry missing symbol/base/quote asset — \
skipping base/quote map entry"
);
continue;
};
map.insert(
InstrumentNameExchange::new(symbol),
(AssetNameExchange::new(base), AssetNameExchange::new(quote)),
);
}
map
}
async fn acquire_all_isolated_tokens(
rest_config: &Arc<ConfigurationRestApi>,
rate_limiter: &Arc<RateLimitTracker>,
symbols: &[InstrumentNameExchange],
) -> Result<Vec<(InstrumentNameExchange, UserListenToken)>, UnindexedClientError> {
use futures::{StreamExt as _, TryStreamExt as _};
futures::stream::iter(symbols.iter().cloned().map(|sym| {
let rest_config = rest_config.clone();
let rate_limiter = rate_limiter.clone();
async move {
let token = acquire_user_listen_token(&rest_config, &rate_limiter, Some(&sym)).await?;
Ok::<_, UnindexedClientError>((sym, token))
}
}))
.buffer_unordered(8)
.try_collect()
.await
}
fn earliest_token_expiry_ms(tokens: &[(InstrumentNameExchange, UserListenToken)]) -> i64 {
tokens
.iter()
.map(|(_, t)| t.expiration_time_ms)
.min()
.unwrap_or(i64::MAX)
}
struct IsolatedLiveConn {
ws: Arc<WsApiBase>,
subscription: Subscription,
signal_rx: oneshot::Receiver<()>,
heartbeat_flag: Arc<AtomicBool>,
earliest_expiry_ms: i64,
}
async fn isolated_connect_and_subscribe(
ws_config: &ConfigurationWebsocketApi,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: &SharedDedupCache,
base_quote: &Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
tokens: &[(InstrumentNameExchange, UserListenToken)],
) -> anyhow::Result<IsolatedLiveConn> {
let ws = connect_margin_ws(ws_config).await?;
let sub_map = Arc::new(Mutex::new(
HashMap::<i64, InstrumentNameExchange>::with_capacity(tokens.len()),
));
let (signal_tx, signal_rx) = oneshot::channel::<()>();
let heartbeat_flag = Arc::new(AtomicBool::new(true));
let handle_position = {
let sub_map = sub_map.clone();
let base_quote = base_quote.clone();
move |position, subscription_id, buf: &mut Vec<UnindexedAccountEvent>| {
route_isolated_account_position(position, subscription_id, &sub_map, &base_quote, buf);
}
};
let subscription = register_user_data_listener(
&ws,
tx.clone(),
dedup.clone(),
heartbeat_flag.clone(),
signal_tx,
handle_position,
);
let earliest_expiry_ms = earliest_token_expiry_ms(tokens);
for (sym, token) in tokens {
match subscribe_listen_token_capture(&ws, &token.token).await {
Ok(Some(id)) => {
sub_map
.lock()
.unwrap_or_else(|p| p.into_inner())
.insert(id, sym.clone());
}
Ok(None) => warn!(
symbol = %sym.name(),
"BinanceMargin isolated: subscribe ack carried no subscriptionId — live per-pair \
balances unavailable for this symbol (fills/orders unaffected; balances via snapshot)"
),
Err(e) => {
warn!(%e, symbol = %sym.name(), "BinanceMargin isolated subscribe failed");
subscription.unsubscribe();
if let Err(de) = ws.disconnect().await {
warn!(%de, "BinanceMargin isolated: disconnect after subscribe failure failed");
}
return Err(e);
}
}
}
Ok(IsolatedLiveConn {
ws,
subscription,
signal_rx,
heartbeat_flag,
earliest_expiry_ms,
})
}
#[allow(
clippy::cognitive_complexity,
reason = "inherent reconnect-loop complexity (tokens + connect + subscribe + monitor + cleanup \
+ backoff); mirrors the cross manager, not worth splitting further"
)]
async fn isolated_connection_manager(
tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: SharedDedupCache,
ws_config: ConfigurationWebsocketApi,
rest_config: Arc<ConfigurationRestApi>,
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
symbols: Vec<InstrumentNameExchange>,
base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
initial: Option<IsolatedLiveConn>,
) {
enum DisconnectReason {
Signal,
HeartbeatTimeout,
TokenRefresh,
ConsumerDropped,
}
let mut backoff = ExponentialBackoff::new();
let mut disconnect_time: Option<DateTime<Utc>> = None;
let mut current = initial;
loop {
let IsolatedLiveConn {
ws,
subscription,
signal_rx,
heartbeat_flag,
earliest_expiry_ms,
} = match current.take() {
Some(live) => live,
None => {
let tokens = match acquire_all_isolated_tokens(
&rest_config,
&rate_limiter,
&symbols,
)
.await
{
Ok(t) => t,
Err(e) => {
error!(%e, "BinanceMargin isolated userListenToken acquisition failed");
if !backoff.wait().await {
error!("BinanceMargin isolated max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
};
match isolated_connect_and_subscribe(&ws_config, &tx, &dedup, &base_quote, &tokens)
.await
{
Ok(live) => live,
Err(e) => {
error!(%e, "BinanceMargin isolated connect/subscribe failed");
if !backoff.wait().await {
error!("BinanceMargin isolated max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
}
}
};
info!(
symbols = symbols.len(),
"BinanceMargin isolated account_stream connected and subscribed"
);
backoff.reset();
let token_deadline = tokio::time::Instant::now() + token_renew_after(earliest_expiry_ms);
if let Some(dt) = disconnect_time.take()
&& tokio::time::timeout(
Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS),
recover_margin_fills(&rest, &rate_limiter, &symbols, dt, &tx, &dedup, true),
)
.await
.is_err()
{
warn!(
timeout_secs = FILL_RECOVERY_TIMEOUT_SECS,
"BinanceMargin isolated fill recovery timed out — remaining instruments not queried"
);
}
let reason = {
let mut signal_rx = signal_rx;
loop {
tokio::select! {
biased;
_ = tx.closed() => {
debug!("BinanceMargin isolated consumer dropped, terminating");
break DisconnectReason::ConsumerDropped;
}
_ = &mut signal_rx => {
warn!("BinanceMargin isolated WS disconnected, will attempt reconnect");
break DisconnectReason::Signal;
}
() = tokio::time::sleep_until(token_deadline) => {
info!("BinanceMargin isolated userListenToken nearing expiry, renewing all");
break DisconnectReason::TokenRefresh;
}
() = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
if heartbeat_flag.swap(false, Ordering::AcqRel) {
continue;
}
warn!("BinanceMargin isolated heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
break DisconnectReason::HeartbeatTimeout;
}
}
}
};
let should_reconnect = !matches!(reason, DisconnectReason::ConsumerDropped);
if should_reconnect {
disconnect_time = Some(match reason {
DisconnectReason::HeartbeatTimeout => {
Utc::now()
- chrono::Duration::seconds(HEARTBEAT_TIMEOUT_SECS as i64)
- chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
}
DisconnectReason::TokenRefresh => {
Utc::now()
- chrono::Duration::seconds(CONNECT_TIMEOUT_SECS as i64)
- chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
}
_ => Utc::now() - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS),
});
}
subscription.unsubscribe();
if let Err(e) = ws.disconnect().await {
warn!(%e, "BinanceMargin isolated failed to disconnect WebSocket");
}
if !should_reconnect || tx.is_closed() {
debug!("BinanceMargin isolated connection manager exiting");
break;
}
if !matches!(reason, DisconnectReason::TokenRefresh) && !backoff.wait().await {
error!("BinanceMargin isolated max reconnect attempts exhausted, stream terminating");
let last_error = match reason {
DisconnectReason::HeartbeatTimeout => {
format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
}
_ => "WebSocket disconnected (server close/error)".to_string(),
};
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error,
},
);
break;
}
}
}
async fn fetch_margin_open_orders_for_instrument(
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
instrument: InstrumentNameExchange,
is_isolated: bool,
) -> Result<
(
InstrumentNameExchange,
Vec<Order<ExchangeId, InstrumentNameExchange, Open>>,
),
UnindexedClientError,
> {
let symbol_str = instrument.name().to_string();
let isolated = isolated_str(is_isolated);
let response = rest_call_with_retry(&rest, &rate_limiter, |rest| {
let sym = symbol_str.clone();
let isolated = isolated.clone();
Box::pin(async move {
let params = QueryMarginAccountsOpenOrdersParams::builder()
.symbol(sym)
.is_isolated(isolated)
.build()?;
rest.query_margin_accounts_open_orders(params).await
})
})
.await
.map_err(connectivity_error)?;
let orders_data = response
.data()
.await
.map_err(|e| connectivity_error(e.into()))?;
let orders = orders_data
.into_iter()
.filter_map(|o| convert_margin_open_order(&o, &instrument))
.collect();
Ok((instrument, orders))
}
async fn fetch_margin_all_open_orders(
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
is_isolated: bool,
) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
let isolated = isolated_str(is_isolated);
let response = rest_call_with_retry(&rest, &rate_limiter, |rest| {
let isolated = isolated.clone();
Box::pin(async move {
let params = QueryMarginAccountsOpenOrdersParams::builder()
.is_isolated(isolated)
.build()?;
rest.query_margin_accounts_open_orders(params).await
})
})
.await
.map_err(connectivity_error)?;
let orders_data = response
.data()
.await
.map_err(|e| connectivity_error(e.into()))?;
let orders = orders_data
.into_iter()
.filter_map(|o| convert_margin_open_order_owned_symbol(&o))
.collect();
Ok(orders)
}
async fn paginate_margin_my_trades(
rest: &Arc<RestApi>,
rate_limiter: &Arc<RateLimitTracker>,
instrument: &InstrumentNameExchange,
start_time_ms: i64,
is_isolated: bool,
) -> Result<Vec<QueryMarginAccountsTradeListResponseInner>, UnindexedClientError> {
let symbol_str = instrument.name().to_string();
let isolated = isolated_str(is_isolated);
#[allow(clippy::cast_possible_wrap)]
let limit = BINANCE_MAX_TRADES as i64;
let mut all_pages = Vec::new();
let mut cursor: Option<i64> = None;
loop {
let fid = cursor; let response = rest_call_with_retry(rest, rate_limiter, |rest| {
let sym = symbol_str.clone();
let isolated = isolated.clone();
let stm = start_time_ms;
Box::pin(async move {
let builder = QueryMarginAccountsTradeListParams::builder(sym)
.is_isolated(isolated)
.limit(limit);
let params = if let Some(id) = fid {
builder.from_id(id).build()?
} else {
builder.start_time(stm).build()?
};
rest.query_margin_accounts_trade_list(params).await
})
})
.await
.map_err(connectivity_error)?;
let page = response
.data()
.await
.map_err(|e| connectivity_error(e.into()))?;
let page_len = page.len();
let last_id = page.last().and_then(|t| t.id);
all_pages.extend(page);
if page_len < BINANCE_MAX_TRADES {
break;
}
match last_id {
Some(id) => {
debug!(%instrument, "BinanceMargin paginate_my_trades: fetching next page ({page_len} results)");
match id.checked_add(1) {
Some(next) => cursor = Some(next),
None => break, }
}
None => {
warn!(%instrument, "BinanceMargin paginate_my_trades: trade missing ID, stopping pagination");
break;
}
}
}
Ok(all_pages)
}
fn convert_margin_balance_entry(
b: QueryCrossMarginAccountDetailsResponseUserAssetsInner,
now: DateTime<Utc>,
) -> Option<AssetBalance<AssetNameExchange>> {
let asset_name = AssetNameExchange::new(b.asset.as_deref()?);
let free = match b.free.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%asset_name, "BinanceMargin balance missing/unparseable 'free' field");
return None;
}
};
let locked = match b.locked.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%asset_name, "BinanceMargin balance missing/unparseable 'locked' field");
return None;
}
};
let borrowed = b
.borrowed
.as_deref()
.and_then(|s| Decimal::from_str(s).ok())
.unwrap_or(Decimal::ZERO);
let interest = b
.interest
.as_deref()
.and_then(|s| Decimal::from_str(s).ok())
.unwrap_or(Decimal::ZERO);
Some(AssetBalance::new(
asset_name,
Balance::new_margin(free + locked, free, borrowed, interest),
now,
))
}
fn filter_and_convert_margin_balances(
user_assets: Vec<QueryCrossMarginAccountDetailsResponseUserAssetsInner>,
assets: &[AssetNameExchange],
) -> Vec<AssetBalance<AssetNameExchange>> {
let now = Utc::now();
if assets.is_empty() {
return user_assets
.into_iter()
.filter_map(|b| convert_margin_balance_entry(b, now))
.collect();
}
if assets.len() <= 16 {
return user_assets
.into_iter()
.filter_map(|b| {
let asset_name_str = b.asset.as_deref()?;
if !assets.iter().any(|a| a.name().as_str() == asset_name_str) {
return None;
}
convert_margin_balance_entry(b, now)
})
.collect();
}
use std::collections::HashSet;
let asset_set: HashSet<&str> = assets.iter().map(|a| a.name().as_str()).collect();
user_assets
.into_iter()
.filter_map(|b| {
let asset_name_str = b.asset.as_deref()?;
if !asset_set.contains(asset_name_str) {
return None;
}
convert_margin_balance_entry(b, now)
})
.collect()
}
fn active_order_snapshot(
o: Order<ExchangeId, InstrumentNameExchange, Open>,
) -> Order<ExchangeId, InstrumentNameExchange, OrderState<AssetNameExchange, InstrumentNameExchange>>
{
Order {
key: o.key,
side: o.side,
price: o.price,
quantity: o.quantity,
kind: o.kind,
time_in_force: o.time_in_force,
state: OrderState::active(o.state),
}
}
fn convert_isolated_asset_balance(
asset: Option<&str>,
free: Option<&str>,
locked: Option<&str>,
borrowed: Option<&str>,
interest: Option<&str>,
now: DateTime<Utc>,
) -> Option<AssetBalance<AssetNameExchange>> {
let asset_name = AssetNameExchange::new(asset?);
let free = match free.and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%asset_name, "BinanceMargin isolated balance missing/unparseable 'free' field");
return None;
}
};
let locked = match locked.and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%asset_name, "BinanceMargin isolated balance missing/unparseable 'locked' field");
return None;
}
};
let borrowed = borrowed
.and_then(|s| Decimal::from_str(s).ok())
.unwrap_or(Decimal::ZERO);
let interest = interest
.and_then(|s| Decimal::from_str(s).ok())
.unwrap_or(Decimal::ZERO);
Some(AssetBalance::new(
asset_name,
Balance::new_margin(free + locked, free, borrowed, interest),
now,
))
}
fn convert_isolated_margin_assets(
assets: Vec<QueryIsolatedMarginAccountInfoResponseAssetsInner>,
) -> HashMap<InstrumentNameExchange, IsolatedInstrumentState<AssetNameExchange>> {
let now = Utc::now();
let mut map = HashMap::with_capacity(assets.len());
for entry in assets {
let Some(symbol) = entry.symbol.as_deref() else {
warn!("BinanceMargin isolated asset entry missing 'symbol' — skipping");
continue;
};
let instrument = InstrumentNameExchange::new(symbol);
let Some(base_raw) = entry.base_asset.as_deref() else {
warn!(%instrument, "BinanceMargin isolated entry missing baseAsset — skipping");
continue;
};
let Some(quote_raw) = entry.quote_asset.as_deref() else {
warn!(%instrument, "BinanceMargin isolated entry missing quoteAsset — skipping");
continue;
};
let Some(base) = convert_isolated_asset_balance(
base_raw.asset.as_deref(),
base_raw.free.as_deref(),
base_raw.locked.as_deref(),
base_raw.borrowed.as_deref(),
base_raw.interest.as_deref(),
now,
) else {
warn!(%instrument, "BinanceMargin isolated baseAsset missing required field — skipping");
continue;
};
let Some(quote) = convert_isolated_asset_balance(
quote_raw.asset.as_deref(),
quote_raw.free.as_deref(),
quote_raw.locked.as_deref(),
quote_raw.borrowed.as_deref(),
quote_raw.interest.as_deref(),
now,
) else {
warn!(%instrument, "BinanceMargin isolated quoteAsset missing required field — skipping");
continue;
};
let risk = IsolatedMarginRisk {
margin_level: entry
.margin_level
.as_deref()
.and_then(|s| Decimal::from_str(s).ok()),
margin_ratio: entry
.margin_ratio
.as_deref()
.and_then(|s| Decimal::from_str(s).ok()),
liquidation_price: entry
.liquidate_price
.as_deref()
.and_then(|s| Decimal::from_str(s).ok()),
};
map.insert(instrument, IsolatedInstrumentState { base, quote, risk });
}
map
}
fn chunk_symbols(symbols: &[InstrumentNameExchange]) -> Vec<String> {
symbols
.chunks(5)
.map(|chunk| {
chunk
.iter()
.map(|s| s.name().as_str())
.collect::<Vec<_>>()
.join(",")
})
.collect()
}
async fn fetch_isolated_margin_account_info(
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
symbols: Vec<InstrumentNameExchange>,
) -> Result<Vec<QueryIsolatedMarginAccountInfoResponseAssetsInner>, UnindexedClientError> {
use futures::{StreamExt as _, TryStreamExt as _};
let chunks = chunk_symbols(&symbols);
futures::stream::iter(chunks.into_iter().map(|symbols_param| {
let rest = rest.clone();
let rate_limiter = rate_limiter.clone();
async move {
let response = rest_call_with_retry(&rest, &rate_limiter, |rest| {
let symbols_param = symbols_param.clone();
Box::pin(async move {
let params = QueryIsolatedMarginAccountInfoParams::builder()
.symbols(symbols_param)
.build()?;
rest.query_isolated_margin_account_info(params).await
})
})
.await
.map_err(connectivity_error)?;
let info = response
.data()
.await
.map_err(|e| connectivity_error(e.into()))?;
Ok::<_, UnindexedClientError>(info.assets.unwrap_or_default())
}
}))
.buffer_unordered(8)
.try_fold(
Vec::with_capacity(symbols.len()),
|mut acc, assets| async move {
acc.extend(assets);
Ok(acc)
},
)
.await
}
fn convert_margin_open_order(
o: &QueryMarginAccountsOpenOrdersResponseInner,
instrument: &InstrumentNameExchange,
) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
let order_id_raw = match o.order_id {
Some(id) => id,
None => {
warn!(%instrument, "BinanceMargin open order missing orderId");
return None;
}
};
let order_id = OrderId(format_smolstr!("{order_id_raw}"));
if o.client_order_id.is_none() {
warn!(%instrument, order_id = %order_id_raw, "BinanceMargin open order missing clientOrderId, using orderId as fallback — order may not reconcile with engine state");
}
let cid = ClientOrderId::new(
o.client_order_id
.as_deref()
.unwrap_or(&format_smolstr!("{order_id_raw}")),
);
let side = match o.side.as_deref() {
Some(s) => parse_side(s)?,
None => {
warn!(%instrument, order_id = %order_id_raw, "BinanceMargin open order missing side");
return None;
}
};
let price = o.price.as_deref().and_then(|s| Decimal::from_str(s).ok());
let quantity = match o
.orig_qty
.as_deref()
.and_then(|s| Decimal::from_str(s).ok())
{
Some(v) => v,
None => {
warn!(%instrument, order_id = %order_id_raw, "BinanceMargin open order missing/unparseable origQty");
return None;
}
};
let filled_qty = match o.executed_qty.as_deref() {
Some(s) => match Decimal::from_str(s) {
Ok(v) => v,
Err(_) => {
warn!(%instrument, order_id = %order_id_raw, executed_qty = s, "BinanceMargin open order unparseable executedQty, defaulting to 0");
Decimal::ZERO
}
},
None => Decimal::ZERO,
};
let kind = match o.r#type.as_deref() {
Some(t) => parse_order_kind(t)?,
None => {
warn!(%instrument, order_id = %order_id_raw, "BinanceMargin open order missing type");
return None;
}
};
let time_in_force = parse_time_in_force(o.time_in_force.as_deref().unwrap_or("GTC"));
let time_exchange = match o.time.and_then(|ms| Utc.timestamp_millis_opt(ms).single()) {
Some(ts) => ts,
None => {
warn!(%instrument, order_id = %order_id_raw, "BinanceMargin open order missing/unparseable time, using now");
Utc::now()
}
};
Some(Order {
key: OrderKey::new(
ExchangeId::BinanceMargin,
instrument.clone(),
StrategyId::unknown(),
cid,
),
side,
price,
quantity,
kind,
time_in_force,
state: Open::new(order_id, time_exchange, filled_qty),
})
}
fn convert_margin_open_order_owned_symbol(
o: &QueryMarginAccountsOpenOrdersResponseInner,
) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
let instrument = match o.symbol.as_deref() {
Some(s) => InstrumentNameExchange::new(s),
None => {
warn!("BinanceMargin open order missing symbol in return-all query, dropping order");
return None;
}
};
convert_margin_open_order(o, &instrument)
}
fn convert_margin_trade(
t: &QueryMarginAccountsTradeListResponseInner,
instrument: &InstrumentNameExchange,
) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
let trade_id_raw = match t.id {
Some(id) => id,
None => {
warn!(%instrument, "BinanceMargin trade missing id");
return None;
}
};
let trade_id = TradeId(format_smolstr!("{trade_id_raw}"));
let order_id = match t.order_id {
Some(id) => OrderId(format_smolstr!("{id}")),
None => {
warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing orderId");
return None;
}
};
let side = match t.is_buyer {
Some(true) => Side::Buy,
Some(false) => Side::Sell,
None => {
warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing isBuyer");
return None;
}
};
let price = match t.price.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable price");
return None;
}
};
let quantity = match t.qty.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable qty");
return None;
}
};
let commission = t
.commission
.as_deref()
.and_then(|s| Decimal::from_str(s).ok())
.unwrap_or(Decimal::ZERO);
let time_exchange = match t.time.and_then(|ms| Utc.timestamp_millis_opt(ms).single()) {
Some(ts) => ts,
None => {
warn!(%instrument, trade_id = %trade_id_raw, "BinanceMargin trade missing/unparseable time, using now");
Utc::now()
}
};
let fee_asset = t
.commission_asset
.as_deref()
.map(AssetNameExchange::from)
.unwrap_or_else(|| AssetNameExchange::from("UNKNOWN"));
Some(Trade::new(
trade_id,
order_id,
instrument.clone(),
StrategyId::unknown(), time_exchange,
side,
price,
quantity,
AssetFees::new(fee_asset, commission, None),
))
}
#[derive(Debug)]
enum BuildOrderError {
Unsupported,
Build(String),
}
fn isolated_str(is_isolated: bool) -> String {
if is_isolated { "TRUE" } else { "FALSE" }.to_string()
}
fn build_cancel_order_params(
symbol: String,
id: Option<&OrderId>,
cid: &ClientOrderId,
is_isolated: bool,
) -> Result<MarginAccountCancelOrderParams, String> {
let mut builder =
MarginAccountCancelOrderParams::builder(symbol).is_isolated(isolated_str(is_isolated));
match id {
Some(order_id) => match order_id.0.parse::<i64>() {
Ok(id) => builder = builder.order_id(id),
Err(_) => {
error!(
order_id = %order_id.0,
"BinanceMargin cancel: exchange orderId not parseable as i64, falling back to clientOrderId"
);
builder = builder.orig_client_order_id(cid.0.to_string());
}
},
None => builder = builder.orig_client_order_id(cid.0.to_string()),
}
builder.build().map_err(|e| e.to_string())
}
#[allow(clippy::too_many_arguments)] fn build_new_order_params(
symbol: String,
side: Side,
price: Option<Decimal>,
quantity: Decimal,
kind: OrderKind,
time_in_force: TimeInForce,
new_client_order_id: String,
side_effect: MarginSideEffect,
is_isolated: bool,
) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
let binance_side = match side {
Side::Buy => MarginAccountNewOrderSideEnum::Buy,
Side::Sell => MarginAccountNewOrderSideEnum::Sell,
};
let (binance_type, binance_tif) =
convert_order_kind_tif_margin(kind, time_in_force).ok_or(BuildOrderError::Unsupported)?;
let mut builder = MarginAccountNewOrderParams::builder(
symbol,
binance_side,
binance_type.as_binance_str().to_string(),
)
.quantity(quantity)
.is_isolated(isolated_str(is_isolated))
.side_effect_type(side_effect.as_binance_str().to_string())
.new_client_order_id(new_client_order_id)
.new_order_resp_type(MarginAccountNewOrderNewOrderRespTypeEnum::Full);
if side_effect == MarginSideEffect::AutoBorrowRepay {
builder = builder.auto_repay_at_cancel(true);
}
if let Some(tif) = binance_tif {
builder = builder.time_in_force(tif);
}
match kind {
OrderKind::Limit => {
builder = builder.price(price);
}
OrderKind::Stop { trigger_price } | OrderKind::TakeProfit { trigger_price } => {
builder = builder.stop_price(trigger_price);
}
OrderKind::StopLimit { trigger_price } | OrderKind::TakeProfitLimit { trigger_price } => {
builder = builder.price(price).stop_price(trigger_price);
}
_ => {}
}
builder
.build()
.map_err(|e| BuildOrderError::Build(e.to_string()))
}
fn convert_order_kind_tif_margin(
kind: OrderKind,
tif: TimeInForce,
) -> Option<(
BinanceOrderType,
Option<MarginAccountNewOrderTimeInForceEnum>,
)> {
if matches!(
kind,
OrderKind::TrailingStop { .. } | OrderKind::TrailingStopLimit { .. }
) {
warn!(
?kind,
"BinanceMargin does not support trailing-stop orders (SDK trailingDelta binding gap)"
);
return None;
}
let (binance_type, binance_tif) = classify_order_kind_tif(kind, tif)?;
let margin_tif = binance_tif.map(|t| match t {
BinanceTimeInForce::Gtc => MarginAccountNewOrderTimeInForceEnum::Gtc,
BinanceTimeInForce::Ioc => MarginAccountNewOrderTimeInForceEnum::Ioc,
BinanceTimeInForce::Fok => MarginAccountNewOrderTimeInForceEnum::Fok,
});
Some((binance_type, margin_tif))
}
fn margin_avg_price(cummulative_quote_qty: Option<&str>, filled_qty: Decimal) -> Option<Decimal> {
if filled_qty.is_zero() {
return None;
}
let s = cummulative_quote_qty?;
match Decimal::from_str(s) {
Ok(cumulative) => cumulative.checked_div(filled_qty),
Err(_) => {
warn!(
cummulative_quote_qty = s,
"BinanceMargin: failed to parse cummulativeQuoteQty; avg price unavailable"
);
None
}
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
use super::*;
#[test]
fn margin_side_effect_default_is_auto_borrow_repay() {
assert_eq!(
MarginSideEffect::default(),
MarginSideEffect::AutoBorrowRepay
);
}
#[test]
fn margin_side_effect_wire_strings() {
assert_eq!(
MarginSideEffect::AutoBorrowRepay.as_binance_str(),
"AUTO_BORROW_REPAY"
);
assert_eq!(
MarginSideEffect::NoBorrow.as_binance_str(),
"NO_SIDE_EFFECT"
);
}
#[test]
fn margin_side_effect_serde_round_trip() {
assert_eq!(
serde_json::to_string(&MarginSideEffect::AutoBorrowRepay).unwrap(),
r#""auto_borrow_repay""#
);
assert_eq!(
serde_json::to_string(&MarginSideEffect::NoBorrow).unwrap(),
r#""no_borrow""#
);
assert_eq!(
serde_json::from_str::<MarginSideEffect>(r#""auto_borrow_repay""#).unwrap(),
MarginSideEffect::AutoBorrowRepay
);
assert_eq!(
serde_json::from_str::<MarginSideEffect>(r#""no_borrow""#).unwrap(),
MarginSideEffect::NoBorrow
);
}
#[test]
fn config_debug_redacts_secrets() {
let config = BinanceMarginConfig::new(
"my_api_key".to_string(),
"my_secret_key".to_string(),
false,
false,
Vec::new(),
MarginSideEffect::default(),
);
let debug = format!("{config:?}");
assert!(!debug.contains("my_api_key"));
assert!(!debug.contains("my_secret_key"));
assert!(debug.contains("***"));
}
#[test]
fn cross_margin_uses_common_case_defaults() {
let config = BinanceMarginConfig::cross_margin("k".to_string(), "s".to_string());
assert!(!config.testnet);
assert!(!config.is_isolated);
assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
assert_eq!(config.api_key(), "k");
}
#[test]
fn config_deserializes_with_defaults() {
let config: BinanceMarginConfig =
serde_json::from_str(r#"{"api_key":"k","secret_key":"s","testnet":false}"#)
.expect("deserialize");
assert!(!config.is_isolated);
assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
}
#[test]
fn config_deserializes_explicit_side_effect() {
let config: BinanceMarginConfig = serde_json::from_str(
r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true,"side_effect":"no_borrow"}"#,
)
.expect("deserialize");
assert!(config.is_isolated);
assert_eq!(config.side_effect, MarginSideEffect::NoBorrow);
}
#[test]
fn isolated_ctor_sets_symbols_and_flag() {
let symbols = vec![
InstrumentNameExchange::new("BTCUSDT"),
InstrumentNameExchange::new("ETHUSDT"),
];
let config =
BinanceMarginConfig::isolated("k".to_string(), "s".to_string(), symbols.clone());
assert!(config.is_isolated);
assert_eq!(config.isolated_symbols, symbols);
assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
}
#[test]
fn isolated_config_with_symbols_constructs() {
let config = BinanceMarginConfig::isolated(
"k".to_string(),
"s".to_string(),
vec![InstrumentNameExchange::new("BTCUSDT")],
);
let _client = BinanceMargin::new(config); }
#[test]
#[should_panic(expected = "non-empty isolated_symbols")]
fn isolated_empty_symbols_panics_at_new() {
let config = BinanceMarginConfig::new(
"k".to_string(),
"s".to_string(),
false,
true,
Vec::new(),
MarginSideEffect::default(),
);
let _ = BinanceMargin::new(config);
}
#[test]
#[should_panic(expected = "non-empty isolated_symbols")]
fn isolated_empty_symbols_panics_via_deserialize_path() {
let config: BinanceMarginConfig = serde_json::from_str(
r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true}"#,
)
.expect("deserialize");
assert!(config.isolated_symbols.is_empty());
let _ = BinanceMargin::new(config);
}
#[test]
fn isolated_str_maps_mode() {
assert_eq!(isolated_str(false), "FALSE");
assert_eq!(isolated_str(true), "TRUE");
}
#[test]
fn cancel_params_cross_with_parseable_order_id() {
let p = build_cancel_order_params(
"BTCUSDT".to_string(),
Some(&OrderId::new("12345")),
&ClientOrderId::new("cid-1"),
false,
)
.expect("build");
assert_eq!(p.symbol, "BTCUSDT");
assert_eq!(p.is_isolated.as_deref(), Some("FALSE"));
assert_eq!(p.order_id, Some(12345));
assert_eq!(p.orig_client_order_id, None);
}
#[test]
fn cancel_params_isolated_is_config_driven() {
let p = build_cancel_order_params(
"BTCUSDT".to_string(),
Some(&OrderId::new("12345")),
&ClientOrderId::new("cid-1"),
true,
)
.expect("build");
assert_eq!(p.is_isolated.as_deref(), Some("TRUE"));
assert_eq!(p.order_id, Some(12345));
}
#[test]
fn cancel_params_non_i64_order_id_falls_back_to_cid() {
let p = build_cancel_order_params(
"BTCUSDT".to_string(),
Some(&OrderId::new("not-an-i64")),
&ClientOrderId::new("cid-1"),
false,
)
.expect("build");
assert_eq!(p.order_id, None);
assert_eq!(p.orig_client_order_id.as_deref(), Some("cid-1"));
}
#[test]
fn cancel_params_absent_order_id_uses_cid() {
let p = build_cancel_order_params(
"BTCUSDT".to_string(),
None,
&ClientOrderId::new("cid-1"),
false,
)
.expect("build");
assert_eq!(p.order_id, None);
assert_eq!(p.orig_client_order_id.as_deref(), Some("cid-1"));
}
use crate::order::TrailingOffsetType;
fn params(
side: Side,
price: Option<Decimal>,
kind: OrderKind,
tif: TimeInForce,
side_effect: MarginSideEffect,
) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
build_new_order_params(
"BTCUSDT".to_string(),
side,
price,
Decimal::from(2),
kind,
tif,
"cid-1".to_string(),
side_effect,
false, )
}
fn gtc() -> TimeInForce {
TimeInForce::GoodUntilCancelled { post_only: false }
}
#[test]
fn new_order_limit_maps_core_fields() {
let p = params(
Side::Buy,
Some(Decimal::from(50_000)),
OrderKind::Limit,
gtc(),
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(p.symbol, "BTCUSDT");
assert_eq!(p.side.as_str(), "BUY");
assert_eq!(p.r#type, "LIMIT");
assert_eq!(p.quantity, Some(Decimal::from(2)));
assert_eq!(p.price, Some(Decimal::from(50_000)));
assert_eq!(p.stop_price, None);
assert_eq!(p.time_in_force.as_ref().map(|t| t.as_str()), Some("GTC"));
assert_eq!(p.new_client_order_id.as_deref(), Some("cid-1"));
assert_eq!(
p.new_order_resp_type.as_ref().map(|r| r.as_str()),
Some("FULL")
);
}
#[test]
fn new_order_is_isolated_is_config_driven() {
let cross = build_new_order_params(
"BTCUSDT".to_string(),
Side::Sell,
Some(Decimal::from(10)),
Decimal::from(2),
OrderKind::Limit,
gtc(),
"cid-1".to_string(),
MarginSideEffect::AutoBorrowRepay,
false,
)
.expect("build");
assert_eq!(cross.is_isolated.as_deref(), Some("FALSE"));
assert_eq!(cross.side.as_str(), "SELL");
let isolated = build_new_order_params(
"BTCUSDT".to_string(),
Side::Sell,
Some(Decimal::from(10)),
Decimal::from(2),
OrderKind::Limit,
gtc(),
"cid-1".to_string(),
MarginSideEffect::AutoBorrowRepay,
true,
)
.expect("build");
assert_eq!(isolated.is_isolated.as_deref(), Some("TRUE"));
}
#[test]
fn side_effect_and_auto_repay_gating() {
let auto = params(
Side::Buy,
Some(Decimal::from(1)),
OrderKind::Limit,
gtc(),
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(auto.side_effect_type.as_deref(), Some("AUTO_BORROW_REPAY"));
assert_eq!(auto.auto_repay_at_cancel, Some(true));
let no_borrow = params(
Side::Buy,
Some(Decimal::from(1)),
OrderKind::Limit,
gtc(),
MarginSideEffect::NoBorrow,
)
.expect("build");
assert_eq!(
no_borrow.side_effect_type.as_deref(),
Some("NO_SIDE_EFFECT")
);
assert_eq!(no_borrow.auto_repay_at_cancel, None);
}
#[test]
fn new_order_market_has_no_price_or_tif() {
let p = params(
Side::Buy,
None,
OrderKind::Market,
gtc(),
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(p.r#type, "MARKET");
assert_eq!(p.price, None);
assert_eq!(p.stop_price, None);
assert!(p.time_in_force.is_none());
}
#[test]
fn post_only_limit_maps_to_limit_maker() {
let p = params(
Side::Buy,
Some(Decimal::from(10)),
OrderKind::Limit,
TimeInForce::GoodUntilCancelled { post_only: true },
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(p.r#type, "LIMIT_MAKER");
assert!(p.time_in_force.is_none());
}
#[test]
fn conditional_kinds_set_stop_price() {
let trigger = Decimal::from(48_000);
let stop = params(
Side::Sell,
None,
OrderKind::Stop {
trigger_price: trigger,
},
gtc(),
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(stop.r#type, "STOP_LOSS");
assert_eq!(stop.stop_price, Some(trigger));
assert_eq!(stop.price, None);
let stop_limit = params(
Side::Sell,
Some(Decimal::from(47_900)),
OrderKind::StopLimit {
trigger_price: trigger,
},
gtc(),
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(stop_limit.r#type, "STOP_LOSS_LIMIT");
assert_eq!(stop_limit.stop_price, Some(trigger));
assert_eq!(stop_limit.price, Some(Decimal::from(47_900)));
assert_eq!(
stop_limit.time_in_force.as_ref().map(|t| t.as_str()),
Some("GTC")
);
let take_profit = params(
Side::Sell,
None,
OrderKind::TakeProfit {
trigger_price: trigger,
},
gtc(),
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(take_profit.r#type, "TAKE_PROFIT");
assert_eq!(take_profit.stop_price, Some(trigger));
}
#[test]
fn trailing_kinds_are_rejected() {
let trailing_stop = params(
Side::Sell,
None,
OrderKind::TrailingStop {
offset: Decimal::from(100),
offset_type: TrailingOffsetType::BasisPoints,
},
gtc(),
MarginSideEffect::AutoBorrowRepay,
);
assert!(matches!(trailing_stop, Err(BuildOrderError::Unsupported)));
let trailing_stop_limit = params(
Side::Sell,
Some(Decimal::from(100)),
OrderKind::TrailingStopLimit {
offset: Decimal::from(100),
offset_type: TrailingOffsetType::BasisPoints,
limit_offset: Decimal::from(10),
},
gtc(),
MarginSideEffect::AutoBorrowRepay,
);
assert!(matches!(
trailing_stop_limit,
Err(BuildOrderError::Unsupported)
));
}
#[test]
fn tif_variants_map_to_margin_enum() {
for (tif, expected) in [
(TimeInForce::ImmediateOrCancel, "IOC"),
(TimeInForce::FillOrKill, "FOK"),
(gtc(), "GTC"),
] {
let p = params(
Side::Buy,
Some(Decimal::from(1)),
OrderKind::Limit,
tif,
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(p.time_in_force.as_ref().map(|t| t.as_str()), Some(expected));
}
}
#[test]
fn avg_price_from_cumulative_quote_qty() {
assert_eq!(
margin_avg_price(Some("100"), Decimal::from(4)),
Some(Decimal::from(25))
);
assert_eq!(margin_avg_price(Some("100"), Decimal::ZERO), None);
assert_eq!(margin_avg_price(None, Decimal::from(4)), None);
assert_eq!(
margin_avg_price(Some("not-a-number"), Decimal::from(4)),
None
);
}
fn user_asset(
asset: &str,
free: &str,
locked: &str,
borrowed: &str,
interest: &str,
) -> QueryCrossMarginAccountDetailsResponseUserAssetsInner {
let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
a.asset = Some(asset.to_string());
a.free = Some(free.to_string());
a.locked = Some(locked.to_string());
a.borrowed = Some(borrowed.to_string());
a.interest = Some(interest.to_string());
a
}
#[test]
fn margin_balance_entry_maps_debt() {
let ab =
convert_margin_balance_entry(user_asset("USDT", "10", "2", "3", "0.5"), Utc::now())
.expect("convert");
assert_eq!(ab.asset.name().as_str(), "USDT");
assert_eq!(ab.balance.total, Decimal::from(12));
assert_eq!(ab.balance.free, Decimal::from(10));
assert!(
ab.balance.margin.is_some(),
"REST snapshot must populate margin"
);
assert_eq!(ab.balance.net_asset(), Decimal::from(9));
}
#[test]
fn margin_balance_short_is_negative_net() {
let ab = convert_margin_balance_entry(user_asset("BTC", "1", "0", "5", "0"), Utc::now())
.expect("convert");
assert_eq!(ab.balance.net_asset(), Decimal::from(-4));
}
#[test]
fn margin_balance_missing_debt_defaults_to_zero() {
let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
a.asset = Some("ETH".to_string());
a.free = Some("4".to_string());
a.locked = Some("0".to_string());
let ab = convert_margin_balance_entry(a, Utc::now()).expect("convert");
assert!(ab.balance.margin.is_some());
assert_eq!(ab.balance.net_asset(), Decimal::from(4));
}
#[test]
fn margin_balance_missing_free_is_dropped() {
let mut a = QueryCrossMarginAccountDetailsResponseUserAssetsInner::new();
a.asset = Some("USDT".to_string());
a.locked = Some("0".to_string());
assert!(convert_margin_balance_entry(a, Utc::now()).is_none());
}
#[test]
fn margin_balance_filtering() {
let assets = vec![
user_asset("USDT", "10", "0", "0", "0"),
user_asset("BTC", "1", "0", "0", "0"),
user_asset("ETH", "5", "0", "0", "0"),
];
assert_eq!(
filter_and_convert_margin_balances(assets.clone(), &[]).len(),
3
);
let filtered = filter_and_convert_margin_balances(
assets,
&[AssetNameExchange::new("BTC"), AssetNameExchange::new("ETH")],
);
assert_eq!(filtered.len(), 2);
assert!(filtered.iter().all(|b| b.asset.name().as_str() != "USDT"));
}
#[test]
fn margin_balance_filtering_large_slice_uses_hashset_branch() {
let assets = vec![
user_asset("BTC", "1", "0", "0", "0"),
user_asset("ETH", "5", "0", "0", "0"),
user_asset("DOGE", "9", "0", "0", "0"), ];
let mut requested: Vec<AssetNameExchange> = (0..17)
.map(|i| AssetNameExchange::new(format!("A{i:02}")))
.collect();
requested.push(AssetNameExchange::new("BTC"));
requested.push(AssetNameExchange::new("ETH"));
assert!(
requested.len() > 16,
"must exceed the linear-scan threshold"
);
let filtered = filter_and_convert_margin_balances(assets, &requested);
assert_eq!(filtered.len(), 2);
assert!(filtered.iter().all(|b| b.asset.name().as_str() != "DOGE"));
}
use binance_sdk::margin_trading::rest_api::{
QueryIsolatedMarginAccountInfoResponseAssetsInnerBaseAsset as IsoBase,
QueryIsolatedMarginAccountInfoResponseAssetsInnerQuoteAsset as IsoQuote,
};
fn d(s: &str) -> Decimal {
Decimal::from_str(s).unwrap()
}
fn iso_base(asset: &str, free: &str, locked: &str, borrowed: &str, interest: &str) -> IsoBase {
let mut b = IsoBase::new();
b.asset = Some(asset.to_string());
b.free = Some(free.to_string());
b.locked = Some(locked.to_string());
b.borrowed = Some(borrowed.to_string());
b.interest = Some(interest.to_string());
b
}
fn iso_quote(
asset: &str,
free: &str,
locked: &str,
borrowed: &str,
interest: &str,
) -> IsoQuote {
let mut q = IsoQuote::new();
q.asset = Some(asset.to_string());
q.free = Some(free.to_string());
q.locked = Some(locked.to_string());
q.borrowed = Some(borrowed.to_string());
q.interest = Some(interest.to_string());
q
}
fn iso_entry(
symbol: &str,
base: IsoBase,
quote: IsoQuote,
margin_level: Option<&str>,
margin_ratio: Option<&str>,
liquidate_price: Option<&str>,
) -> QueryIsolatedMarginAccountInfoResponseAssetsInner {
let mut e = QueryIsolatedMarginAccountInfoResponseAssetsInner::new();
e.symbol = Some(symbol.to_string());
e.base_asset = Some(Box::new(base));
e.quote_asset = Some(Box::new(quote));
e.margin_level = margin_level.map(str::to_string);
e.margin_ratio = margin_ratio.map(str::to_string);
e.liquidate_price = liquidate_price.map(str::to_string);
e
}
fn isolated_client(symbols: &[&str]) -> BinanceMargin {
let config = BinanceMarginConfig::isolated(
"k".to_string(),
"s".to_string(),
symbols
.iter()
.map(|s| InstrumentNameExchange::new(*s))
.collect(),
);
BinanceMargin::new(config)
}
#[test]
fn isolated_assets_map_base_quote_and_risk() {
let entry = iso_entry(
"BTCUSDT",
iso_base("BTC", "0.5", "0.1", "0.2", "0.001"),
iso_quote("USDT", "1000", "50", "300", "1.5"),
Some("3.5"),
Some("0.12"),
Some("48000"),
);
let map = convert_isolated_margin_assets(vec![entry]);
let state = map
.get(&InstrumentNameExchange::new("BTCUSDT"))
.expect("entry present");
assert_eq!(state.base.asset.name().as_str(), "BTC");
assert_eq!(state.base.balance.total, d("0.6"));
assert_eq!(state.base.balance.free, d("0.5"));
assert_eq!(state.base.balance.net_asset(), d("0.4"));
assert_eq!(state.quote.asset.name().as_str(), "USDT");
assert_eq!(state.quote.balance.total, d("1050"));
assert_eq!(state.quote.balance.net_asset(), d("750"));
assert_eq!(state.risk.margin_level, Some(d("3.5")));
assert_eq!(state.risk.margin_ratio, Some(d("0.12")));
assert_eq!(state.risk.liquidation_price, Some(d("48000")));
}
#[test]
fn isolated_base_short_is_negative_net() {
let entry = iso_entry(
"BTCUSDT",
iso_base("BTC", "0", "0", "1.5", "0.001"),
iso_quote("USDT", "100", "0", "0", "0"),
None,
None,
None,
);
let map = convert_isolated_margin_assets(vec![entry]);
let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
assert_eq!(state.base.balance.net_asset(), d("-1.5"));
}
#[test]
fn isolated_missing_debt_defaults_zero_and_risk_optional() {
let mut base = IsoBase::new();
base.asset = Some("BTC".to_string());
base.free = Some("0.5".to_string());
base.locked = Some("0".to_string());
let mut quote = IsoQuote::new();
quote.asset = Some("USDT".to_string());
quote.free = Some("100".to_string());
quote.locked = Some("0".to_string());
let map = convert_isolated_margin_assets(vec![iso_entry(
"BTCUSDT", base, quote, None, None, None,
)]);
let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
assert!(state.base.balance.margin.is_some());
assert_eq!(state.base.balance.net_asset(), d("0.5"));
assert_eq!(state.risk.margin_level, None);
assert_eq!(state.risk.liquidation_price, None);
}
#[test]
fn isolated_unparseable_risk_is_none_but_entry_kept() {
let entry = iso_entry(
"BTCUSDT",
iso_base("BTC", "1", "0", "0", "0"),
iso_quote("USDT", "1", "0", "0", "0"),
Some("not_a_number"),
None,
None,
);
let map = convert_isolated_margin_assets(vec![entry]);
let state = map.get(&InstrumentNameExchange::new("BTCUSDT")).unwrap();
assert_eq!(state.risk.margin_level, None);
}
#[test]
fn isolated_missing_base_free_drops_entry() {
let mut base = IsoBase::new();
base.asset = Some("BTC".to_string());
base.locked = Some("0".to_string()); let entry = iso_entry(
"BTCUSDT",
base,
iso_quote("USDT", "100", "0", "0", "0"),
None,
None,
None,
);
assert!(convert_isolated_margin_assets(vec![entry]).is_empty());
}
#[test]
fn isolated_missing_symbol_drops_entry() {
let mut e = QueryIsolatedMarginAccountInfoResponseAssetsInner::new();
e.base_asset = Some(Box::new(iso_base("BTC", "1", "0", "0", "0")));
e.quote_asset = Some(Box::new(iso_quote("USDT", "1", "0", "0", "0")));
assert!(convert_isolated_margin_assets(vec![e]).is_empty());
}
#[test]
fn chunk_symbols_batches_by_five() {
let syms: Vec<InstrumentNameExchange> = (0..12)
.map(|i| InstrumentNameExchange::new(format!("S{i:02}")))
.collect();
let chunks = chunk_symbols(&syms);
assert_eq!(chunks.len(), 3, "12 symbols → 5 + 5 + 2");
assert_eq!(chunks[0].split(',').count(), 5);
assert_eq!(chunks[1].split(',').count(), 5);
assert_eq!(chunks[2].split(',').count(), 2);
let three: Vec<_> = (0..3)
.map(|i| InstrumentNameExchange::new(format!("S{i}")))
.collect();
assert_eq!(chunk_symbols(&three).len(), 1);
let five: Vec<_> = (0..5)
.map(|i| InstrumentNameExchange::new(format!("S{i}")))
.collect();
let c = chunk_symbols(&five);
assert_eq!(c.len(), 1);
assert_eq!(c[0].split(',').count(), 5);
assert!(chunk_symbols(&[]).is_empty());
}
#[test]
fn effective_isolated_set_empty_returns_all_configured() {
let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
assert_eq!(
client.effective_isolated_set(&[]),
vec![
InstrumentNameExchange::new("BTCUSDT"),
InstrumentNameExchange::new("ETHUSDT"),
]
);
}
#[test]
fn effective_isolated_set_intersects_requested() {
let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
assert_eq!(
client.effective_isolated_set(&[InstrumentNameExchange::new("ETHUSDT")]),
vec![InstrumentNameExchange::new("ETHUSDT")]
);
}
#[test]
fn effective_isolated_set_skips_out_of_set() {
let client = isolated_client(&["BTCUSDT", "ETHUSDT"]);
assert_eq!(
client.effective_isolated_set(&[
InstrumentNameExchange::new("BTCUSDT"),
InstrumentNameExchange::new("DOGEUSDT"),
]),
vec![InstrumentNameExchange::new("BTCUSDT")]
);
}
#[tokio::test]
async fn fetch_balances_isolated_returns_empty() {
let client = isolated_client(&["BTCUSDT"]);
assert!(client.fetch_balances(&[]).await.expect("ok").is_empty());
}
fn open_order(order_id: Option<i64>, side: &str) -> QueryMarginAccountsOpenOrdersResponseInner {
let mut o = QueryMarginAccountsOpenOrdersResponseInner::new();
o.order_id = order_id;
o.client_order_id = Some("cid-9".to_string());
o.side = Some(side.to_string());
o.price = Some("50000".to_string());
o.orig_qty = Some("2".to_string());
o.executed_qty = Some("0.5".to_string());
o.r#type = Some("LIMIT".to_string());
o.time_in_force = Some("GTC".to_string());
o.time = Some(1_700_000_000_000);
o
}
#[test]
fn margin_open_order_converts() {
let inst = InstrumentNameExchange::new("BTCUSDT");
let order =
convert_margin_open_order(&open_order(Some(42), "BUY"), &inst).expect("convert");
assert_eq!(order.key.exchange, ExchangeId::BinanceMargin);
assert_eq!(order.state.id.0.as_str(), "42");
assert_eq!(order.key.cid.0.as_str(), "cid-9");
assert_eq!(order.side, Side::Buy);
assert_eq!(order.price, Some(Decimal::from(50_000)));
assert_eq!(order.quantity, Decimal::from(2));
assert_eq!(order.state.filled_quantity, Decimal::new(5, 1));
assert_eq!(order.kind, OrderKind::Limit);
}
#[test]
fn margin_open_order_missing_order_id_is_dropped() {
let inst = InstrumentNameExchange::new("BTCUSDT");
assert!(convert_margin_open_order(&open_order(None, "BUY"), &inst).is_none());
}
#[test]
fn margin_open_order_owned_symbol_recovers_instrument() {
let mut o = open_order(Some(42), "SELL");
o.symbol = Some("ETHUSDT".to_string());
let order = convert_margin_open_order_owned_symbol(&o).expect("convert");
assert_eq!(order.key.instrument.name().as_str(), "ETHUSDT");
assert_eq!(order.key.exchange, ExchangeId::BinanceMargin);
assert_eq!(order.side, Side::Sell);
}
#[test]
fn margin_open_order_owned_symbol_missing_symbol_is_dropped() {
let o = open_order(Some(42), "BUY");
assert!(o.symbol.is_none());
assert!(convert_margin_open_order_owned_symbol(&o).is_none());
}
fn trade(id: Option<i64>, is_buyer: Option<bool>) -> QueryMarginAccountsTradeListResponseInner {
let mut t = QueryMarginAccountsTradeListResponseInner::new();
t.id = id;
t.order_id = Some(7);
t.is_buyer = is_buyer;
t.price = Some("48000".to_string());
t.qty = Some("0.25".to_string());
t.commission = Some("0.001".to_string());
t.commission_asset = Some("BNB".to_string());
t.time = Some(1_700_000_000_000);
t
}
#[test]
fn margin_trade_converts() {
let inst = InstrumentNameExchange::new("BTCUSDT");
let tr = convert_margin_trade(&trade(Some(11), Some(false)), &inst).expect("convert");
assert_eq!(tr.id.0.as_str(), "11");
assert_eq!(tr.order_id.0.as_str(), "7");
assert_eq!(tr.side, Side::Sell);
assert_eq!(tr.price, Decimal::from(48_000));
assert_eq!(tr.quantity, Decimal::new(25, 2));
assert_eq!(tr.fees.asset.name().as_str(), "BNB");
assert_eq!(tr.fees.fees, Decimal::new(1, 3));
}
#[test]
fn margin_trade_missing_fields_are_dropped() {
let inst = InstrumentNameExchange::new("BTCUSDT");
assert!(convert_margin_trade(&trade(None, Some(true)), &inst).is_none());
assert!(convert_margin_trade(&trade(Some(11), None), &inst).is_none());
}
#[test]
fn token_renew_after_applies_margin_and_floor() {
let now_ms = Utc::now().timestamp_millis();
let far = token_renew_after(now_ms + 86_400_000).as_secs();
assert!(
far > 80_000 && far <= 86_400,
"expected ~24h-minus-margin, got {far}"
);
assert_eq!(
token_renew_after(now_ms - 10_000).as_secs(),
TOKEN_MIN_LIFETIME_SECS
);
assert_eq!(
token_renew_after(now_ms + 60_000).as_secs(),
TOKEN_MIN_LIFETIME_SECS
);
}
fn push(event: serde_json::Value) -> String {
serde_json::json!({ "subscriptionId": 1, "event": event }).to_string()
}
#[test]
fn margin_ws_rpc_ack_yields_no_events() {
let frame = serde_json::json!({ "id": "abc", "status": 200, "result": {} }).to_string();
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
assert!(buf.is_empty());
}
#[test]
fn margin_ws_execution_report_trade_maps_to_trade() {
let frame = push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
"x": "TRADE", "X": "PARTIALLY_FILLED", "i": 12_345_i64, "c": "cid-1",
"t": 99_i64, "l": "0.5", "L": "48000", "z": "0.5",
"n": "0.001", "N": "BNB", "T": 1_700_000_000_000_i64,
}));
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
assert_eq!(buf.len(), 1);
assert_eq!(buf[0].exchange, ExchangeId::BinanceMargin);
match &buf[0].kind {
AccountEventKind::Trade(t) => {
assert_eq!(t.instrument.name().as_str(), "BTCUSDT");
assert_eq!(t.side, Side::Buy);
assert_eq!(t.price, Decimal::from(48_000));
assert_eq!(t.quantity, Decimal::new(5, 1));
assert_eq!(t.fees.asset.name().as_str(), "BNB"); assert_eq!(t.fees.fees, Decimal::new(1, 3));
}
other => panic!("expected Trade, got {other:?}"),
}
}
#[test]
fn margin_ws_new_report_maps_to_active_order_snapshot() {
let frame = push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "LIMIT",
"x": "NEW", "X": "NEW", "i": 7_i64, "c": "cid-2",
"p": "48000", "q": "1", "f": "GTC", "z": "0", "T": 1_700_000_000_000_i64,
}));
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
assert_eq!(buf.len(), 1);
match &buf[0].kind {
AccountEventKind::OrderSnapshot(snap) => {
assert_eq!(snap.0.side, Side::Sell);
assert_eq!(snap.0.price, Some(Decimal::from(48_000)));
assert_eq!(snap.0.quantity, Decimal::from(1));
assert!(
matches!(snap.0.state, OrderState::Active(_)),
"NEW should be an active (resting) order"
);
}
other => panic!("expected OrderSnapshot, got {other:?}"),
}
}
#[test]
fn margin_ws_canceled_report_maps_to_order_cancelled() {
let frame = push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
"x": "CANCELED", "X": "CANCELED", "i": 8_i64, "c": "cid-3",
"z": "0", "T": 1_700_000_000_000_i64,
}));
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
assert_eq!(buf.len(), 1);
match &buf[0].kind {
AccountEventKind::OrderCancelled(resp) => assert!(resp.state.is_ok()),
other => panic!("expected OrderCancelled, got {other:?}"),
}
}
#[test]
fn margin_ws_execution_report_missing_symbol_is_dropped() {
let frame = push(serde_json::json!({
"e": "executionReport", "S": "BUY", "x": "TRADE", "i": 1_i64, "t": 1_i64,
"l": "1", "L": "100", "T": 1_700_000_000_000_i64,
}));
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
assert!(buf.is_empty());
}
#[test]
fn margin_ws_outbound_account_position_maps_to_balance_stream_updates() {
let frame = push(serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [
{ "a": "USDT", "f": "100.0", "l": "5.0" },
{ "a": "BTC", "f": "0.5", "l": "0" },
],
}));
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
assert_eq!(buf.len(), 2);
for ev in &buf {
assert!(matches!(ev.kind, AccountEventKind::BalanceStreamUpdate(_)));
}
}
#[test]
fn margin_ws_stream_terminated_signals_reconnect() {
let frame = push(serde_json::json!({ "e": "eventStreamTerminated" }));
let mut buf = Vec::new();
assert!(
convert_margin_user_data_events(&frame, &mut buf),
"eventStreamTerminated must signal reconnect"
);
assert!(buf.is_empty());
}
#[test]
fn margin_ws_margin_specific_events_are_observable_only() {
let mut buf = Vec::new();
let liability = push(serde_json::json!({
"e": "userLiabilityChange", "a": "USDT", "t": "BORROW", "p": "100", "i": "0.01",
}));
assert!(!convert_margin_user_data_events(&liability, &mut buf));
let level = push(serde_json::json!({
"e": "marginLevelStatusChange", "l": "1.5", "s": "MARGIN_LEVEL_2",
}));
assert!(!convert_margin_user_data_events(&level, &mut buf));
assert!(buf.is_empty());
}
#[test]
fn margin_trade_event_dedup_by_key() {
let frame = push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
"x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 4_242_i64,
"l": "1", "L": "100", "z": "1", "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
}));
let mut buf = Vec::new();
convert_margin_user_data_events(&frame, &mut buf);
let event = buf.pop().expect("a Trade event");
let cache = new_dedup_cache();
let first = dedup_key_from_event(&event).expect("Trade events have a dedup key");
assert!(!is_duplicate(&cache, first), "first sighting is fresh");
let second = dedup_key_from_event(&event).expect("Trade events have a dedup key");
assert!(
is_duplicate(&cache, second),
"second sighting is a duplicate"
);
}
fn push_with_sub(subscription_id: i64, event: serde_json::Value) -> String {
serde_json::json!({ "subscriptionId": subscription_id, "event": event }).to_string()
}
fn isolated_handler(
sub_map: Arc<Mutex<HashMap<i64, InstrumentNameExchange>>>,
base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
) -> impl FnMut(Outboundaccountposition, Option<i64>, &mut Vec<UnindexedAccountEvent>) {
move |position, subscription_id, buf| {
route_isolated_account_position(position, subscription_id, &sub_map, &base_quote, buf);
}
}
fn btcusdt() -> InstrumentNameExchange {
InstrumentNameExchange::new("BTCUSDT")
}
fn token(expiration_time_ms: i64) -> UserListenToken {
UserListenToken {
token: "t".to_string(),
expiration_time_ms,
}
}
#[test]
fn listen_token_query_cross_vs_isolated() {
assert!(build_listen_token_query(None).is_empty());
let q = build_listen_token_query(Some(&btcusdt()));
assert_eq!(q.get("isIsolated").and_then(|v| v.as_str()), Some("TRUE"));
assert_eq!(q.get("symbol").and_then(|v| v.as_str()), Some("BTCUSDT"));
assert_eq!(q.len(), 2);
}
#[test]
fn earliest_token_expiry_picks_min() {
let tokens = vec![
(btcusdt(), token(3_000)),
(InstrumentNameExchange::new("ETHUSDT"), token(1_000)),
(InstrumentNameExchange::new("BNBUSDT"), token(2_000)),
];
assert_eq!(earliest_token_expiry_ms(&tokens), 1_000);
assert_eq!(earliest_token_expiry_ms(&[]), i64::MAX);
}
#[test]
fn base_quote_map_extracts_base_and_quote() {
let entries = vec![
iso_entry(
"BTCUSDT",
iso_base("BTC", "0", "0", "0", "0"),
iso_quote("USDT", "0", "0", "0", "0"),
None,
None,
None,
),
iso_entry(
"ETHBTC",
iso_base("ETH", "0", "0", "0", "0"),
iso_quote("BTC", "0", "0", "0", "0"),
None,
None,
None,
),
];
let map = build_base_quote_map(&entries);
assert_eq!(
map.get(&btcusdt())
.map(|(b, q)| (b.name().as_str(), q.name().as_str())),
Some(("BTC", "USDT"))
);
assert_eq!(
map.get(&InstrumentNameExchange::new("ETHBTC"))
.map(|(b, q)| (b.name().as_str(), q.name().as_str())),
Some(("ETH", "BTC"))
);
}
#[test]
fn isolated_outbound_position_routes_to_instrument_balance_update() {
let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
let base_quote = Arc::new(HashMap::from([(
btcusdt(),
(
AssetNameExchange::new("BTC"),
AssetNameExchange::new("USDT"),
),
)]));
let mut handler = isolated_handler(sub_map, base_quote);
let frame = push_with_sub(
7,
serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [
{ "a": "USDT", "f": "1000.0", "l": "50.0" },
{ "a": "BTC", "f": "0.5", "l": "0.1" },
],
}),
);
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler
));
assert_eq!(buf.len(), 1, "one InstrumentBalanceUpdate for the pair");
match &buf[0].kind {
AccountEventKind::InstrumentBalanceUpdate(ibu) => {
assert_eq!(ibu.instrument.name().as_str(), "BTCUSDT");
assert_eq!(ibu.base.asset.name().as_str(), "BTC");
assert_eq!(ibu.base.update.free, Decimal::new(5, 1));
assert_eq!(ibu.base.update.locked, Decimal::new(1, 1));
assert_eq!(ibu.quote.asset.name().as_str(), "USDT");
assert_eq!(ibu.quote.update.free, Decimal::from(1000));
assert_eq!(ibu.quote.update.locked, Decimal::from(50));
}
other => panic!("expected InstrumentBalanceUpdate, got {other:?}"),
}
assert!(!matches!(
buf[0].kind,
AccountEventKind::BalanceStreamUpdate(_)
));
}
#[test]
fn isolated_outbound_position_unknown_subscription_dropped() {
let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
let base_quote = Arc::new(HashMap::from([(
btcusdt(),
(
AssetNameExchange::new("BTC"),
AssetNameExchange::new("USDT"),
),
)]));
let mut handler = isolated_handler(sub_map, base_quote);
let frame = push_with_sub(
99,
serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [{ "a": "BTC", "f": "1", "l": "0" }, { "a": "USDT", "f": "1", "l": "0" }],
}),
);
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler
));
assert!(buf.is_empty(), "unmapped subscriptionId frame is dropped");
}
#[test]
fn isolated_outbound_position_missing_side_dropped() {
let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
let base_quote = Arc::new(HashMap::from([(
btcusdt(),
(
AssetNameExchange::new("BTC"),
AssetNameExchange::new("USDT"),
),
)]));
let mut handler = isolated_handler(sub_map, base_quote);
let frame = push_with_sub(
7,
serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [{ "a": "BTC", "f": "0.5", "l": "0" }],
}),
);
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler
));
assert!(buf.is_empty(), "missing quote side → frame dropped");
}
#[test]
fn isolated_execution_report_routes_by_inner_symbol_independent_of_map() {
let sub_map = Arc::new(Mutex::new(HashMap::<i64, InstrumentNameExchange>::new()));
let base_quote = Arc::new(HashMap::new());
let mut handler = isolated_handler(sub_map, base_quote);
let frame = push_with_sub(
42,
serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
"x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 5_i64,
"l": "1", "L": "100", "z": "1", "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
}),
);
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler
));
assert_eq!(buf.len(), 1);
match &buf[0].kind {
AccountEventKind::Trade(t) => assert_eq!(t.instrument.name().as_str(), "BTCUSDT"),
other => panic!("expected Trade, got {other:?}"),
}
}
}