mod records;
pub use records::MarginForcedLiquidation;
use super::margin_risk::{
RISK_STREAM_URL, RiskStreamTask, SapiRiskListenKeys, log_liability_change,
log_margin_level_change, margin_level_notice,
};
use super::rules::RulesLoader;
use super::shared::{
AbortOnDropStream, BINANCE_MAX_TRADES, BinanceOrderType, BinanceTimeInForce,
CONNECT_TIMEOUT_SECS, DisconnectOnDrop, ExecutionReport, FILL_RECOVERY_TIMEOUT_SECS,
HEARTBEAT_TIMEOUT_SECS, MyTradesFrom, ORDER_EXECUTIONS_BUDGET, OpenOrderListing,
PlacementResponse, RateLimitTracker, RequestKind, SIGNAL_RECOVERY_LOOKBACK_MS,
SharedDedupCache, TrailingOrderError, UnhandledEvents, UnrecoveredFills, UnsubscribeOnDrop,
UserDataFrame, WeightPool, binance_filled_qty, classify_order_kind_tif,
classify_rest_order_error, classify_rest_query_error, convert_ended_order,
convert_execution_report, convert_open_order_listing, convert_open_order_owned_symbol,
dedup_key_from_event, disconnect_error, drop_after, frame_excerpt, gap_failed, gap_time,
is_duplicate, is_unknown_order, log_unhandled_event, log_unrecognised_frame, new_dedup_cache,
parse_user_data_frame, placed_order_state, recovered_order_cids, recovered_order_totals,
response_decode_error, rest_call_with_retry, retry_or_end, trailing_params,
};
use crate::rules::{OrderRules, PreSendRules};
use crate::{
AccountEventKind, AccountSnapshot, FillRecoveryFailure, InstrumentAccountSnapshot,
InstrumentBalanceUpdate, IsolatedInstrumentState, IsolatedMarginRisk, UnindexedAccountEvent,
UnindexedAccountSnapshot,
balance::{AssetBalance, AssetBalanceUpdate, Balance, BalanceUpdate},
borrow_capacity::BorrowCapacity,
client::{
BorrowCapacityClient, ExecutionClient, OrderStatusClient, ShortabilityClient,
order_recovery::{
KnownLiveOrders, OpenListing, OrderLookup, SharedKnownLiveOrders, UncheckedOrders,
fetch_ended_by_key, recover_ended_orders,
},
reconnect_budget::{BackoffLimits, ReconnectBudget},
},
emit_stream_terminated,
error::{ApiError, ConnectivityError, OrderError, UnindexedClientError},
margin::MarginEffect,
order::{
Order, OrderKey, OrderKind, TimeInForce, UnindexedInactiveOrder, UnindexedOrderKey,
id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
request::{OrderRequestCancel, OrderRequestOpen, UnindexedOrderResponseCancel},
state::{Cancelled, Open, OrderState, UnindexedOrderState},
},
position::PositionReport,
shortability::Shortability,
trade::{AssetFees, Trade, TradeId, TradesRead},
};
use binance_sdk::{
common::{
config::{ConfigurationRestApi, ConfigurationWebsocketApi},
constants::{MARGIN_TRADING_REST_API_PROD_URL, SPOT_WS_API_PROD_URL},
errors::WebsocketError,
models::WebsocketEvent,
websocket::{
SendWebsocketMessageResult, WebsocketApi as WsApiBase, WebsocketMessageSendOptions,
},
},
margin_trading::{
MarginTradingRestApi,
rest_api::{
GetAllCrossMarginPairsParams, GetAllIsolatedMarginSymbolParams,
GetAllMarginAssetsParams, GetDelistScheduleParams,
GetFutureHourlyInterestRateIsIsolatedEnum, GetFutureHourlyInterestRateParams,
MarginAccountCancelOrderIsIsolatedEnum, MarginAccountCancelOrderParams,
MarginAccountNewOrderIsIsolatedEnum, MarginAccountNewOrderNewOrderRespTypeEnum,
MarginAccountNewOrderParams, MarginAccountNewOrderSideEffectTypeEnum,
MarginAccountNewOrderSideEnum, MarginAccountNewOrderTimeInForceEnum,
MarginAccountNewOrderTypeEnum, QueryCrossMarginAccountDetailsParams,
QueryCrossMarginAccountDetailsResponseUserAssetsInner,
QueryIsolatedMarginAccountInfoParams,
QueryIsolatedMarginAccountInfoResponseAssetsInner,
QueryMarginAccountsOpenOrdersIsIsolatedEnum, QueryMarginAccountsOpenOrdersParams,
QueryMarginAccountsOrderIsIsolatedEnum, QueryMarginAccountsOrderParams,
QueryMarginAccountsTradeListIsIsolatedEnum, QueryMarginAccountsTradeListParams,
QueryMarginAccountsTradeListResponseInner, QueryMaxBorrowParams, RestApi,
},
websocket_streams::{
MarginLevelStatusChange, OutboundAccountPosition, UserLiabilityChange,
},
},
};
use chrono::{DateTime, TimeZone, Utc};
use fnv::FnvHashSet;
use futures::stream::BoxStream;
use rust_decimal::Decimal;
use rustrade_instrument::{
Side,
asset::name::AssetNameExchange,
exchange::ExchangeId,
instrument::{kind::InstrumentKindDiscriminant, name::InstrumentNameExchange},
};
use serde::Deserialize;
use smol_str::{SmolStr, format_smolstr};
use std::{
collections::{BTreeMap, HashMap},
str::FromStr,
sync::{
Arc, Mutex,
atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
},
time::Duration,
};
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, error, info, trace, warn};
fn binance_side_effect(effect: MarginEffect) -> MarginAccountNewOrderSideEffectTypeEnum {
match effect {
MarginEffect::NoSideEffect => MarginAccountNewOrderSideEffectTypeEnum::NoSideEffect,
MarginEffect::Borrow => MarginAccountNewOrderSideEffectTypeEnum::MarginBuy,
MarginEffect::Repay => MarginAccountNewOrderSideEffectTypeEnum::AutoRepay,
}
}
#[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 margin_effect: MarginEffect,
#[serde(default = "crate::margin::default_auto_repay_at_cancel")]
pub auto_repay_at_cancel: bool,
#[serde(default)]
pub pre_send_rules: PreSendRules,
}
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("margin_effect", &self.margin_effect)
.field("auto_repay_at_cancel", &self.auto_repay_at_cancel)
.field("pre_send_rules", &self.pre_send_rules)
.finish()
}
}
impl BinanceMarginConfig {
pub fn new(
api_key: String,
secret_key: String,
testnet: bool,
is_isolated: bool,
isolated_symbols: Vec<InstrumentNameExchange>,
margin_effect: MarginEffect,
auto_repay_at_cancel: bool,
) -> Self {
Self {
api_key,
secret_key,
testnet,
is_isolated,
isolated_symbols,
margin_effect,
auto_repay_at_cancel,
pre_send_rules: PreSendRules::default(),
}
}
#[must_use]
pub fn with_pre_send_rules(mut self, pre_send_rules: PreSendRules) -> Self {
self.pre_send_rules = pre_send_rules;
self
}
pub fn cross_margin(api_key: String, secret_key: String) -> Self {
Self::new(
api_key,
secret_key,
false,
false,
Vec::new(),
MarginEffect::default(),
true,
)
}
pub fn isolated(
api_key: String,
secret_key: String,
symbols: Vec<InstrumentNameExchange>,
) -> Self {
Self::new(
api_key,
secret_key,
false,
true,
symbols,
MarginEffect::default(),
true,
)
}
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>,
known_live: SharedKnownLiveOrders,
rules: Arc<RulesLoader>,
}
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("margin_effect", &self.config.margin_effect)
.finish_non_exhaustive()
}
}
impl BinanceMargin {
pub async fn load_rules(
&self,
instruments: &[InstrumentNameExchange],
) -> Result<HashMap<InstrumentNameExchange, OrderRules>, UnindexedClientError> {
self.rules.load(instruments).await
}
fn remember_live<'a>(
&self,
orders: impl IntoIterator<Item = &'a Order<ExchangeId, InstrumentNameExchange, Open>>,
) {
let mut known = self.known_live.lock();
for order in orders {
known.live(&order.key, order.quantity, order.kind, &order.state);
}
}
#[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,
RequestKind::Query,
)
}))
.buffer_unordered(8)
.map(|result| {
let (inst, listing) = result?;
self.remember_live(&listing.orders);
let wrapped = listing
.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,
listing.complete,
PositionReport::Unreported,
isolated,
))
})
.try_collect()
.await?;
Ok(AccountSnapshot::new(
ExchangeId::BinanceMargin,
Vec::new(),
instrument_snapshots,
))
}
}
impl ExecutionClient for BinanceMargin {
const EXCHANGE: ExchangeId = ExchangeId::BinanceMargin;
const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] =
&[InstrumentKindDiscriminant::Spot];
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 rules = Arc::new(RulesLoader::new(
config.pre_send_rules,
Arc::new(binance_sdk::spot::SpotRestApi::production(
rest_config.clone(),
)),
Arc::new(RateLimitTracker::new(WeightPool::Spot)),
));
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(WeightPool::Sapi)),
known_live: KnownLiveOrders::shared(ExchangeId::BinanceMargin),
rules,
}
}
async fn open_order(
&self,
request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
) -> 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| 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(),
request
.state
.margin_effect
.unwrap_or(self.config.margin_effect),
self.config.auto_repay_at_cancel,
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::Trailing(refused)) => {
return inactive(OrderState::inactive(refused.into_order_error()));
}
Err(BuildOrderError::Build(msg)) => {
error!(%msg, "BinanceMargin failed to build new order params");
return inactive(OrderState::inactive(OrderError::Rejected(
ApiError::OrderRejected(msg),
)));
}
};
if let Err(refused) = self
.rules
.check(instrument.name(), kind, price, quantity)
.await
{
return inactive(OrderState::inactive(refused));
}
let response =
match rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Order, |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 state = placed_order_state(
ExchangeId::BinanceMargin,
&instrument,
quantity,
PlacementResponse {
status: data.status.as_deref(),
order_id: exchange_order_id,
time_exchange,
executed_qty: data.executed_qty.as_deref(),
cumulative_quote_qty: data.cummulative_quote_qty.as_deref(),
},
);
self.known_live
.lock()
.placed(&order_key, quantity, kind, &state);
Order {
key: order_key,
side,
price,
quantity,
kind,
time_in_force,
state,
}
}
async fn cancel_order(
&self,
request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
) -> 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().and_then(VenueOrderId::assigned),
&request.key.cid,
self.config.is_isolated,
) {
Ok(p) => p,
Err(e) => {
error!(%e, "BinanceMargin failed to build cancel order params");
return UnindexedOrderResponseCancel {
key,
state: Err(OrderError::Rejected(ApiError::OrderRejected(e))),
};
}
};
let response =
match rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Order, |rest| {
let params = params.clone();
Box::pin(async move { rest.margin_account_cancel_order(params).await })
})
.await
{
Ok(response) => response,
Err(e) => {
return UnindexedOrderResponseCancel {
key,
state: Err(classify_rest_order_error(&e, &instrument)),
};
}
};
let data = match response.data().await {
Ok(data) => data,
Err(e) => {
return 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 UnindexedOrderResponseCancel {
key,
state: Err(OrderError::Rejected(ApiError::OrderRejected(
"cancel response missing orderId".into(),
))),
};
}
};
let filled_qty = binance_filled_qty(
ExchangeId::BinanceMargin,
&exchange_order_id,
data.executed_qty.as_deref(),
);
self.known_live.lock().ended(&key.cid);
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, RequestKind::Query, |rest| {
Box::pin(async move {
let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
rest.query_cross_margin_account_details(params).await
})
})
.await
.map_err(|e| classify_rest_query_error(&e, None))?;
let account = response.data().await.map_err(response_decode_error)?;
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,
RequestKind::Query,
)
}))
.buffer_unordered(8)
.map(|result| {
let (inst, listing) = result?;
self.remember_live(&listing.orders);
let wrapped = listing
.orders
.into_iter()
.map(active_order_snapshot)
.collect();
Ok::<_, UnindexedClientError>(InstrumentAccountSnapshot::new(
inst,
wrapped,
listing.complete,
PositionReport::Unreported,
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, RequestKind::Query, |rest| {
Box::pin(async move {
let params = QueryCrossMarginAccountDetailsParams::builder().build()?;
rest.query_cross_margin_account_details(params).await
})
})
.await
.map_err(|e| classify_rest_query_error(&e, None))?;
let account = response.data().await.map_err(response_decode_error)?;
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,
RequestKind::Query,
)
}))
.buffer_unordered(8)
.try_fold(
Vec::with_capacity(capacity),
|mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>,
(_, listing)| async move {
acc.extend(listing.orders);
Ok(acc)
},
)
.await
.inspect(|orders| self.remember_live(orders));
}
if instruments.is_empty() {
let orders =
fetch_margin_all_open_orders(self.rest.clone(), self.rate_limiter.clone(), false)
.await?;
self.remember_live(&orders);
return Ok(orders);
}
futures::stream::iter(instruments.iter().cloned().map(|instrument| {
fetch_margin_open_orders_for_instrument(
self.rest.clone(),
self.rate_limiter.clone(),
instrument,
false,
RequestKind::Query,
)
}))
.buffer_unordered(8)
.try_fold(
Vec::with_capacity(instruments.len()),
|mut acc: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, (_, listing)| async move {
acc.extend(listing.orders);
Ok(acc)
},
)
.await
.inspect(|orders| self.remember_live(orders))
}
async fn fetch_trades(
&self,
start: DateTime<Utc>,
end: DateTime<Utc>,
instruments: &[InstrumentNameExchange],
) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
use futures::StreamExt as _;
if start > end {
return Ok(TradesRead::complete(Vec::new()));
}
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(TradesRead::complete(Vec::new()));
}
let span = MyTradesFrom::Span {
start: start.timestamp_millis(),
end: end.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,
span,
is_isolated,
RequestKind::Query,
)
.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)
&& (start..=end).contains(&trade.time_exchange)
{
all_trades.push(trade);
}
}
}
Ok(TradesRead::complete(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 known_live = self.known_live.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?;
rate_limiter.before_stream_open().await?;
let live = isolated_connect_and_subscribe(
&ws_config,
&tx,
&dedup,
&known_live,
&base_quote,
&tokens,
)
.await
.map_err(|e| {
rate_limiter.stream_open_error(&e, |e| {
UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
})
})?;
tokio::spawn(isolated_connection_manager(
tx,
dedup,
ws_config,
rest_config,
rest,
rate_limiter,
known_live,
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?;
rate_limiter.before_stream_open().await?;
let initial_ws = connect_margin_ws(&ws_config).await.map_err(|e| {
rate_limiter.stream_open_error(&e, |e| {
UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
})
})?;
let risk = RiskStreamTask::spawn(
SapiRiskListenKeys {
rest: rest.clone(),
rate_limiter: rate_limiter.clone(),
},
RISK_STREAM_URL.to_owned(),
tx.clone(),
dedup.clone(),
);
let manager = margin_connection_manager(
tx,
dedup,
ws_config,
rest_config,
rest,
rate_limiter,
known_live,
instruments,
Some((initial_ws, initial_token)),
);
tokio::spawn(async move {
let _risk = risk;
manager.await;
})
};
let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
let guarded_stream = AbortOnDropStream::new(rx_stream, cm_handle);
Ok(futures::StreamExt::boxed(guarded_stream))
}
}
impl OrderStatusClient for BinanceMargin {
async fn fetch_ended_orders(
&self,
orders: &[UnindexedOrderKey],
) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError> {
fetch_ended_by_key(orders, |key| {
let recorded_kind = self.known_live.lock().kind_of(&key.cid);
fetch_margin_order_lookup(
self.rest.clone(),
self.rate_limiter.clone(),
key,
self.config.is_isolated,
RequestKind::Query,
recorded_kind,
)
})
.await
}
}
const HOURS_PER_YEAR: u32 = 24 * 365;
impl ShortabilityClient for BinanceMargin {
async fn fetch_shortability(
&self,
instrument: &InstrumentNameExchange,
) -> Result<Shortability, UnindexedClientError> {
let pair = self.fetch_margin_pair(instrument).await?;
let (borrowable, hourly_rate, delisting_at) = tokio::try_join!(
self.fetch_asset_borrowable(&pair.base),
self.fetch_next_hourly_interest_rate(&pair.base),
self.fetch_delisting_at(instrument, &pair),
)?;
let mut shortability = Shortability::new();
if let Some(delisting_at) = delisting_at {
shortability = shortability.with_delisting_at(delisting_at);
}
let not_delisted = Some(delisting_at.is_none_or(|delisting_at| delisting_at > Utc::now()));
let flags = [
pair.is_margin_trade,
pair.is_sell_allowed,
borrowable,
not_delisted,
];
if flags.contains(&Some(false)) {
shortability = shortability.with_shortable(false);
} else if flags.iter().all(|flag| *flag == Some(true)) {
shortability = shortability.with_shortable(true);
}
if let Some(hourly_rate) = hourly_rate {
shortability = shortability.with_fee_rate(hourly_rate * Decimal::from(HOURS_PER_YEAR));
}
Ok(shortability)
}
}
impl BorrowCapacityClient for BinanceMargin {
async fn fetch_borrow_capacity(
&self,
instrument: &InstrumentNameExchange,
side: Side,
) -> Result<BorrowCapacity, UnindexedClientError> {
let pair = self.fetch_margin_pair(instrument).await?;
let asset = match side {
Side::Buy => pair.quote,
Side::Sell => pair.base,
};
let isolated_symbol = self
.config
.is_isolated
.then(|| instrument.name().to_string());
let response =
rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
let asset = asset.name().to_string();
let isolated_symbol = isolated_symbol.clone();
Box::pin(async move {
let params = QueryMaxBorrowParams::builder(asset)
.isolated_symbol(isolated_symbol)
.build()?;
rest.query_max_borrow(params).await
})
})
.await
.map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
let max_borrow = response.data().await.map_err(response_decode_error)?;
let mut capacity = BorrowCapacity::new(asset);
if let Some(amount) = optional_decimal("amount", max_borrow.amount.as_deref())? {
capacity = capacity.with_borrowable_now(amount);
}
if let Some(limit) = optional_decimal("borrowLimit", max_borrow.borrow_limit.as_deref())? {
capacity = capacity.with_account_limit(limit);
}
Ok(capacity)
}
}
#[derive(Debug)]
struct MarginPair {
base: AssetNameExchange,
quote: AssetNameExchange,
is_margin_trade: Option<bool>,
is_sell_allowed: Option<bool>,
}
impl BinanceMargin {
async fn fetch_margin_pair(
&self,
instrument: &InstrumentNameExchange,
) -> Result<MarginPair, UnindexedClientError> {
let symbol = instrument.name().to_string();
let listed = if self.config.is_isolated {
let response =
rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
let symbol = symbol.clone();
Box::pin(async move {
let params = GetAllIsolatedMarginSymbolParams::builder()
.symbol(symbol)
.build()?;
rest.get_all_isolated_margin_symbol(params).await
})
})
.await
.map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
response
.data()
.await
.map_err(response_decode_error)?
.into_iter()
.find(|pair| pair.symbol.as_deref() == Some(symbol.as_str()))
.map(|pair| {
(
pair.base,
pair.quote,
pair.is_margin_trade,
pair.is_sell_allowed,
)
})
} else {
let response =
rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
let symbol = symbol.clone();
Box::pin(async move {
let params = GetAllCrossMarginPairsParams::builder()
.symbol(symbol)
.build()?;
rest.get_all_cross_margin_pairs(params).await
})
})
.await
.map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
response
.data()
.await
.map_err(response_decode_error)?
.into_iter()
.find(|pair| pair.symbol.as_deref() == Some(symbol.as_str()))
.map(|pair| {
(
pair.base,
pair.quote,
pair.is_margin_trade,
pair.is_sell_allowed,
)
})
};
let margin = if self.config.is_isolated {
"isolated"
} else {
"cross"
};
let Some((base, quote, is_margin_trade, is_sell_allowed)) = listed else {
return Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(
instrument.clone(),
format!("Binance lists no {margin} margin pair {symbol}"),
)));
};
let (Some(base), Some(quote)) = (base, quote) else {
return Err(UnindexedClientError::Internal(format!(
"Binance lists {margin} margin pair {symbol} without its base or quote asset"
)));
};
Ok(MarginPair {
base: AssetNameExchange::new(base),
quote: AssetNameExchange::new(quote),
is_margin_trade,
is_sell_allowed,
})
}
async fn fetch_asset_borrowable(
&self,
asset: &AssetNameExchange,
) -> Result<Option<bool>, UnindexedClientError> {
let name = asset.name().to_string();
let response =
rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
let name = name.clone();
Box::pin(async move {
let params = GetAllMarginAssetsParams::builder().asset(name).build()?;
rest.get_all_margin_assets(params).await
})
})
.await
.map_err(|e| classify_rest_query_error(&e, None))?;
Ok(response
.data()
.await
.map_err(response_decode_error)?
.into_iter()
.find(|listed| listed.asset_name.as_deref() == Some(name.as_str()))
.and_then(|listed| listed.is_borrowable))
}
async fn fetch_delisting_at(
&self,
instrument: &InstrumentNameExchange,
pair: &MarginPair,
) -> Result<Option<DateTime<Utc>>, UnindexedClientError> {
let response =
rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
Box::pin(async move {
let params = GetDelistScheduleParams::builder().build()?;
rest.get_delist_schedule(params).await
})
})
.await
.map_err(|e| classify_rest_query_error(&e, None))?;
let schedule = response.data().await.map_err(response_decode_error)?;
let symbol = instrument.name().as_str();
let (base, quote) = (pair.base.name().as_str(), pair.quote.name().as_str());
let mut earliest: Option<DateTime<Utc>> = None;
for entry in schedule {
let scheduled = if self.config.is_isolated {
entry
.isolated_margin_symbols
.iter()
.flatten()
.any(|listed| listed == symbol)
} else {
entry
.cross_margin_assets
.iter()
.flatten()
.any(|listed| listed == base || listed == quote)
};
if !scheduled {
continue;
}
let delisting_at = entry
.delist_time
.and_then(DateTime::from_timestamp_millis)
.ok_or_else(|| {
UnindexedClientError::Internal(format!(
"decoding Binance response: delist-schedule lists {symbol} with delistTime \
{:?}",
entry.delist_time
))
})?;
earliest = Some(earliest.map_or(delisting_at, |earliest| earliest.min(delisting_at)));
}
Ok(earliest)
}
async fn fetch_next_hourly_interest_rate(
&self,
asset: &AssetNameExchange,
) -> Result<Option<Decimal>, UnindexedClientError> {
let name = asset.name().to_string();
let is_isolated = self.config.is_isolated;
let response =
rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
let name = name.clone();
Box::pin(async move {
let params = GetFutureHourlyInterestRateParams::builder(
name,
IsolatedFlag::from_flag(is_isolated),
)
.build()?;
rest.get_future_hourly_interest_rate(params).await
})
})
.await
.map_err(|e| classify_rest_query_error(&e, None))?;
let rate = response
.data()
.await
.map_err(response_decode_error)?
.into_iter()
.find(|listed| listed.asset.as_deref() == Some(name.as_str()))
.and_then(|listed| listed.next_hourly_interest_rate);
optional_decimal("nextHourlyInterestRate", rate.as_deref())
}
}
fn optional_decimal(
field: &str,
value: Option<&str>,
) -> Result<Option<Decimal>, UnindexedClientError> {
value
.map(|value| {
Decimal::from_str(value).map_err(|error| {
UnindexedClientError::Internal(format!(
"decoding Binance response: {field} {value:?}: {error}"
))
})
})
.transpose()
}
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, RequestKind::Essential, |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(|e| classify_rest_query_error(&e, symbol))?;
let data = response.data().await.map_err(response_decode_error)?;
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<DisconnectOnDrop<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(DisconnectOnDrop::new(ws, "BinanceMargin account stream"))
}
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(subscribe_error)?;
Ok(())
}
fn subscribe_error(e: WebsocketError) -> anyhow::Error {
let message = format!("userDataStream.subscribe.listenToken failed: {e}");
anyhow::Error::new(e).context(message)
}
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(subscribe_error)?;
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);
}
fn cross_notice_instrument(_subscription_id: Option<i64>) -> Option<InstrumentNameExchange> {
None
}
fn isolated_notice_instrument(
subscription_id: Option<i64>,
sub_map: &Mutex<HashMap<i64, InstrumentNameExchange>>,
) -> Option<InstrumentNameExchange> {
let Some(sub_id) = subscription_id else {
warn!(
"BinanceMargin isolated: MARGIN_LEVEL_STATUS_CHANGE without subscriptionId — \
delivering the notice without an instrument"
);
return None;
};
let instrument = sub_map
.lock()
.unwrap_or_else(|p| p.into_inner())
.get(&sub_id)
.cloned();
if instrument.is_none() {
warn!(
subscription_id = sub_id,
"BinanceMargin isolated: MARGIN_LEVEL_STATUS_CHANGE for unmapped subscriptionId — \
delivering the notice without an instrument"
);
}
instrument
}
#[cfg(test)]
fn convert_margin_user_data_events(frame: &str, buf: &mut Vec<UnindexedAccountEvent>) -> bool {
matches!(
convert_margin_user_data_events_with(
frame,
buf,
&mut cross_account_position_handler,
&mut cross_notice_instrument,
),
FrameOutcome::Pushed { terminated: true }
)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FrameOutcome {
Response,
Pushed { terminated: bool },
}
fn convert_margin_user_data_events_with(
frame: &str,
buf: &mut Vec<UnindexedAccountEvent>,
handle_position: &mut impl FnMut(
OutboundAccountPosition,
Option<i64>,
&mut Vec<UnindexedAccountEvent>,
),
notice_instrument: &mut impl FnMut(Option<i64>) -> Option<InstrumentNameExchange>,
) -> FrameOutcome {
let (subscription_id, event_type, event_raw) = match parse_user_data_frame(frame) {
UserDataFrame::Response => return FrameOutcome::Response,
UserDataFrame::Event {
subscription_id,
event_type,
event,
} => (subscription_id, event_type, event),
UserDataFrame::Unrecognised => {
static SEEN: AtomicU64 = AtomicU64::new(0);
log_unrecognised_frame("BinanceMargin", &SEEN, frame);
return FrameOutcome::Pushed { terminated: false };
}
};
let terminated = match event_type {
"executionReport" => {
match serde_json::from_str::<ExecutionReport>(event_raw) {
Ok(report) => {
debug!(
symbol = report.symbol.as_deref().unwrap_or("?"),
order_id = ?report.order_id,
client_order_id = report.client_order_id.as_deref().unwrap_or("?"),
order_type = report.order_type.as_deref().unwrap_or("?"),
execution_type = report.execution_type.as_deref().unwrap_or("?"),
order_status = report.order_status.as_deref().unwrap_or("?"),
"BinanceMargin executionReport"
);
convert_execution_report(&report, ExchangeId::BinanceMargin, buf);
}
Err(e) => {
warn!(error = %e, frame = frame_excerpt(event_raw), "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, frame = frame_excerpt(event_raw), "BinanceMargin: undeserializable outboundAccountPosition, dropping")
}
}
false
}
"balanceUpdate" => {
false
}
"USER_LIABILITY_CHANGE" | "userLiabilityChange" => {
match serde_json::from_str::<UserLiabilityChange>(event_raw) {
Ok(c) => log_liability_change(&c, "user data"),
Err(e) => warn!(
error = %e,
frame = frame_excerpt(event_raw),
"BinanceMargin: undeserializable USER_LIABILITY_CHANGE, dropping"
),
}
false
}
"MARGIN_LEVEL_STATUS_CHANGE" | "marginLevelStatusChange" => {
match serde_json::from_str::<MarginLevelStatusChange>(event_raw) {
Ok(change) => {
log_margin_level_change(&change, "user data");
buf.extend(margin_level_notice(
change,
notice_instrument(subscription_id),
));
}
Err(e) => warn!(
error = %e,
frame = frame_excerpt(event_raw),
"BinanceMargin: undeserializable MARGIN_LEVEL_STATUS_CHANGE, dropping"
),
}
false
}
"eventStreamTerminated" => {
warn!("BinanceMargin user data stream terminated by exchange, signalling reconnect");
true
}
"listStatus" => {
trace!("BinanceMargin ignoring listStatus user data event");
false
}
other => {
static SEEN: UnhandledEvents = UnhandledEvents::new();
log_unhandled_event("BinanceMargin", &SEEN, other, event_raw);
false
}
};
FrameOutcome::Pushed { terminated }
}
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);
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;
};
if asset != base_asset.name().as_str() && asset != quote_asset.name().as_str() {
debug!(%asset, instrument = %instrument.name(), "BinanceMargin isolated account position: skipping asset outside the pair");
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;
};
buf.push(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::InstrumentBalanceUpdate(InstrumentBalanceUpdate::new(
instrument.clone(),
AssetBalanceUpdate::new(
AssetNameExchange::new(asset),
BalanceUpdate::new(free, locked),
time_exchange,
),
)),
));
}
}
fn register_user_data_listener(
ws: &Arc<WsApiBase>,
tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: SharedDedupCache,
known: SharedKnownLiveOrders,
heartbeat_flag: Arc<AtomicBool>,
signal_tx: oneshot::Sender<()>,
mut handle_position: impl FnMut(
OutboundAccountPosition,
Option<i64>,
&mut Vec<UnindexedAccountEvent>,
) + Send
+ 'static,
mut notice_instrument: impl FnMut(Option<i64>) -> Option<InstrumentNameExchange> + Send + 'static,
) -> UnsubscribeOnDrop {
let mut signal_tx_opt = Some(signal_tx);
let mut event_tx = Some(tx);
let mut event_buf = Vec::with_capacity(32);
UnsubscribeOnDrop::new(ws.common.events.subscribe(move |event| {
let Some(ref sender) = event_tx else { return };
match event {
WebsocketEvent::Message(json_str) => {
let terminated = match convert_margin_user_data_events_with(
&json_str,
&mut event_buf,
&mut handle_position,
&mut notice_instrument,
) {
FrameOutcome::Response => return,
FrameOutcome::Pushed { terminated } => {
heartbeat_flag.store(true, Ordering::Release);
terminated
}
};
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;
}
let held = KnownLiveOrders::observes(&ev.kind).then(|| {
let mut held = known.lock();
held.observe(&ev);
held
});
if sender.send(ev).is_err() {
drop(held);
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>,
unrecovered: &mut UnrecoveredFills,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: &SharedDedupCache,
known: &SharedKnownLiveOrders,
is_isolated: bool,
) {
use futures::StreamExt as _;
let due = unrecovered.due(tokio::time::Instant::now());
let Some(oldest) = due.iter().map(|(_, gap)| gap.start_ms).min() else {
return;
};
info!(
gaps = due.len(),
oldest = %gap_time(oldest),
is_isolated,
"BinanceMargin recovering fills missed while disconnected"
);
let mut settled = vec![false; due.len()];
let started = AtomicUsize::new(0);
let recovery = async {
let order_executions_deadline = tokio::time::Instant::now() + ORDER_EXECUTIONS_BUDGET;
let mut recovered = 0u32;
let mut duplicates = 0u32;
let mut stream =
futures::stream::iter(due.iter().cloned().enumerate().map(|(index, (inst, gap))| {
started.store(index + 1, Ordering::Relaxed);
let rest = rest.clone();
let rl = rate_limiter.clone();
async move {
let span = MyTradesFrom::Span {
start: gap.start_ms,
end: gap.end_ms,
};
let raw = match paginate_margin_my_trades(
&rest,
&rl,
&inst,
span,
is_isolated,
RequestKind::Essential,
)
.await
{
Ok(pages) => pages,
Err(e) => return (index, Err(e)),
};
let totals = recovered_order_totals(
ExchangeId::BinanceMargin,
&inst,
&raw,
order_executions_deadline,
|order_id| {
paginate_margin_my_trades(
&rest,
&rl,
&inst,
MyTradesFrom::Order(order_id),
is_isolated,
RequestKind::Essential,
)
},
)
.await;
let cids = recovered_order_cids(
ExchangeId::BinanceMargin,
&inst,
&raw,
order_executions_deadline,
|order_id| fetch_margin_order_cid(&rest, &rl, &inst, order_id, is_isolated),
)
.await;
let trades = raw
.iter()
.filter_map(|t| {
let mut trade = convert_margin_trade(t, &inst)?;
trade.order_filled_quantity =
t.id.and_then(|id| totals.get(&id).copied());
trade.client_order_id =
t.order_id.and_then(|id| cids.get(&id).cloned());
Some(trade)
})
.collect::<Vec<_>>();
(index, Ok(trades))
}
}))
.buffer_unordered(8);
while let Some((index, result)) = stream.next().await {
let (inst, gap) = &due[index];
settled[index] = true;
let trades = match result {
Ok(trades) => trades,
Err(e) => {
gap_failed(
unrecovered,
tx,
ExchangeId::BinanceMargin,
inst,
gap,
FillRecoveryFailure::Request(e.to_string()),
);
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;
}
let sent = {
let mut known = known.lock();
known.observe(&event);
tx.send(event)
};
if sent.is_err() {
debug!("BinanceMargin fill recovery: consumer dropped during recovery");
return;
}
recovered += 1;
}
unrecovered.recovered(inst, gap);
}
info!(
recovered,
duplicates, is_isolated, "BinanceMargin fill recovery complete"
);
};
if tokio::time::timeout(Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS), recovery)
.await
.is_err()
{
let started = started.load(Ordering::Relaxed);
for ((inst, gap), _) in due[..started]
.iter()
.zip(&settled)
.filter(|(_, settled)| !**settled)
{
gap_failed(
unrecovered,
tx,
ExchangeId::BinanceMargin,
inst,
gap,
FillRecoveryFailure::TimedOut {
timeout_secs: FILL_RECOVERY_TIMEOUT_SECS,
},
);
}
}
}
fn open_reconnect_recovery(
unrecovered: &mut UnrecoveredFills,
unchecked: &mut UncheckedOrders,
known: &SharedKnownLiveOrders,
instruments: &[InstrumentNameExchange],
disconnected: DateTime<Utc>,
) {
unrecovered.open(instruments, disconnected, Utc::now());
let held = known.lock().instruments_among(instruments);
unchecked.open(held);
}
#[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>,
known: SharedKnownLiveOrders,
instruments: Vec<InstrumentNameExchange>,
initial: Option<(DisconnectOnDrop<Arc<WsApiBase>>, UserListenToken)>,
) {
enum DisconnectReason {
Signal,
HeartbeatTimeout,
TokenRefresh,
ConsumerDropped,
}
let mut budget = ReconnectBudget::new("Binance", BackoffLimits::ACCOUNT_STREAM);
let mut disconnect_time: Option<DateTime<Utc>> = None;
let mut unrecovered = UnrecoveredFills::default();
let mut unchecked = UncheckedOrders::default();
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 => {
if let Err(reason) = rate_limiter.before_stream_connect().await {
emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
break;
}
match acquire_user_listen_token(&rest_config, &rate_limiter, None).await {
Ok(t) => t,
Err(e) => {
error!(%e, "BinanceMargin userListenToken acquisition failed");
if let Err(reason) =
retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await
{
emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
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");
rate_limiter.on_stream_refused(&e);
if let Err(reason) =
retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await
{
emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
break;
}
continue;
}
},
};
let (signal_tx, signal_rx) = oneshot::channel::<()>();
let heartbeat_flag = Arc::new(AtomicBool::new(false));
let subscription = register_user_data_listener(
&ws,
tx.clone(),
dedup.clone(),
known.clone(),
heartbeat_flag.clone(),
signal_tx,
cross_account_position_handler,
cross_notice_instrument,
);
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");
}
rate_limiter.on_stream_refused(&e);
if let Err(reason) = retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await {
emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
break;
}
continue;
}
info!("BinanceMargin account_stream connected and subscribed");
let token_deadline =
tokio::time::Instant::now() + token_renew_after(token.expiration_time_ms);
if let Some(dt) = disconnect_time.take() {
open_reconnect_recovery(&mut unrecovered, &mut unchecked, &known, &instruments, dt);
}
recover_margin_fills(
&rest,
&rate_limiter,
&mut unrecovered,
&tx,
&dedup,
&known,
false,
)
.await;
recover_margin_ended_orders(
&rest,
&rate_limiter,
&known,
&mut unchecked,
&unrecovered,
&tx,
false,
)
.await;
let reason = {
let mut signal_rx = signal_rx;
let retry_gaps = async {
loop {
let next_check = unchecked.next_due(&unrecovered);
match unrecovered.next_due().into_iter().chain(next_check).min() {
Some(due) => tokio::time::sleep_until(due).await,
None => std::future::pending::<()>().await,
}
recover_margin_fills(
&rest,
&rate_limiter,
&mut unrecovered,
&tx,
&dedup,
&known,
false,
)
.await;
recover_margin_ended_orders(
&rest,
&rate_limiter,
&known,
&mut unchecked,
&unrecovered,
&tx,
false,
)
.await;
}
};
tokio::pin!(retry_gaps);
let mut grace = true;
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)) => {
let in_grace = std::mem::take(&mut grace);
if heartbeat_flag.swap(false, Ordering::AcqRel) {
budget.reset();
continue;
}
if in_grace {
continue;
}
warn!("BinanceMargin heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
break DisconnectReason::HeartbeatTimeout;
}
() = &mut retry_gaps => {}
}
}
};
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) {
let heartbeat_timed_out = matches!(reason, DisconnectReason::HeartbeatTimeout);
let last_error = || disconnect_error(heartbeat_timed_out);
if let Err(reason) = retry_or_end(&mut budget, &rate_limiter, last_error).await {
emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
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: DisconnectOnDrop<Arc<WsApiBase>>,
subscription: UnsubscribeOnDrop,
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,
known: &SharedKnownLiveOrders,
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(false));
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 notice_instrument = {
let sub_map = sub_map.clone();
move |subscription_id| isolated_notice_instrument(subscription_id, &sub_map)
};
let subscription = register_user_data_listener(
&ws,
tx.clone(),
dedup.clone(),
known.clone(),
heartbeat_flag.clone(),
signal_tx,
handle_position,
notice_instrument,
);
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>,
known: SharedKnownLiveOrders,
symbols: Vec<InstrumentNameExchange>,
base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
initial: Option<IsolatedLiveConn>,
) {
enum DisconnectReason {
Signal,
HeartbeatTimeout,
TokenRefresh,
ConsumerDropped,
}
let mut budget = ReconnectBudget::new("Binance", BackoffLimits::ACCOUNT_STREAM);
let mut disconnect_time: Option<DateTime<Utc>> = None;
let mut unrecovered = UnrecoveredFills::default();
let mut unchecked = UncheckedOrders::default();
let mut current = initial;
loop {
let IsolatedLiveConn {
ws,
subscription,
signal_rx,
heartbeat_flag,
earliest_expiry_ms,
} = match current.take() {
Some(live) => live,
None => {
if let Err(reason) = rate_limiter.before_stream_connect().await {
emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
break;
}
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 let Err(reason) =
retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await
{
emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
break;
}
continue;
}
};
match isolated_connect_and_subscribe(
&ws_config,
&tx,
&dedup,
&known,
&base_quote,
&tokens,
)
.await
{
Ok(live) => live,
Err(e) => {
error!(%e, "BinanceMargin isolated connect/subscribe failed");
rate_limiter.on_stream_refused(&e);
if let Err(reason) =
retry_or_end(&mut budget, &rate_limiter, || e.to_string()).await
{
emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
break;
}
continue;
}
}
}
};
info!(
symbols = symbols.len(),
"BinanceMargin isolated account_stream connected and subscribed"
);
let token_deadline = tokio::time::Instant::now() + token_renew_after(earliest_expiry_ms);
if let Some(dt) = disconnect_time.take() {
open_reconnect_recovery(&mut unrecovered, &mut unchecked, &known, &symbols, dt);
}
recover_margin_fills(
&rest,
&rate_limiter,
&mut unrecovered,
&tx,
&dedup,
&known,
true,
)
.await;
recover_margin_ended_orders(
&rest,
&rate_limiter,
&known,
&mut unchecked,
&unrecovered,
&tx,
true,
)
.await;
let reason = {
let mut signal_rx = signal_rx;
let retry_gaps = async {
loop {
let next_check = unchecked.next_due(&unrecovered);
match unrecovered.next_due().into_iter().chain(next_check).min() {
Some(due) => tokio::time::sleep_until(due).await,
None => std::future::pending::<()>().await,
}
recover_margin_fills(
&rest,
&rate_limiter,
&mut unrecovered,
&tx,
&dedup,
&known,
true,
)
.await;
recover_margin_ended_orders(
&rest,
&rate_limiter,
&known,
&mut unchecked,
&unrecovered,
&tx,
true,
)
.await;
}
};
tokio::pin!(retry_gaps);
let mut grace = true;
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)) => {
let in_grace = std::mem::take(&mut grace);
if heartbeat_flag.swap(false, Ordering::AcqRel) {
budget.reset();
continue;
}
if in_grace {
continue;
}
warn!("BinanceMargin isolated heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
break DisconnectReason::HeartbeatTimeout;
}
() = &mut retry_gaps => {}
}
}
};
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) {
let heartbeat_timed_out = matches!(reason, DisconnectReason::HeartbeatTimeout);
let last_error = || disconnect_error(heartbeat_timed_out);
if let Err(reason) = retry_or_end(&mut budget, &rate_limiter, last_error).await {
emit_stream_terminated(&tx, ExchangeId::BinanceMargin, reason);
break;
}
}
}
}
async fn fetch_margin_open_orders_for_instrument(
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
instrument: InstrumentNameExchange,
is_isolated: bool,
kind: RequestKind,
) -> Result<(InstrumentNameExchange, OpenOrderListing), UnindexedClientError> {
let symbol_str = instrument.name().to_string();
let isolated = QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(is_isolated);
let response = rest_call_with_retry(&rest, &rate_limiter, kind, |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(|e| classify_rest_query_error(&e, Some(&instrument)))?;
let orders_data = response.data().await.map_err(response_decode_error)?;
let listing = convert_open_order_listing(&orders_data, ExchangeId::BinanceMargin, &instrument);
Ok((instrument, listing))
}
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 = QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(is_isolated);
let response = rest_call_with_retry(&rest, &rate_limiter, RequestKind::Query, |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(|e| classify_rest_query_error(&e, None))?;
let orders_data = response.data().await.map_err(response_decode_error)?;
let orders = orders_data
.into_iter()
.filter_map(|o| convert_open_order_owned_symbol(&o, ExchangeId::BinanceMargin))
.collect();
Ok(orders)
}
async fn fetch_margin_order_lookup(
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
key: UnindexedOrderKey,
is_isolated: bool,
kind: RequestKind,
recorded_kind: Option<OrderKind>,
) -> Result<OrderLookup, UnindexedClientError> {
let symbol = key.instrument.name().to_string();
let cid = key.cid.0.to_string();
let isolated = QueryMarginAccountsOrderIsIsolatedEnum::from_flag(is_isolated);
let response = match rest_call_with_retry(&rest, &rate_limiter, kind, |rest| {
let (symbol, cid, isolated) = (symbol.clone(), cid.clone(), isolated.clone());
Box::pin(async move {
let params = QueryMarginAccountsOrderParams::builder(symbol)
.orig_client_order_id(cid)
.is_isolated(isolated)
.build()?;
rest.query_margin_accounts_order(params).await
})
})
.await
{
Ok(response) => response,
Err(e) if is_unknown_order(&e) => {
debug!(instrument = %key.instrument, cid = %key.cid, error = %e, "BinanceMargin does not know this order");
return Ok(OrderLookup::Unknown);
}
Err(e) => return Err(classify_rest_query_error(&e, Some(&key.instrument))),
};
let row = response.data().await.map_err(response_decode_error)?;
Ok(
convert_ended_order(&row, ExchangeId::BinanceMargin, &key, recorded_kind)
.map_or(OrderLookup::NotEnded, |order| {
OrderLookup::Ended(Box::new(order))
}),
)
}
async fn fetch_margin_order_cid(
rest: &Arc<RestApi>,
rate_limiter: &RateLimitTracker,
instrument: &InstrumentNameExchange,
order_id: i64,
is_isolated: bool,
) -> Result<Option<ClientOrderId>, UnindexedClientError> {
let symbol = instrument.name().to_string();
let isolated = QueryMarginAccountsOrderIsIsolatedEnum::from_flag(is_isolated);
let response = match rest_call_with_retry(rest, rate_limiter, RequestKind::Essential, |rest| {
let (symbol, isolated) = (symbol.clone(), isolated.clone());
Box::pin(async move {
let params = QueryMarginAccountsOrderParams::builder(symbol)
.order_id(order_id)
.is_isolated(isolated)
.build()?;
rest.query_margin_accounts_order(params).await
})
})
.await
{
Ok(response) => response,
Err(e) if is_unknown_order(&e) => {
debug!(%instrument, order_id, error = %e, "BinanceMargin does not know this order");
return Ok(None);
}
Err(e) => return Err(classify_rest_query_error(&e, Some(instrument))),
};
let row = response.data().await.map_err(response_decode_error)?;
Ok(row
.client_order_id
.filter(|cid| !cid.is_empty())
.map(ClientOrderId::new))
}
async fn listed_margin_open_cids(
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
instruments: Vec<InstrumentNameExchange>,
is_isolated: bool,
) -> Result<FnvHashSet<ClientOrderId>, UnindexedClientError> {
let mut listed = FnvHashSet::default();
for instrument in instruments {
let (_, listing) = fetch_margin_open_orders_for_instrument(
rest.clone(),
rate_limiter.clone(),
instrument,
is_isolated,
RequestKind::Essential,
)
.await?;
listed.extend(listing.orders.into_iter().map(|order| order.key.cid));
}
Ok(listed)
}
async fn recover_margin_ended_orders(
rest: &Arc<RestApi>,
rate_limiter: &Arc<RateLimitTracker>,
known: &SharedKnownLiveOrders,
unchecked: &mut UncheckedOrders,
unrecovered: &UnrecoveredFills,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
is_isolated: bool,
) {
recover_ended_orders(
ExchangeId::BinanceMargin,
known,
unchecked,
unrecovered,
tx,
OpenListing::PerInstrument,
|instruments| {
listed_margin_open_cids(rest.clone(), rate_limiter.clone(), instruments, is_isolated)
},
|key| {
let recorded_kind = known.lock().kind_of(&key.cid);
fetch_margin_order_lookup(
rest.clone(),
rate_limiter.clone(),
key,
is_isolated,
RequestKind::Essential,
recorded_kind,
)
},
)
.await;
}
const MARGIN_MY_TRADES_WINDOW_MS: i64 = 23 * 60 * 60 * 1000;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MarginTradesQuery {
Window { start: i64, end: i64 },
FromId(i64),
Order(i64),
OrderFromId { order_id: i64, from_id: i64 },
}
impl MarginTradesQuery {
fn first(from: MyTradesFrom, now_ms: i64) -> Option<Self> {
match from {
MyTradesFrom::Span { start, .. } => {
(start <= now_ms).then(|| Self::window(start, now_ms))
}
MyTradesFrom::Order(order_id) => Some(Self::Order(order_id)),
}
}
fn window(start: i64, now_ms: i64) -> Self {
Self::Window {
start,
end: start.saturating_add(MARGIN_MY_TRADES_WINDOW_MS).min(now_ms),
}
}
fn next(self, page_len: usize, last_id: Option<i64>, now_ms: i64) -> Option<Self> {
let full = page_len >= BINANCE_MAX_TRADES;
let after_last = || last_id?.checked_add(1);
match self {
Self::Window { end, .. } if page_len == 0 => {
(end < now_ms).then(|| Self::window(end, now_ms))
}
Self::Window { end, .. } if end >= now_ms && !full => None,
Self::Window { .. } => after_last().map(Self::FromId),
Self::FromId(_) if full => after_last().map(Self::FromId),
Self::Order(order_id) | Self::OrderFromId { order_id, .. } if full => {
after_last().map(|from_id| Self::OrderFromId { order_id, from_id })
}
Self::FromId(_) | Self::Order(_) | Self::OrderFromId { .. } => None,
}
}
}
async fn paginate_margin_my_trades(
rest: &Arc<RestApi>,
rate_limiter: &Arc<RateLimitTracker>,
instrument: &InstrumentNameExchange,
from: MyTradesFrom,
is_isolated: bool,
kind: RequestKind,
) -> Result<Vec<QueryMarginAccountsTradeListResponseInner>, UnindexedClientError> {
let symbol_str = instrument.name().to_string();
let isolated = QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(is_isolated);
#[allow(clippy::cast_possible_wrap)]
let limit = BINANCE_MAX_TRADES as i64;
let now_ms = match from {
MyTradesFrom::Span { end, .. } => end.min(Utc::now().timestamp_millis()),
MyTradesFrom::Order(_) => Utc::now().timestamp_millis(),
};
let mut all_pages = Vec::new();
let mut requests = 0u32;
let mut next = MarginTradesQuery::first(from, now_ms);
while let Some(query) = next {
requests += 1;
let response = rest_call_with_retry(rest, rate_limiter, kind, |rest| {
let sym = symbol_str.clone();
let isolated = isolated.clone();
Box::pin(async move {
let builder = QueryMarginAccountsTradeListParams::builder(sym)
.is_isolated(isolated)
.limit(limit);
let params = match query {
MarginTradesQuery::Window { start, end } => {
builder.start_time(start).end_time(end)
}
MarginTradesQuery::FromId(id) => builder.from_id(id),
MarginTradesQuery::Order(order_id) => builder.order_id(order_id),
MarginTradesQuery::OrderFromId { order_id, from_id } => {
builder.order_id(order_id).from_id(from_id)
}
}
.build()?;
rest.query_margin_accounts_trade_list(params).await
})
})
.await
.map_err(|e| classify_rest_query_error(&e, Some(instrument)))?;
let mut page = response.data().await.map_err(response_decode_error)?;
let page_len = page.len();
let last_id = page.last().and_then(|t| t.id);
let past_end = match from {
MyTradesFrom::Span { end, .. } => drop_after(&mut page, end),
MyTradesFrom::Order(_) => false,
};
all_pages.extend(page);
if past_end {
break;
}
if page_len > 0 && last_id.is_none() {
warn!(%instrument, "BinanceMargin paginate_my_trades: trade missing ID, cannot page past it");
break;
}
next = query.next(page_len, last_id, now_ms);
}
debug!(
%instrument,
requests,
trades = all_pages.len(),
"BinanceMargin paginate_my_trades: walk complete"
);
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, RequestKind::Query, |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(|e| classify_rest_query_error(&e, None))?;
let info = response.data().await.map_err(response_decode_error)?;
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_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,
None,
AssetFees::new(fee_asset, commission, None),
))
}
#[derive(Debug)]
enum BuildOrderError {
Unsupported,
Build(String),
Trailing(TrailingOrderError),
}
trait IsolatedFlag {
fn from_flag(is_isolated: bool) -> Self;
}
macro_rules! impl_isolated_flag {
($($t:ty),* $(,)?) => {
$(
impl IsolatedFlag for $t {
fn from_flag(is_isolated: bool) -> Self {
if is_isolated { Self::True } else { Self::False }
}
}
)*
};
}
impl_isolated_flag!(
MarginAccountNewOrderIsIsolatedEnum,
MarginAccountCancelOrderIsIsolatedEnum,
QueryMarginAccountsOpenOrdersIsIsolatedEnum,
QueryMarginAccountsOrderIsIsolatedEnum,
QueryMarginAccountsTradeListIsIsolatedEnum,
GetFutureHourlyInterestRateIsIsolatedEnum,
);
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(
MarginAccountCancelOrderIsIsolatedEnum::from_flag(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,
margin_effect: MarginEffect,
auto_repay_at_cancel: bool,
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)
.quantity(quantity)
.is_isolated(MarginAccountNewOrderIsIsolatedEnum::from_flag(is_isolated))
.side_effect_type(binance_side_effect(margin_effect))
.new_client_order_id(new_client_order_id)
.new_order_resp_type(MarginAccountNewOrderNewOrderRespTypeEnum::Full);
if margin_effect == MarginEffect::Borrow {
builder = builder.auto_repay_at_cancel(auto_repay_at_cancel);
}
if let Some(tif) = binance_tif {
builder = builder.time_in_force(tif);
}
if let Some(trailing) = trailing_params(&kind, price).map_err(BuildOrderError::Trailing)? {
builder = builder.trailing_delta(i64::from(trailing.trailing_delta));
if let Some(limit) = trailing.price {
builder = builder.price(limit);
}
if let Some(activation_price) = trailing.stop_price {
builder = builder.stop_price(activation_price);
}
}
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);
}
OrderKind::TrailingStop { .. }
| OrderKind::TrailingStopFixedLimit { .. }
| OrderKind::Market
| OrderKind::TrailingStopLimit { .. } => {}
}
builder
.build()
.map_err(|e| BuildOrderError::Build(e.to_string()))
}
fn convert_order_kind_tif_margin(
kind: OrderKind,
tif: TimeInForce,
) -> Option<(
MarginAccountNewOrderTypeEnum,
Option<MarginAccountNewOrderTimeInForceEnum>,
)> {
let (binance_type, binance_tif) = classify_order_kind_tif(kind, tif)?;
let margin_type = match binance_type {
BinanceOrderType::Market => MarginAccountNewOrderTypeEnum::Market,
BinanceOrderType::Limit => MarginAccountNewOrderTypeEnum::Limit,
BinanceOrderType::LimitMaker => MarginAccountNewOrderTypeEnum::LimitMaker,
BinanceOrderType::StopLoss => MarginAccountNewOrderTypeEnum::StopLoss,
BinanceOrderType::StopLossLimit => MarginAccountNewOrderTypeEnum::StopLossLimit,
BinanceOrderType::TakeProfit => MarginAccountNewOrderTypeEnum::TakeProfit,
BinanceOrderType::TakeProfitLimit => MarginAccountNewOrderTypeEnum::TakeProfitLimit,
};
let margin_tif = binance_tif.map(|t| match t {
BinanceTimeInForce::Gtc => MarginAccountNewOrderTimeInForceEnum::Gtc,
BinanceTimeInForce::Ioc => MarginAccountNewOrderTimeInForceEnum::Ioc,
BinanceTimeInForce::Fok => MarginAccountNewOrderTimeInForceEnum::Fok,
});
Some((margin_type, margin_tif))
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
use super::*;
use crate::client::binance::shared::{
convert_open_order,
test_ws_api::{self, Lifetime},
unix_ms,
};
use crate::error::StreamTerminationReason;
use crate::notice::{AccountNotice, NoticeKind};
use crate::order::state::ActiveOrderState;
use binance_sdk::margin_trading::rest_api::QueryMarginAccountsOpenOrdersResponseInner;
#[tokio::test]
async fn a_banned_listen_token_subscribe_starts_the_cooldown() {
let until = i64::try_from(unix_ms()).unwrap() + 120_000;
let error = subscribe_error(WebsocketError::ResponseError {
code: -1003,
message: "Way too much request weight used".to_string(),
retry_after: Some(until),
});
assert!(
error
.to_string()
.starts_with("userDataStream.subscribe.listenToken failed: "),
"{error}"
);
let tracker = RateLimitTracker::new(WeightPool::Sapi);
assert!(tracker.on_stream_refused(&error));
assert!(
tracker
.cooldown_remaining()
.is_some_and(|left| left > Duration::from_secs(115)),
"{:?}",
tracker.cooldown_remaining()
);
}
struct ManagerRig {
ws_config: ConfigurationWebsocketApi,
rest_config: Arc<ConfigurationRestApi>,
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
requests: Arc<parking_lot::Mutex<Vec<tokio::time::Instant>>>,
}
impl ManagerRig {
async fn new(
answer: fn(&serde_json::Value) -> serde_json::Value,
lifetime: fn(usize) -> Lifetime,
) -> Self {
let (ws_url, requests) = test_ws_api::serve(answer, lifetime).await;
let rest_url = crate::client::test_http::serve(|method, path| {
if (method, path) != ("POST", "/sapi/v1/userListenToken") {
return (404, String::new());
}
let token = serde_json::json!({
"token": "token",
"expirationTime": Utc::now().timestamp_millis() + 86_400_000,
});
(200, token.to_string())
})
.await;
let rest_config = Arc::new(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(rest_url)
.build()
.unwrap(),
);
Self {
ws_config: ConfigurationWebsocketApi::builder()
.api_key("key")
.api_secret("secret")
.ws_url(ws_url)
.build()
.unwrap(),
rest: Arc::new(MarginTradingRestApi::from_config((*rest_config).clone())),
rest_config,
rate_limiter: Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
requests,
}
}
fn token() -> UserListenToken {
UserListenToken {
token: "token".to_string(),
expiration_time_ms: Utc::now().timestamp_millis() + 86_400_000,
}
}
async fn cross(&self) -> mpsc::UnboundedReceiver<UnindexedAccountEvent> {
let (tx, rx) = mpsc::unbounded_channel();
let ws = connect_margin_ws(&self.ws_config).await.unwrap();
tokio::spawn(margin_connection_manager(
tx,
new_dedup_cache(),
self.ws_config.clone(),
Arc::clone(&self.rest_config),
Arc::clone(&self.rest),
Arc::clone(&self.rate_limiter),
KnownLiveOrders::shared(ExchangeId::BinanceMargin),
Vec::new(),
Some((ws, Self::token())),
));
rx
}
async fn isolated(&self) -> mpsc::UnboundedReceiver<UnindexedAccountEvent> {
let (tx, rx) = mpsc::unbounded_channel();
let (dedup, known) = (
new_dedup_cache(),
KnownLiveOrders::shared(ExchangeId::BinanceMargin),
);
let symbol = InstrumentNameExchange::new("BTCUSDT");
let base_quote = Arc::new(HashMap::from([(
symbol.clone(),
(
AssetNameExchange::new("BTC"),
AssetNameExchange::new("USDT"),
),
)]));
let tokens = [(symbol.clone(), Self::token())];
let live = isolated_connect_and_subscribe(
&self.ws_config,
&tx,
&dedup,
&known,
&base_quote,
&tokens,
)
.await
.unwrap();
tokio::spawn(isolated_connection_manager(
tx,
dedup,
self.ws_config.clone(),
Arc::clone(&self.rest_config),
Arc::clone(&self.rest),
Arc::clone(&self.rate_limiter),
known,
vec![symbol],
base_quote,
Some(live),
));
rx
}
}
fn subscribed(request: &serde_json::Value) -> serde_json::Value {
serde_json::json!({
"id": request["id"],
"status": 200,
"result": { "subscriptionId": 0 },
})
}
fn banned_for_two_minutes(request: &serde_json::Value) -> serde_json::Value {
let until = unix_ms() + 120_000;
serde_json::json!({
"id": request["id"],
"status": 418,
"error": {
"code": -1003,
"msg": format!("Way too much request weight used; IP banned until {until}."),
"data": { "serverTime": unix_ms(), "retryAfter": until },
},
})
}
async fn rate_limited_end(rx: &mut mpsc::UnboundedReceiver<UnindexedAccountEvent>) -> Duration {
let event = tokio::time::timeout(Duration::from_secs(5), rx.recv())
.await
.expect("the stream ends rate-limited within the timeout")
.expect("it reports why before it ends");
let AccountEventKind::StreamTerminated(StreamTerminationReason::RateLimited {
retry_after,
}) = event.kind
else {
panic!("expected a rate-limited termination, got {:?}", event.kind);
};
assert!(rx.recv().await.is_none(), "the stream has ended");
retry_after
}
#[tokio::test]
async fn a_banned_subscribe_ends_the_cross_margin_stream_rate_limited() {
let rig = ManagerRig::new(banned_for_two_minutes, |_| Lifetime::Open).await;
let mut events = rig.cross().await;
let wait = rate_limited_end(&mut events).await;
assert!(wait > Duration::from_secs(115), "{wait:?}");
assert_eq!(rig.requests.lock().len(), 1, "no second subscribe");
}
#[tokio::test]
async fn a_banned_isolated_subscribe_is_recognised_as_a_refusal() {
let rig = ManagerRig::new(banned_for_two_minutes, |_| Lifetime::Open).await;
let (tx, _rx) = mpsc::unbounded_channel();
let symbol = InstrumentNameExchange::new("BTCUSDT");
let Err(error) = isolated_connect_and_subscribe(
&rig.ws_config,
&tx,
&new_dedup_cache(),
&KnownLiveOrders::shared(ExchangeId::BinanceMargin),
&Arc::new(HashMap::new()),
&[(symbol, ManagerRig::token())],
)
.await
else {
panic!("a banned subscribe fails");
};
assert!(rig.rate_limiter.on_stream_refused(&error), "{error:#}");
assert!(
rig.rate_limiter
.cooldown_remaining()
.is_some_and(|left| left > Duration::from_secs(115))
);
}
#[tokio::test]
async fn a_drop_during_a_long_cooldown_ends_the_isolated_margin_stream_rate_limited() {
let rig = ManagerRig::new(subscribed, |_| Lifetime::Silent(Duration::from_secs(1))).await;
let mut events = rig.isolated().await;
rig.rate_limiter
.on_rate_limited(Some(Duration::from_secs(60)));
let wait = rate_limited_end(&mut events).await;
assert!(wait > Duration::from_secs(55), "{wait:?}");
assert_eq!(rig.requests.lock().len(), 1, "no second subscribe");
}
#[tokio::test]
async fn a_dropped_margin_connection_backs_off_before_reconnecting() {
for isolated in [false, true] {
let rig =
ManagerRig::new(subscribed, |_| Lifetime::Silent(Duration::from_millis(100))).await;
let _events = if isolated {
rig.isolated().await
} else {
rig.cross().await
};
let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
while rig.requests.lock().len() < 2 {
assert!(
tokio::time::Instant::now() < deadline,
"isolated {isolated}: no reconnect"
);
tokio::time::sleep(Duration::from_millis(20)).await;
}
let requests = rig.requests.lock();
let between = requests[1] - requests[0];
assert!(
between >= BackoffLimits::ACCOUNT_STREAM.initial,
"isolated {isolated}: reconnected {between:?} after the first subscribe"
);
}
}
#[tokio::test(start_paused = true)]
async fn a_margin_reconnect_budget_resets_only_once_a_connection_proves_stable() {
async fn run(isolated: bool, pinging: bool) {
let lifetime = if pinging {
test_ws_api::dropped_then_pinging
} else {
test_ws_api::dropped_then_silent
};
let rig = ManagerRig::new(subscribed, lifetime).await;
let _events = if isolated {
rig.isolated().await
} else {
rig.cross().await
};
let label = format!("isolated {isolated}");
if pinging {
test_ws_api::assert_reset_once_pinged(&rig.requests, &label).await;
} else {
test_ws_api::assert_no_reset_when_silent(&rig.requests, &label).await;
}
}
tokio::join!(
run(false, true),
run(true, true),
run(false, false),
run(true, false)
);
}
#[tokio::test]
async fn a_dropped_margin_connection_is_closed() {
use futures::StreamExt as _;
use tokio_tungstenite::tungstenite::Message;
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("ws://{}", listener.local_addr().unwrap());
let (closed_tx, mut closed) = mpsc::unbounded_channel();
tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap();
while let Some(Ok(message)) = ws.next().await {
if let Message::Close(_) = message {
let _ = closed_tx.send(());
return;
}
}
});
let config = ConfigurationWebsocketApi::builder()
.api_key("key")
.api_secret("secret")
.ws_url(url)
.build()
.unwrap();
drop(connect_margin_ws(&config).await.unwrap());
tokio::time::timeout(Duration::from_secs(5), closed.recv())
.await
.expect("the dropped connection is closed")
.unwrap();
}
#[test]
fn margin_effect_default_is_no_side_effect() {
assert_eq!(MarginEffect::default(), MarginEffect::NoSideEffect);
}
#[test]
fn margin_effect_wire_values() {
for (effect, wire) in [
(MarginEffect::NoSideEffect, "NO_SIDE_EFFECT"),
(MarginEffect::Borrow, "MARGIN_BUY"),
(MarginEffect::Repay, "AUTO_REPAY"),
] {
assert_eq!(binance_side_effect(effect).as_str(), wire);
}
}
#[test]
fn config_debug_redacts_secrets() {
let config = BinanceMarginConfig::new(
"my_api_key".to_string(),
"my_secret_key".to_string(),
false,
false,
Vec::new(),
MarginEffect::default(),
true,
);
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.margin_effect, MarginEffect::NoSideEffect);
assert!(config.auto_repay_at_cancel);
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.margin_effect, MarginEffect::NoSideEffect);
assert!(config.auto_repay_at_cancel);
}
#[test]
fn config_deserializes_explicit_margin_effect() {
let config: BinanceMarginConfig = serde_json::from_str(
r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true,"margin_effect":"borrow","auto_repay_at_cancel":false}"#,
)
.expect("deserialize");
assert!(config.is_isolated);
assert_eq!(config.margin_effect, MarginEffect::Borrow);
assert!(!config.auto_repay_at_cancel);
}
#[test]
fn config_refuses_the_retired_combined_effect() {
let parsed = serde_json::from_str::<BinanceMarginConfig>(
r#"{"api_key":"k","secret_key":"s","testnet":false,"margin_effect":"auto_borrow_repay"}"#,
);
assert!(parsed.is_err());
}
#[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.margin_effect, MarginEffect::NoSideEffect);
}
#[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(),
MarginEffect::default(),
true,
);
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_flag_maps_mode() {
assert_eq!(
MarginAccountNewOrderIsIsolatedEnum::from_flag(true).as_str(),
"TRUE"
);
assert_eq!(
MarginAccountNewOrderIsIsolatedEnum::from_flag(false).as_str(),
"FALSE"
);
assert_eq!(
MarginAccountCancelOrderIsIsolatedEnum::from_flag(true).as_str(),
"TRUE"
);
assert_eq!(
MarginAccountCancelOrderIsIsolatedEnum::from_flag(false).as_str(),
"FALSE"
);
assert_eq!(
QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(true).as_str(),
"TRUE"
);
assert_eq!(
QueryMarginAccountsOpenOrdersIsIsolatedEnum::from_flag(false).as_str(),
"FALSE"
);
assert_eq!(
QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(true).as_str(),
"TRUE"
);
assert_eq!(
QueryMarginAccountsTradeListIsIsolatedEnum::from_flag(false).as_str(),
"FALSE"
);
}
#[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_ref().map(|i| i.as_str()), 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_ref().map(|i| i.as_str()), 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,
margin_effect: MarginEffect,
) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
build_new_order_params(
"BTCUSDT".to_string(),
side,
price,
Decimal::from(2),
kind,
tif,
"cid-1".to_string(),
margin_effect,
true,
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(),
MarginEffect::Borrow,
)
.expect("build");
assert_eq!(p.symbol, "BTCUSDT");
assert_eq!(p.side.as_str(), "BUY");
assert_eq!(p.r#type.as_str(), "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(),
MarginEffect::Borrow,
true,
false,
)
.expect("build");
assert_eq!(
cross.is_isolated.as_ref().map(|i| i.as_str()),
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(),
MarginEffect::Borrow,
true,
true,
)
.expect("build");
assert_eq!(
isolated.is_isolated.as_ref().map(|i| i.as_str()),
Some("TRUE")
);
}
#[test]
fn side_effect_and_auto_repay_gating() {
let build = |effect, auto_repay_at_cancel| {
build_new_order_params(
"BTCUSDT".to_string(),
Side::Buy,
Some(Decimal::from(1)),
Decimal::from(2),
OrderKind::Limit,
gtc(),
"cid-1".to_string(),
effect,
auto_repay_at_cancel,
false,
)
.expect("build")
};
let wire = |params: &MarginAccountNewOrderParams| {
params
.side_effect_type
.as_ref()
.map(|s| s.as_str().to_owned())
};
let borrow = build(MarginEffect::Borrow, true);
assert_eq!(wire(&borrow).as_deref(), Some("MARGIN_BUY"));
assert_eq!(borrow.auto_repay_at_cancel, Some(true));
let borrow_off = build(MarginEffect::Borrow, false);
assert_eq!(borrow_off.auto_repay_at_cancel, Some(false));
let repay = build(MarginEffect::Repay, true);
assert_eq!(wire(&repay).as_deref(), Some("AUTO_REPAY"));
assert_eq!(repay.auto_repay_at_cancel, None);
let none = build(MarginEffect::NoSideEffect, true);
assert_eq!(wire(&none).as_deref(), Some("NO_SIDE_EFFECT"));
assert_eq!(none.auto_repay_at_cancel, None);
}
#[test]
fn new_order_market_has_no_price_or_tif() {
let p = params(
Side::Buy,
None,
OrderKind::Market,
gtc(),
MarginEffect::Borrow,
)
.expect("build");
assert_eq!(p.r#type.as_str(), "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 },
MarginEffect::Borrow,
)
.expect("build");
assert_eq!(p.r#type.as_str(), "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(),
MarginEffect::Borrow,
)
.expect("build");
assert_eq!(stop.r#type.as_str(), "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(),
MarginEffect::Borrow,
)
.expect("build");
assert_eq!(stop_limit.r#type.as_str(), "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(),
MarginEffect::Borrow,
)
.expect("build");
assert_eq!(take_profit.r#type.as_str(), "TAKE_PROFIT");
assert_eq!(take_profit.stop_price, Some(trigger));
}
#[test]
fn a_trailing_stop_maps_to_stop_loss_with_trailing_delta() {
let cases = [
(Decimal::from(25), TrailingOffsetType::BasisPoints, 25),
(Decimal::new(15, 1), TrailingOffsetType::Percentage, 150),
];
for (offset, offset_type, trailing_delta) in cases {
let p = params(
Side::Sell,
None,
OrderKind::TrailingStop {
offset,
offset_type,
activation_price: None,
},
gtc(),
MarginEffect::NoSideEffect,
)
.expect("build");
assert_eq!(p.r#type.as_str(), "STOP_LOSS");
assert_eq!(p.trailing_delta, Some(trailing_delta), "{offset_type:?}");
assert_eq!(p.stop_price, None);
assert_eq!(p.price, None);
assert!(p.time_in_force.is_none());
}
}
#[test]
fn trailing_orders_carry_their_activation_price_and_fixed_limit() {
let activation = Some(Decimal::from(50_000));
let trailing_stop = params(
Side::Sell,
None,
OrderKind::TrailingStop {
offset: Decimal::from(25),
offset_type: TrailingOffsetType::BasisPoints,
activation_price: activation,
},
gtc(),
MarginEffect::NoSideEffect,
)
.expect("build");
assert_eq!(trailing_stop.r#type.as_str(), "STOP_LOSS");
assert_eq!(trailing_stop.trailing_delta, Some(25));
assert_eq!(trailing_stop.stop_price, activation);
assert_eq!(trailing_stop.price, None);
for activation_price in [None, activation] {
let fixed_limit = params(
Side::Sell,
Some(Decimal::from(49_000)),
OrderKind::TrailingStopFixedLimit {
offset: Decimal::new(15, 1),
offset_type: TrailingOffsetType::Percentage,
activation_price,
},
TimeInForce::ImmediateOrCancel,
MarginEffect::NoSideEffect,
)
.expect("build");
assert_eq!(fixed_limit.r#type.as_str(), "STOP_LOSS_LIMIT");
assert_eq!(fixed_limit.trailing_delta, Some(150));
assert_eq!(fixed_limit.price, Some(Decimal::from(49_000)));
assert_eq!(fixed_limit.stop_price, activation_price);
assert_eq!(
fixed_limit.time_in_force.as_ref().map(|t| t.as_str()),
Some("IOC")
);
}
let absolute = params(
Side::Sell,
Some(Decimal::from(49_000)),
OrderKind::TrailingStopFixedLimit {
offset: Decimal::from(100),
offset_type: TrailingOffsetType::Absolute,
activation_price: None,
},
gtc(),
MarginEffect::NoSideEffect,
);
assert!(matches!(absolute, Err(BuildOrderError::Unsupported)));
let no_limit = params(
Side::Sell,
None,
OrderKind::TrailingStopFixedLimit {
offset: Decimal::from(25),
offset_type: TrailingOffsetType::BasisPoints,
activation_price: None,
},
gtc(),
MarginEffect::NoSideEffect,
);
assert!(matches!(
no_limit,
Err(BuildOrderError::Trailing(TrailingOrderError::LimitPrice(
None
)))
));
let zero_activation = params(
Side::Sell,
None,
OrderKind::TrailingStop {
offset: Decimal::from(25),
offset_type: TrailingOffsetType::BasisPoints,
activation_price: Some(Decimal::ZERO),
},
gtc(),
MarginEffect::NoSideEffect,
);
assert!(matches!(
zero_activation,
Err(BuildOrderError::Trailing(
TrailingOrderError::NonPositiveActivation(_)
))
));
}
#[test]
fn absolute_trailing_offsets_and_trailing_stop_limits_are_unsupported() {
let absolute = params(
Side::Sell,
None,
OrderKind::TrailingStop {
offset: Decimal::from(100),
offset_type: TrailingOffsetType::Absolute,
activation_price: None,
},
gtc(),
MarginEffect::Borrow,
);
assert!(matches!(absolute, Err(BuildOrderError::Unsupported)));
let trailing_stop_limit = params(
Side::Sell,
Some(Decimal::from(100)),
OrderKind::TrailingStopLimit {
offset: Decimal::from(100),
offset_type: TrailingOffsetType::BasisPoints,
stop_price: Decimal::from(95),
limit_offset: Decimal::from(10),
},
gtc(),
MarginEffect::Borrow,
);
assert!(matches!(
trailing_stop_limit,
Err(BuildOrderError::Unsupported)
));
}
#[test]
fn a_fractional_trailing_offset_is_refused() {
let fractional = params(
Side::Sell,
None,
OrderKind::TrailingStop {
offset: Decimal::new(155, 3),
offset_type: TrailingOffsetType::Percentage,
activation_price: None,
},
gtc(),
MarginEffect::Borrow,
);
assert!(matches!(
fractional,
Err(BuildOrderError::Trailing(TrailingOrderError::Delta(
crate::client::binance::shared::TrailingDeltaError::Invalid(_)
)))
));
}
#[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,
MarginEffect::Borrow,
)
.expect("build");
assert_eq!(p.time_in_force.as_ref().map(|t| t.as_str()), Some(expected));
}
}
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 client_debug_shows_mode_and_margin_effect_without_credentials() {
let debug = format!("{:?}", isolated_client(&["BTCUSDT"]));
assert!(debug.contains("BinanceMargin"));
assert!(debug.contains("is_isolated: true"));
assert!(debug.contains("margin_effect: NoSideEffect"));
assert!(!debug.contains("\"k\"") && !debug.contains("\"s\""));
}
#[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.status = Some("PARTIALLY_FILLED".to_string());
o
}
#[test]
fn margin_open_order_converts() {
let inst = InstrumentNameExchange::new("BTCUSDT");
let order = convert_open_order(
&open_order(Some(42), "BUY"),
ExchangeId::BinanceMargin,
&inst,
)
.expect("convert");
assert_eq!(order.key.exchange, ExchangeId::BinanceMargin);
assert_eq!(
order.state.id.assigned().map(|id| id.0.as_str()),
Some("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_that_is_not_live_is_dropped() {
let inst = InstrumentNameExchange::new("BTCUSDT");
let mut o = open_order(Some(42), "BUY");
o.status = Some("CANCELED".to_string());
assert!(convert_open_order(&o, ExchangeId::BinanceMargin, &inst).is_none());
}
#[test]
fn margin_open_order_missing_order_id_is_dropped() {
let inst = InstrumentNameExchange::new("BTCUSDT");
assert!(
convert_open_order(&open_order(None, "BUY"), ExchangeId::BinanceMargin, &inst)
.is_none()
);
}
#[test]
fn margin_open_order_prefers_update_time_over_creation_time() {
let inst = InstrumentNameExchange::new("BTCUSDT");
let mut updated = open_order(Some(42), "BUY");
updated.update_time = Some(1_700_000_005_000);
let order =
convert_open_order(&updated, ExchangeId::BinanceMargin, &inst).expect("convert");
assert_eq!(
order.state.time_exchange,
Utc.timestamp_millis_opt(1_700_000_005_000)
.single()
.unwrap(),
"update_time must win over time"
);
}
#[test]
fn margin_open_order_falls_back_to_creation_time_when_update_time_absent() {
let inst = InstrumentNameExchange::new("BTCUSDT");
let base = open_order(Some(42), "BUY");
assert!(base.update_time.is_none(), "fixture must omit update_time");
let order = convert_open_order(&base, ExchangeId::BinanceMargin, &inst).expect("convert");
assert_eq!(
order.state.time_exchange,
Utc.timestamp_millis_opt(1_700_000_000_000)
.single()
.unwrap(),
"absent update_time falls back to time"
);
}
#[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_open_order_owned_symbol(&o, ExchangeId::BinanceMargin).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_open_order_owned_symbol(&o, ExchangeId::BinanceMargin).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()
}
fn cross_outcome(frame: &str, buf: &mut Vec<UnindexedAccountEvent>) -> FrameOutcome {
convert_margin_user_data_events_with(
frame,
buf,
&mut cross_account_position_handler,
&mut cross_notice_instrument,
)
}
#[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_eq!(cross_outcome(&frame, &mut buf), FrameOutcome::Response);
assert!(buf.is_empty());
}
#[test]
fn margin_ws_unrecognised_frames_yield_no_events() {
for frame in [
r#"{"e":"eventStreamTerminated","E":1}"#,
r#"{"subscriptionId":1,"event":{"E":1}}"#,
] {
let mut buf = Vec::new();
assert_eq!(
cross_outcome(frame, &mut buf),
FrameOutcome::Pushed { terminated: false },
"{frame}"
);
assert!(buf.is_empty(), "{frame}");
}
}
#[test]
fn margin_ws_trailing_stop_report_decodes_as_a_trailing_stop() {
for trailing_delta in [serde_json::json!(25), serde_json::json!("25")] {
let frame = push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "STOP_LOSS",
"x": "NEW", "X": "NEW", "i": 12_345_i64, "c": "cid-1", "t": -1_i64,
"p": "0.00000000", "P": "0.00000000", "d": trailing_delta, "D": 1_700_000_000_000_i64,
"q": "2", "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));
let [event] = buf.as_slice() else {
panic!("one snapshot expected for {frame}, got {buf:?}");
};
let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
panic!("expected an order snapshot, got {event:?}");
};
assert_eq!(
snapshot.0.kind,
OrderKind::TrailingStop {
offset: Decimal::from(25),
offset_type: TrailingOffsetType::BasisPoints,
activation_price: None,
}
);
assert_eq!(snapshot.0.price, None);
}
}
#[test]
fn margin_ws_execution_report_trade_maps_to_trade_and_order_snapshot() {
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",
"p": "48000", "q": "2", "f": "GTC",
"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(),
2,
"a fill must produce the execution and the order snapshot, got: {buf:?}"
);
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));
assert_eq!(t.client_order_id, Some(ClientOrderId::new("cid-1")));
}
other => panic!("expected Trade, got {other:?}"),
}
assert_eq!(buf[1].exchange, ExchangeId::BinanceMargin);
match &buf[1].kind {
AccountEventKind::OrderSnapshot(snap) => {
assert_eq!(
snap.0.quantity,
Decimal::from(2),
"the order's total quantity"
);
match &snap.0.state {
OrderState::Active(ActiveOrderState::Open(open)) => {
assert_eq!(open.id.assigned().map(|id| id.0.as_str()), Some("12345"));
assert_eq!(
open.filled_quantity,
Decimal::new(5, 1),
"the snapshot must carry the order's cumulative filled quantity (z)"
);
assert_eq!(
open.time_exchange.timestamp_millis(),
1_700_000_000_000,
"stamped with the execution's transaction time (T), which is what \
gives the engine's staleness gate an ordering key"
);
}
other => panic!("expected an Open order, got {other:?}"),
}
}
other => panic!("expected OrderSnapshot, got {other:?}"),
}
}
#[test]
fn margin_trade_for_an_order_no_longer_working_emits_no_snapshot() {
for status in ["CANCELED", "EXPIRED", "REJECTED", "PENDING_CANCEL", "NEW"] {
let frame = push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
"x": "TRADE", "X": status, "i": 12_345_i64, "c": "cid-1",
"t": 99_i64, "l": "0.5", "L": "48000", "z": "0.5",
"p": "48000", "q": "2", "f": "GTC", "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,
"status {status} must produce the execution alone, got: {buf:?}"
);
assert!(
matches!(buf[0].kind, AccountEventKind::Trade(_)),
"the execution must still be reported for status {status}"
);
}
}
#[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_trade_without_a_client_order_id_reports_none() {
let frame = push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "SELL", "o": "MARKET",
"x": "TRADE", "X": "FILLED", "i": 12_345_i64,
"t": 99_i64, "l": "1", "L": "48000", "z": "1",
"q": "1", "n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
}));
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
let AccountEventKind::Trade(trade) = &buf[0].kind else {
panic!("expected Trade, got {:?}", buf[0].kind);
};
assert_eq!(trade.order_id.0.as_str(), "12345");
assert_eq!(trade.client_order_id, None);
}
#[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_liability_change_is_observable_only() {
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(
&push(documented_liability_change("USER_LIABILITY_CHANGE")),
&mut buf
));
assert!(buf.is_empty());
}
fn only_notice(buf: &[UnindexedAccountEvent]) -> &AccountNotice<InstrumentNameExchange> {
match buf {
[
UnindexedAccountEvent {
exchange: ExchangeId::BinanceMargin,
kind: AccountEventKind::Notice(notice),
},
] => notice,
other => panic!("expected one BinanceMargin notice, got {other:?}"),
}
}
#[test]
fn margin_ws_margin_level_change_is_forwarded_as_a_notice() {
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(
&push(documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE")),
&mut buf
));
assert_eq!(
only_notice(&buf),
&AccountNotice::new(
NoticeKind::MarginCall,
None,
Utc.timestamp_millis_opt(1_573_200_697_110).unwrap(),
Some(Decimal::new(125, 2)),
SmolStr::new("MARGIN_CALL"),
),
"cross names no instrument"
);
}
#[test]
fn margin_level_statuses_map_to_notice_kinds() {
for (status, kind) in [
("MARGIN_CALL", NoticeKind::MarginCall),
("PRE_LIQUIDATION", NoticeKind::LiquidationWarning),
("FORCE_LIQUIDATION", NoticeKind::Liquidation),
("NORMAL", NoticeKind::MarginRestored),
("EXCESSIVE", NoticeKind::MarginRestored),
("SOME_FUTURE_STATUS", NoticeKind::Other),
] {
let mut buf = Vec::new();
let frame = push(serde_json::json!({
"e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "1.5",
"s": status,
}));
convert_margin_user_data_events(&frame, &mut buf);
let notice = only_notice(&buf);
assert_eq!(notice.kind, kind, "{status}");
assert_eq!(notice.status, status, "the raw status is kept");
}
}
#[test]
fn a_margin_level_change_missing_parts_is_still_delivered_where_it_can_be() {
let mut buf = Vec::new();
convert_margin_user_data_events(
&push(serde_json::json!({
"e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "n/a",
"s": "FORCE_LIQUIDATION",
})),
&mut buf,
);
let notice = only_notice(&buf);
assert_eq!(notice.kind, NoticeKind::Liquidation);
assert_eq!(notice.margin_level, None);
let before = Utc::now();
let mut buf = Vec::new();
convert_margin_user_data_events(
&push(serde_json::json!({ "e": "MARGIN_LEVEL_STATUS_CHANGE", "s": "MARGIN_CALL" })),
&mut buf,
);
assert!(only_notice(&buf).time_exchange >= before);
let mut buf = Vec::new();
convert_margin_user_data_events(
&push(serde_json::json!({
"e": "MARGIN_LEVEL_STATUS_CHANGE", "E": 1_700_000_000_000_i64, "l": "1.5",
})),
&mut buf,
);
assert!(buf.is_empty());
}
fn documented_liability_change(e: &str) -> serde_json::Value {
serde_json::json!({
"e": e, "E": 1_573_200_697_110_i64, "a": "BTC", "t": "BORROW",
"p": "1.03453430", "i": "0",
})
}
fn documented_margin_level_change(e: &str) -> serde_json::Value {
serde_json::json!({
"e": e, "E": 1_573_200_697_110_i64, "l": "1.25", "s": "MARGIN_CALL",
})
}
#[test]
fn margin_ws_margin_specific_events_match_their_documented_names() {
for (e, frame, expected) in [
(
"USER_LIABILITY_CHANGE",
documented_liability_change("USER_LIABILITY_CHANGE"),
"USER_LIABILITY_CHANGE (observable",
),
(
"MARGIN_LEVEL_STATUS_CHANGE",
documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE"),
"MARGIN_LEVEL_STATUS_CHANGE received",
),
(
"userLiabilityChange",
documented_liability_change("userLiabilityChange"),
"USER_LIABILITY_CHANGE (observable",
),
(
"marginLevelStatusChange",
documented_margin_level_change("marginLevelStatusChange"),
"MARGIN_LEVEL_STATUS_CHANGE received",
),
] {
let mut buf = Vec::new();
let (terminated, messages) =
log_messages(|| convert_margin_user_data_events(&push(frame), &mut buf));
assert!(!terminated, "{e}");
assert_eq!(
buf.len(),
usize::from(expected.starts_with("MARGIN")),
"{e}"
);
assert!(
messages.iter().any(|m| m.contains(expected)),
"{e}: expected its own arm's log, got {messages:?}"
);
assert!(
!messages.iter().any(|m| m.contains("undeserializable")
|| m.contains("unhandled")
|| m.contains("does not handle")),
"{e}: {messages:?}"
);
}
}
fn log_messages<T>(f: impl FnOnce() -> T) -> (T, Vec<String>) {
use std::sync::{Arc, Mutex};
use tracing_subscriber::layer::SubscriberExt;
#[derive(Default)]
struct Messages(Arc<Mutex<Vec<String>>>);
struct Message<'a>(&'a mut String);
impl tracing::field::Visit for Message<'_> {
fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
if field.name() == "message" {
*self.0 = format!("{value:?}");
}
}
}
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Messages {
fn on_event(
&self,
event: &tracing::Event<'_>,
_: tracing_subscriber::layer::Context<'_, S>,
) {
let mut message = String::new();
event.record(&mut Message(&mut message));
self.0.lock().unwrap().push(message);
}
}
let layer = Messages::default();
let messages = Arc::clone(&layer.0);
let value =
tracing::subscriber::with_default(tracing_subscriber::registry().with(layer), f);
let messages = messages.lock().unwrap().clone();
(value, messages)
}
#[test]
fn margin_ws_unhandled_event_types_are_ignored_without_teardown() {
for (e, expected) in [
("listStatus", "ignoring listStatus"),
("someFutureEvent", "of a type this client does not handle"),
] {
let mut buf = Vec::new();
let frame = push(serde_json::json!({ "e": e, "E": 1_700_000_000_000_i64 }));
let (terminated, messages) =
log_messages(|| convert_margin_user_data_events(&frame, &mut buf));
assert!(!terminated, "{e}");
assert!(buf.is_empty(), "{e}");
assert!(
messages.iter().any(|m| m.contains(expected)),
"{e}: {messages:?}"
);
}
}
#[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", "p": "100", "q": "1", "f": "GTC",
"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.first().expect("a Trade event");
assert!(
matches!(event.kind, AccountEventKind::Trade(_)),
"the execution is the first of the two events a fill produces"
);
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 drive_ws_pipeline(frames: &[&str]) -> Vec<UnindexedAccountEvent> {
let cache = new_dedup_cache();
let mut delivered = Vec::new();
let mut buf = Vec::new();
for frame in frames {
let _ = convert_margin_user_data_events(frame, &mut buf);
for ev in buf.drain(..) {
if let Some(key) = dedup_key_from_event(&ev)
&& is_duplicate(&cache, key)
{
continue;
}
delivered.push(ev);
}
}
delivered
}
fn ws_new(cumulative_filled: &str) -> String {
push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
"x": "NEW", "X": "NEW", "i": 12_345_i64, "c": "client-1",
"q": "2", "p": "100", "f": "GTC", "z": cumulative_filled,
"T": 1_700_000_000_000_i64,
}))
}
fn ws_partial_fill(cumulative_filled: &str) -> String {
push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
"x": "TRADE", "X": "PARTIALLY_FILLED", "i": 12_345_i64, "c": "client-1",
"q": "2", "p": "100", "f": "GTC", "z": cumulative_filled,
"l": "1", "L": "100", "t": 555_i64, "n": "0.1", "N": "USDT",
"T": 1_700_000_001_000_i64,
}))
}
fn open_filled_quantities(events: &[UnindexedAccountEvent]) -> Vec<Decimal> {
events
.iter()
.filter_map(|ev| match &ev.kind {
AccountEventKind::OrderSnapshot(snap) => match &snap.0.state {
OrderState::Active(ActiveOrderState::Open(open)) => Some(open.filled_quantity),
_ => None,
},
_ => None,
})
.collect()
}
#[test]
fn margin_ws_fill_snapshot_survives_the_dedup_gate() {
let delivered = drive_ws_pipeline(&[&ws_new("0"), &ws_partial_fill("1")]);
assert_eq!(
delivered.len(),
3,
"expected the ack snapshot, then the fill's execution and its snapshot, got: {delivered:?}"
);
assert_eq!(
open_filled_quantities(&delivered),
vec![Decimal::ZERO, Decimal::ONE],
"the fill's snapshot must reach the consumer carrying the advanced filled quantity"
);
}
#[test]
fn margin_ws_replayed_frame_is_still_deduplicated() {
let delivered = drive_ws_pipeline(&[
&ws_new("0"),
&ws_partial_fill("1"),
&ws_new("0"),
&ws_partial_fill("1"),
]);
assert_eq!(
delivered.len(),
3,
"re-delivering both frames must add nothing, got: {delivered:?}"
);
}
#[test]
fn margin_ws_an_ack_and_its_fill_reporting_one_state_deliver_that_state_once() {
let delivered = drive_ws_pipeline(&[&ws_new("1"), &ws_partial_fill("1")]);
assert_eq!(
delivered.len(),
2,
"the ack snapshot and the execution, with the fill's duplicate snapshot collapsed, \
got: {delivered:?}"
);
assert!(
delivered
.iter()
.any(|ev| matches!(ev.kind, AccountEventKind::Trade(_))),
"the execution must reach the consumer regardless"
);
assert_eq!(
open_filled_quantities(&delivered),
vec![Decimal::ONE],
"one snapshot, carrying the state both reports agree on"
);
}
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 isolated_balances(buf: &[UnindexedAccountEvent]) -> Vec<(&str, &str, Decimal, Decimal)> {
buf.iter()
.map(|event| match &event.kind {
AccountEventKind::InstrumentBalanceUpdate(ibu) => (
ibu.instrument.name().as_str(),
ibu.balance.asset.name().as_str(),
ibu.balance.update.free,
ibu.balance.update.locked,
),
other => panic!("expected InstrumentBalanceUpdate, got {other:?}"),
})
.collect()
}
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_margin_level_change_names_its_subscriptions_pair_or_none() {
let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
let notice_on = |subscription_id: i64| {
let mut buf = Vec::new();
let frame = push_with_sub(
subscription_id,
documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE"),
);
assert_eq!(
convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut cross_account_position_handler,
&mut |id| isolated_notice_instrument(id, &sub_map),
),
FrameOutcome::Pushed { terminated: false }
);
only_notice(&buf).instrument.clone()
};
assert_eq!(notice_on(7), Some(btcusdt()));
assert_eq!(notice_on(99), None);
assert_eq!(isolated_notice_instrument(None, &sub_map), None);
}
#[test]
fn the_two_streams_key_one_notice_alike() {
use crate::client::binance::margin_risk::{RiskFrame, convert_risk_frame};
let change = documented_margin_level_change("MARGIN_LEVEL_STATUS_CHANGE");
let mut from_user_data = Vec::new();
convert_margin_user_data_events(&push(change.clone()), &mut from_user_data);
let mut from_risk = Vec::new();
let risk_frame = serde_json::json!({ "stream": "listen-key", "data": change });
assert_eq!(
convert_risk_frame(&risk_frame.to_string(), "listen-key", &mut from_risk),
RiskFrame::Continue
);
let dedup = new_dedup_cache();
let key = |events: &[UnindexedAccountEvent]| dedup_key_from_event(&events[0]).unwrap();
assert!(!is_duplicate(&dedup, key(&from_user_data)));
assert!(
is_duplicate(&dedup, key(&from_risk)),
"the risk data stream's copy is dropped"
);
}
#[test]
fn a_notice_from_both_streams_is_delivered_once() {
let dedup = new_dedup_cache();
let notice = |time_ms: i64| {
let mut buf = Vec::new();
convert_margin_user_data_events(
&push(serde_json::json!({
"e": "MARGIN_LEVEL_STATUS_CHANGE", "E": time_ms, "l": "1.1",
"s": "MARGIN_CALL",
})),
&mut buf,
);
buf.pop().unwrap()
};
let is_new = |event: &UnindexedAccountEvent| {
!is_duplicate(&dedup, dedup_key_from_event(event).unwrap())
};
assert!(is_new(¬ice(1_700_000_000_000)));
assert!(!is_new(¬ice(1_700_000_000_000)), "second copy dropped");
assert!(is_new(¬ice(1_700_000_060_000)), "a later notice is new");
}
#[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_eq!(
convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler,
&mut cross_notice_instrument,
),
FrameOutcome::Pushed { terminated: false }
);
assert_eq!(
isolated_balances(&buf),
vec![
("BTCUSDT", "USDT", Decimal::from(1000), Decimal::from(50)),
("BTCUSDT", "BTC", Decimal::new(5, 1), Decimal::new(1, 1)),
]
);
let time = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
assert!(buf.iter().all(|event| matches!(
&event.kind,
AccountEventKind::InstrumentBalanceUpdate(ibu) if ibu.balance.time_exchange == time
)));
}
#[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_eq!(
convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler,
&mut cross_notice_instrument,
),
FrameOutcome::Pushed { terminated: false }
);
assert!(buf.is_empty(), "unmapped subscriptionId frame is dropped");
}
#[test]
fn isolated_outbound_position_with_one_side_emits_that_side() {
for (listed, free) in [("BTC", Decimal::new(5, 1)), ("USDT", Decimal::from(20))] {
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": listed, "f": free.to_string(), "l": "0" }],
}),
);
let mut buf = Vec::new();
assert_eq!(
convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler,
&mut cross_notice_instrument,
),
FrameOutcome::Pushed { terminated: false }
);
assert_eq!(
isolated_balances(&buf),
vec![("BTCUSDT", listed, free, Decimal::ZERO)],
"only {listed} listed"
);
}
}
#[test]
fn isolated_outbound_position_skips_assets_outside_the_pair() {
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 mixed = push_with_sub(
7,
serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [{ "a": "BNB", "f": "3", "l": "0" }, { "a": "USDT", "f": "20", "l": "1" }],
}),
);
let mut buf = Vec::new();
assert_eq!(
convert_margin_user_data_events_with(
&mixed,
&mut buf,
&mut handler,
&mut cross_notice_instrument,
),
FrameOutcome::Pushed { terminated: false }
);
assert_eq!(
isolated_balances(&buf),
vec![("BTCUSDT", "USDT", Decimal::from(20), Decimal::ONE)]
);
let stray_only = push_with_sub(
7,
serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [{ "a": "BNB", "f": "3", "l": "0" }],
}),
);
let mut buf = Vec::new();
assert_eq!(
convert_margin_user_data_events_with(
&stray_only,
&mut buf,
&mut handler,
&mut cross_notice_instrument,
),
FrameOutcome::Pushed { terminated: false }
);
assert!(buf.is_empty(), "no asset of the pair → nothing emitted");
}
#[test]
fn isolated_outbound_position_unparseable_entry_keeps_the_other_side() {
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": "x", "l": "0" }, { "a": "USDT", "f": "20", "l": "1" }],
}),
);
let mut buf = Vec::new();
assert_eq!(
convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler,
&mut cross_notice_instrument,
),
FrameOutcome::Pushed { terminated: false }
);
assert_eq!(
isolated_balances(&buf),
vec![("BTCUSDT", "USDT", Decimal::from(20), Decimal::ONE)]
);
}
#[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", "p": "100", "q": "1", "f": "GTC",
"n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
}),
);
let mut buf = Vec::new();
assert_eq!(
convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler,
&mut cross_notice_instrument,
),
FrameOutcome::Pushed { terminated: false }
);
assert_eq!(buf.len(), 2, "the execution and its order snapshot");
match &buf[0].kind {
AccountEventKind::Trade(t) => assert_eq!(t.instrument.name().as_str(), "BTCUSDT"),
other => panic!("expected Trade, got {other:?}"),
}
match &buf[1].kind {
AccountEventKind::OrderSnapshot(snap) => {
assert_eq!(snap.0.key.instrument.name().as_str(), "BTCUSDT");
}
other => panic!("expected OrderSnapshot, got {other:?}"),
}
}
struct MarginMyTradesVenue {
trades: Vec<(i64, i64, i64)>,
end_exclusive: bool,
}
impl wiremock::Respond for MarginMyTradesVenue {
fn respond(&self, request: &wiremock::Request) -> wiremock::ResponseTemplate {
const DAY_MS: i64 = 24 * 60 * 60 * 1000;
let query: std::collections::HashMap<String, String> =
request.url.query_pairs().into_owned().collect();
let param = |name: &str| query.get(name).map(|v| v.parse::<i64>().unwrap());
let limit = usize::try_from(param("limit").unwrap_or(500)).unwrap();
let order = param("orderId");
let mut selected: Vec<&(i64, i64, i64)> = self
.trades
.iter()
.filter(|(_, _, order_id)| order.is_none_or(|order| *order_id == order))
.collect();
if let Some(from_id) = param("fromId") {
selected.retain(|(id, _, _)| *id >= from_id);
} else if order.is_none() {
let start = param("startTime").expect("a query by time sends startTime");
let end = match param("endTime") {
Some(end) if end - start >= DAY_MS => {
return wiremock::ResponseTemplate::new(400).set_body_json(
serde_json::json!({"code": -1127, "msg": "More than 24 hours between startTime and endTime."}),
);
}
Some(end) if self.end_exclusive => end - 1,
Some(end) => end,
None => start + DAY_MS - 1,
};
selected.retain(|(_, time, _)| (start..=end).contains(time));
}
let page: Vec<serde_json::Value> = selected
.into_iter()
.take(limit)
.map(|(id, time, order_id)| {
serde_json::json!({
"id": id, "symbol": "BTCUSDT", "time": time, "orderId": order_id,
"price": "50000", "qty": "0.01", "commission": "0",
"commissionAsset": "USDT", "isBuyer": true, "isMaker": false,
"isBestMatch": true, "isIsolated": true,
})
})
.collect();
wiremock::ResponseTemplate::new(200).set_body_json(page)
}
}
fn since(start: i64) -> MyTradesFrom {
MyTradesFrom::Span {
start,
end: i64::MAX,
}
}
async fn read_margin_trades(
venue: MarginMyTradesVenue,
from: MyTradesFrom,
) -> (Vec<i64>, Vec<std::collections::HashMap<String, String>>) {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
.respond_with(venue)
.mount(&server)
.await;
let rest = Arc::new(MarginTradingRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.build()
.unwrap(),
));
let read = paginate_margin_my_trades(
&rest,
&Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
&InstrumentNameExchange::new("BTCUSDT"),
from,
true,
RequestKind::Query,
)
.await
.unwrap();
let requests = server
.received_requests()
.await
.unwrap()
.iter()
.map(|request| request.url.query_pairs().into_owned().collect())
.collect();
(read.into_iter().map(|t| t.id.unwrap()).collect(), requests)
}
#[tokio::test]
async fn recovered_margin_fills_carry_their_orders_client_order_ids() {
recovered_margin_fills_carry_client_order_ids(true).await;
recovered_margin_fills_carry_client_order_ids(false).await;
}
async fn recovered_margin_fills_carry_client_order_ids(is_isolated: bool) {
let disconnect = Utc::now() - chrono::Duration::minutes(10);
let reconnect = Utc::now() - chrono::Duration::minutes(5);
let fill_time = (disconnect + chrono::Duration::minutes(1)).timestamp_millis();
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
.respond_with(MarginMyTradesVenue {
trades: vec![(1, fill_time, 1), (2, fill_time, 2)],
end_exclusive: false,
})
.mount(&server)
.await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.and(wiremock::matchers::query_param("orderId", "1"))
.and(wiremock::matchers::query_param(
"isIsolated",
if is_isolated { "TRUE" } else { "FALSE" },
))
.respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
serde_json::json!({"symbol": "BTCUSDT", "orderId": 1, "clientOrderId": "repay-1"}),
))
.mount(&server)
.await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.and(wiremock::matchers::query_param("orderId", "2"))
.respond_with(
wiremock::ResponseTemplate::new(400).set_body_json(
serde_json::json!({"code": -2013, "msg": "Order does not exist."}),
),
)
.mount(&server)
.await;
let rest = Arc::new(MarginTradingRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.build()
.unwrap(),
));
let (tx, mut rx) = mpsc::unbounded_channel();
let mut unrecovered = UnrecoveredFills::default();
unrecovered.open(
&[InstrumentNameExchange::new("BTCUSDT")],
disconnect,
reconnect,
);
recover_margin_fills(
&rest,
&Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
&mut unrecovered,
&tx,
&new_dedup_cache(),
&KnownLiveOrders::shared(ExchangeId::BinanceMargin),
is_isolated,
)
.await;
let mut cids: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
.filter_map(|event| match event.kind {
AccountEventKind::Trade(trade) => Some((
trade.id.0.to_string(),
trade.client_order_id,
trade.order_filled_quantity.is_some(),
)),
_ => None,
})
.collect();
cids.sort();
assert_eq!(
cids,
[
("1".to_string(), Some(ClientOrderId::new("repay-1")), true),
("2".to_string(), None, true),
],
"is_isolated: {is_isolated}"
);
let order_reads = server
.received_requests()
.await
.unwrap()
.iter()
.filter(|request| request.url.path() == "/sapi/v1/margin/order")
.count();
assert_eq!(order_reads, 2, "one read per order");
}
fn venue(trades: Vec<(i64, i64)>) -> MarginMyTradesVenue {
MarginMyTradesVenue {
trades: trades
.into_iter()
.map(|(id, time)| (id, time, id))
.collect(),
end_exclusive: false,
}
}
const HOUR_MS: i64 = 60 * 60 * 1000;
#[tokio::test]
async fn margin_trades_read_a_lone_fill_beyond_the_first_day() {
let now = Utc::now().timestamp_millis();
let (ids, _) = read_margin_trades(
venue(vec![(7, now - 72 * HOUR_MS)]),
since(now - 120 * HOUR_MS),
)
.await;
assert_eq!(ids, vec![7]);
}
#[tokio::test]
async fn margin_trades_read_fills_across_several_windows_once_each() {
let now = Utc::now().timestamp_millis();
let trades = vec![
(1, now - 119 * HOUR_MS),
(2, now - 72 * HOUR_MS),
(3, now - HOUR_MS),
];
let (ids, _) = read_margin_trades(venue(trades), since(now - 144 * HOUR_MS)).await;
assert_eq!(ids, vec![1, 2, 3]);
}
#[tokio::test]
async fn margin_trades_read_a_trade_on_a_window_boundary_once() {
for end_exclusive in [false, true] {
let now = Utc::now().timestamp_millis();
let from = now - 48 * HOUR_MS;
let venue = MarginMyTradesVenue {
end_exclusive,
..venue(vec![(5, from + MARGIN_MY_TRADES_WINDOW_MS)])
};
let (ids, _) = read_margin_trades(venue, since(from)).await;
assert_eq!(ids, vec![5], "end_exclusive: {end_exclusive}");
}
}
#[tokio::test]
async fn margin_trades_page_on_by_id_past_a_full_window() {
let now = Utc::now().timestamp_millis();
let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
let trades = (1..=count)
.map(|id| (id, now - 48 * HOUR_MS + id))
.collect();
let (ids, _) = read_margin_trades(venue(trades), since(now - 50 * HOUR_MS)).await;
assert_eq!(ids, (1..=count).collect::<Vec<_>>());
}
#[tokio::test]
async fn margin_trades_page_on_past_a_full_final_window() {
let now = Utc::now().timestamp_millis();
let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
let trades = (1..=count).map(|id| (id, now - HOUR_MS + id)).collect();
let (ids, requests) = read_margin_trades(venue(trades), since(now - 2 * HOUR_MS)).await;
assert_eq!(ids, (1..=count).collect::<Vec<_>>());
assert_eq!(requests.len(), 2);
}
#[tokio::test]
async fn margin_trades_quiet_lookback_sends_one_request_per_window() {
let now = Utc::now().timestamp_millis();
let (ids, requests) =
read_margin_trades(venue(Vec::new()), since(now - 72 * HOUR_MS)).await;
assert!(ids.is_empty());
assert_eq!(requests.len(), 4);
}
#[tokio::test]
async fn margin_trades_read_a_short_gap_in_one_request() {
let now = Utc::now().timestamp_millis();
let (ids, requests) =
read_margin_trades(venue(vec![(9, now - HOUR_MS)]), since(now - 2 * HOUR_MS)).await;
assert_eq!(ids, vec![9]);
assert_eq!(requests.len(), 1);
}
#[tokio::test]
async fn margin_trades_from_the_future_send_no_request() {
let now = Utc::now().timestamp_millis();
let (ids, requests) = read_margin_trades(venue(Vec::new()), since(now + HOUR_MS)).await;
assert!(ids.is_empty());
assert!(requests.is_empty());
}
#[tokio::test]
async fn margin_trades_by_order_page_by_id_without_time_bounds() {
let now = Utc::now().timestamp_millis();
let count = i64::try_from(BINANCE_MAX_TRADES).unwrap() + 1;
let trades = (1..=2 * count)
.map(|id| {
(
id,
now - 48 * HOUR_MS + id,
if id % 2 == 0 { 42 } else { 7 },
)
})
.collect();
let venue = MarginMyTradesVenue {
trades,
end_exclusive: false,
};
let (ids, requests) = read_margin_trades(venue, MyTradesFrom::Order(42)).await;
assert_eq!(ids, (1..=count).map(|n| 2 * n).collect::<Vec<_>>());
assert_eq!(requests.len(), 2);
assert!(
requests.iter().all(
|query| query.get("orderId").map(String::as_str) == Some("42")
&& !query.contains_key("startTime")
&& !query.contains_key("endTime")
)
);
}
async fn margin_trades_error(response: wiremock::ResponseTemplate) -> UnindexedClientError {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
.respond_with(response)
.mount(&server)
.await;
let rest = Arc::new(MarginTradingRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.build()
.unwrap(),
));
paginate_margin_my_trades(
&rest,
&Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
&InstrumentNameExchange::new("BTCUSDT"),
since(Utc::now().timestamp_millis() - HOUR_MS),
true,
RequestKind::Query,
)
.await
.unwrap_err()
}
#[tokio::test]
async fn margin_trades_report_a_venue_rejection_as_a_non_transient_request_rejection() {
let err = margin_trades_error(wiremock::ResponseTemplate::new(400).set_body_json(
serde_json::json!({"code": -1127, "msg": "More than 24 hours between startTime and endTime."}),
))
.await;
let UnindexedClientError::Api(ApiError::RequestRejected(msg)) = &err else {
panic!("expected RequestRejected, got {err:?}");
};
assert!(msg.contains("-1127"), "code missing from {msg:?}");
assert!(!err.is_transient());
}
#[tokio::test]
async fn margin_trades_report_an_undecodable_response_as_internal() {
let err = margin_trades_error(
wiremock::ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"unexpected": 1})),
)
.await;
assert!(
matches!(err, UnindexedClientError::Internal(_)),
"got {err:?}"
);
assert!(!err.is_transient());
}
#[tokio::test]
async fn sapi_used_ip_weight_near_the_limit_pauses_queries() {
for (used, pauses) in [("10799", false), ("10800", true), ("not a number", false)] {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
.respond_with(
wiremock::ResponseTemplate::new(200)
.insert_header("x-sapi-used-ip-weight-1m", used)
.set_body_json(serde_json::json!([])),
)
.mount(&server)
.await;
let rest = Arc::new(MarginTradingRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.build()
.unwrap(),
));
let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
fetch_margin_all_open_orders(rest, Arc::clone(&tracker), false)
.await
.unwrap();
let query_waits = tokio::time::timeout(
Duration::from_millis(50),
tracker.wait_if_blocked(RequestKind::Query),
)
.await
.is_err();
assert_eq!(query_waits, pauses, "used IP weight {used}");
tokio::time::timeout(
Duration::from_millis(50),
tracker.wait_if_blocked(RequestKind::Order),
)
.await
.expect("orders never wait on the pause")
.expect("no cooldown too long to sleep through");
}
}
#[tokio::test]
async fn margin_fill_recovery_does_not_wait_for_the_weight_pause() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
.respond_with(
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([{
"symbol": "BTCUSDT",
"id": 1,
"orderId": 7,
"price": "100",
"qty": "1",
"commission": "0",
"commissionAsset": "USDT",
"time": Utc::now().timestamp_millis(),
"isBuyer": true,
"isMaker": false,
}])),
)
.expect(2)
.mount(&server)
.await;
let rest = Arc::new(MarginTradingRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.build()
.unwrap(),
));
let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
tracker.throttle(Duration::from_secs(60));
let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
let mut unrecovered = UnrecoveredFills::default();
unrecovered.open(
&[InstrumentNameExchange::new("BTCUSDT")],
Utc::now() - chrono::Duration::minutes(1),
Utc::now(),
);
tokio::time::timeout(
Duration::from_millis(800),
recover_margin_fills(
&rest,
&tracker,
&mut unrecovered,
&tx,
&new_dedup_cache(),
&KnownLiveOrders::shared(ExchangeId::BinanceMargin),
false,
),
)
.await
.expect("recovery does not wait for the pause");
}
#[tokio::test]
async fn margin_trades_read_over_a_span_stop_at_its_end() {
let now = Utc::now().timestamp_millis();
let trades = vec![
(1, now - 50 * HOUR_MS),
(2, now - 49 * HOUR_MS),
(3, now - 10 * HOUR_MS),
];
let (ids, _) = read_margin_trades(
venue(trades),
MyTradesFrom::Span {
start: now - 60 * HOUR_MS,
end: now - 20 * HOUR_MS,
},
)
.await;
assert_eq!(ids, vec![1, 2]);
}
#[tokio::test]
async fn a_margin_gap_whose_every_read_fails_is_reported_when_given_up() {
let disconnect = Utc::now() - chrono::Duration::minutes(10);
let reconnect = Utc::now() - chrono::Duration::minutes(5);
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
.respond_with(
wiremock::ResponseTemplate::new(400)
.set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
)
.mount(&server)
.await;
let rest = Arc::new(MarginTradingRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.build()
.expect("valid config"),
));
let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let dedup = new_dedup_cache();
let btc = InstrumentNameExchange::new("BTCUSDT");
let mut unrecovered = UnrecoveredFills::default();
unrecovered.open(std::slice::from_ref(&btc), disconnect, reconnect);
for _ in 0..=crate::client::order_recovery::MAX_GAP_RETRIES {
recover_margin_fills(
&rest,
&tracker,
&mut unrecovered,
&tx,
&dedup,
&KnownLiveOrders::shared(ExchangeId::BinanceMargin),
false,
)
.await;
unrecovered.make_due();
}
assert!(unrecovered.is_empty(), "the gap is given up");
let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
let [event] = forwarded.as_slice() else {
panic!("only the give-up is sent: {forwarded:?}");
};
let AccountEventKind::FillRecoveryGaveUp(gave_up) = &event.kind else {
panic!("expected FillRecoveryGaveUp, got {event:?}");
};
assert_eq!(event.exchange, ExchangeId::BinanceMargin);
assert_eq!(
gave_up.scope,
crate::FillRecoveryScope::Instruments(vec![btc])
);
assert_eq!(
gave_up.start.timestamp_millis(),
disconnect.timestamp_millis()
);
assert_eq!(
gave_up.attempts,
crate::client::order_recovery::MAX_GAP_RETRIES + 1
);
assert!(
matches!(&gave_up.reason, FillRecoveryFailure::Request(reason) if reason.contains("rejected")),
"the last read's error: {:?}",
gave_up.reason
);
}
#[tokio::test]
async fn a_failed_margin_recovery_keeps_its_closed_gap_for_the_retry() {
let disconnect = Utc::now() - chrono::Duration::minutes(10);
let reconnect = Utc::now() - chrono::Duration::minutes(5);
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
.respond_with(
wiremock::ResponseTemplate::new(400)
.set_body_json(serde_json::json!({"code": -1100, "msg": "rejected"})),
)
.up_to_n_times(1)
.with_priority(1)
.mount(&server)
.await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
.respond_with(venue(vec![
(
1,
(disconnect + chrono::Duration::minutes(1)).timestamp_millis(),
),
(
2,
(Utc::now() - chrono::Duration::minutes(1)).timestamp_millis(),
),
]))
.mount(&server)
.await;
let rest = Arc::new(MarginTradingRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.build()
.unwrap(),
));
let tracker = Arc::new(RateLimitTracker::new(WeightPool::Sapi));
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let dedup = new_dedup_cache();
let mut unrecovered = UnrecoveredFills::default();
unrecovered.open(
&[InstrumentNameExchange::new("BTCUSDT")],
disconnect,
reconnect,
);
recover_margin_fills(
&rest,
&tracker,
&mut unrecovered,
&tx,
&dedup,
&KnownLiveOrders::shared(ExchangeId::BinanceMargin),
false,
)
.await;
assert!(!unrecovered.is_empty(), "the failed gap is kept");
assert!(rx.try_recv().is_err(), "nothing forwarded");
unrecovered.make_due();
recover_margin_fills(
&rest,
&tracker,
&mut unrecovered,
&tx,
&dedup,
&KnownLiveOrders::shared(ExchangeId::BinanceMargin),
false,
)
.await;
assert!(unrecovered.is_empty(), "the retry recovered it");
let forwarded: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
assert_eq!(
forwarded.len(),
1,
"only the fill inside the gap: {forwarded:?}"
);
}
#[tokio::test]
async fn a_timed_out_margin_recovery_charges_only_the_gaps_it_started() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
.respond_with(
wiremock::ResponseTemplate::new(200)
.set_body_json(serde_json::json!([]))
.set_delay(Duration::from_secs(3_600)),
)
.mount(&server)
.await;
let rest = Arc::new(MarginTradingRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.timeout(3_600_000_u64)
.retries(0_u32)
.build()
.unwrap(),
));
let instruments: Vec<_> = (0..9)
.map(|i| InstrumentNameExchange::new(format!("SYM{i}USDT")))
.collect();
let mut unrecovered = UnrecoveredFills::default();
unrecovered.open(
&instruments,
Utc::now() - chrono::Duration::minutes(10),
Utc::now(),
);
let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
tokio::time::pause();
recover_margin_fills(
&rest,
&Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
&mut unrecovered,
&tx,
&new_dedup_cache(),
&KnownLiveOrders::shared(ExchangeId::BinanceMargin),
false,
)
.await;
let now = tokio::time::Instant::now();
assert_eq!(
unrecovered.due(now).len(),
1,
"the gap never started stays due"
);
assert_eq!(
unrecovered
.due(now + Duration::from_secs(crate::client::order_recovery::GAP_RETRY_BASE_SECS))
.len(),
9,
"the eight started wait for their first retry"
);
}
fn margin_order_row(cid: &str, status: &str) -> serde_json::Value {
serde_json::json!({
"symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid, "price": "100",
"origQty": "2", "executedQty": "1", "cummulativeQuoteQty": "105", "status": status,
"timeInForce": "GTC", "type": "LIMIT", "side": "BUY", "isIsolated": false,
"time": 1_700_000_000_000_i64, "updateTime": 1_700_000_060_000_i64,
})
}
fn margin_key(instrument: &str, cid: &str) -> UnindexedOrderKey {
OrderKey::new(
ExchangeId::BinanceMargin,
InstrumentNameExchange::new(instrument),
StrategyId::new("strategy"),
ClientOrderId::new(cid),
)
}
#[test]
fn a_margin_order_row_says_how_the_order_ended() {
use crate::order::state::InactiveOrderState;
use binance_sdk::margin_trading::rest_api::QueryMarginAccountsOrderResponse;
let key = margin_key("BTCUSDT", "a");
let ended = |row: serde_json::Value| {
let row: QueryMarginAccountsOrderResponse = serde_json::from_value(row).unwrap();
convert_ended_order(&row, ExchangeId::BinanceMargin, &key, None)
.map(|order| order.state)
};
let mut filled = margin_order_row("a", "FILLED");
filled["executedQty"] = "2".into();
filled["cummulativeQuoteQty"] = "210".into();
let Some(InactiveOrderState::FullyFilled(filled)) = ended(filled) else {
panic!("FILLED is fully filled");
};
assert_eq!(filled.avg_price, Some(Decimal::from(105)));
assert!(matches!(
ended(margin_order_row("a", "CANCELED")),
Some(InactiveOrderState::Cancelled(cancelled)) if cancelled.filled_quantity == Some(Decimal::ONE)
));
assert_eq!(ended(margin_order_row("a", "NEW")), None);
let mut unknown_quote = margin_order_row("a", "FILLED");
unknown_quote["executedQty"] = "2".into();
unknown_quote["cummulativeQuoteQty"] = "-1".into();
let Some(InactiveOrderState::FullyFilled(filled)) = ended(unknown_quote) else {
panic!("still fully filled");
};
assert_eq!(filled.avg_price, None, "no negative average price");
}
fn margin_rest_at(server: &wiremock::MockServer) -> Arc<RestApi> {
Arc::new(MarginTradingRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.build()
.unwrap(),
))
}
pub(super) fn margin_client_at(
server: &wiremock::MockServer,
is_isolated: bool,
) -> BinanceMargin {
let config = if is_isolated {
BinanceMarginConfig::isolated(
"key".into(),
"secret".into(),
vec![InstrumentNameExchange::new("BTCUSDT")],
)
} else {
BinanceMarginConfig::cross_margin("key".into(), "secret".into())
};
let mode = config.pre_send_rules;
let mut client = <BinanceMargin as ExecutionClient>::new(config);
client.rest = margin_rest_at(server);
client.rules = rules_at(server, mode);
client
}
fn rules_at(server: &wiremock::MockServer, mode: PreSendRules) -> Arc<RulesLoader> {
Arc::new(RulesLoader::new(
mode,
Arc::new(binance_sdk::spot::SpotRestApi::from_config(
ConfigurationRestApi::builder()
.api_key("key")
.api_secret("secret")
.base_path(server.uri())
.build()
.unwrap(),
)),
Arc::new(RateLimitTracker::new(WeightPool::Spot)),
))
}
async fn mount_margin_order(
server: &wiremock::MockServer,
cid: &str,
response: wiremock::ResponseTemplate,
) {
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.and(wiremock::matchers::query_param("origClientOrderId", cid))
.respond_with(response)
.mount(server)
.await;
}
fn margin_venue_error(code: i64) -> wiremock::ResponseTemplate {
wiremock::ResponseTemplate::new(400)
.set_body_json(serde_json::json!({"code": code, "msg": "refused"}))
}
fn margin_row_response(cid: &str, status: &str) -> wiremock::ResponseTemplate {
wiremock::ResponseTemplate::new(200).set_body_json(margin_order_row(cid, status))
}
async fn margin_requests(server: &wiremock::MockServer) -> Vec<(String, Option<String>)> {
server
.received_requests()
.await
.unwrap()
.iter()
.map(|request| {
let isolated = request
.url
.query_pairs()
.find(|(key, _)| key == "isIsolated")
.map(|(_, value)| value.into_owned());
(request.url.path().to_owned(), isolated)
})
.collect()
}
#[tokio::test]
async fn margin_fetch_ended_orders_asks_the_configured_account_and_omits_the_rest() {
use crate::order::state::InactiveOrderState;
for (is_isolated, flag) in [(false, "FALSE"), (true, "TRUE")] {
let server = wiremock::MockServer::start().await;
mount_margin_order(&server, "filled", margin_row_response("filled", "FILLED")).await;
mount_margin_order(&server, "live", margin_row_response("live", "NEW")).await;
mount_margin_order(&server, "unknown", margin_venue_error(-2013)).await;
mount_margin_order(&server, "no-symbol", margin_venue_error(-1121)).await;
let client = margin_client_at(&server, is_isolated);
let keys = ["filled", "live", "unknown"].map(|cid| margin_key("BTCUSDT", cid));
let ended = client
.fetch_ended_orders(
&[keys.to_vec(), vec![margin_key("NOSUCH", "no-symbol")]].concat(),
)
.await
.unwrap();
let [order] = ended.as_slice() else {
panic!("one order ended: {ended:?}");
};
assert_eq!(order.key, keys[0], "the strategy asked survives");
assert!(matches!(order.state, InactiveOrderState::FullyFilled(_)));
let requests = margin_requests(&server).await;
assert_eq!(requests.len(), 4);
for (path, isolated) in requests {
assert_eq!(path, "/sapi/v1/margin/order");
assert_eq!(
isolated.as_deref(),
Some(flag),
"is_isolated = {is_isolated}"
);
}
}
}
#[tokio::test]
async fn margin_fetch_ended_orders_fails_when_any_lookup_fails() {
let server = wiremock::MockServer::start().await;
mount_margin_order(
&server,
"cancelled",
margin_row_response("cancelled", "CANCELED"),
)
.await;
mount_margin_order(&server, "refused", margin_venue_error(-1100)).await;
let client = margin_client_at(&server, false);
let result = client
.fetch_ended_orders(&[
margin_key("BTCUSDT", "cancelled"),
margin_key("BTCUSDT", "refused"),
])
.await;
assert!(
matches!(
result,
Err(UnindexedClientError::Api(ApiError::RequestRejected(_)))
),
"{result:?}"
);
}
fn held_margin_orders() -> SharedKnownLiveOrders {
let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
for (n, cid) in ["live", "gone", "unknown"].into_iter().enumerate() {
known.lock().live(
&margin_key("BTCUSDT", cid),
Decimal::TWO,
OrderKind::Limit,
&Open::new(
VenueOrderId::Assigned(OrderId::new(n.to_string())),
Utc::now(),
Decimal::ZERO,
),
);
}
known
}
#[tokio::test]
async fn an_ended_margin_trailing_stop_is_reported_with_the_kind_it_was_held_with() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
.respond_with(wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
.mount(&server)
.await;
let mut row = margin_order_row("trailing", "CANCELED");
row["type"] = "STOP_LOSS".into();
row["price"] = "0.00000000".into();
row["stopPrice"] = "0.00000000".into();
mount_margin_order(
&server,
"trailing",
wiremock::ResponseTemplate::new(200).set_body_json(row),
)
.await;
let trailing_stop = OrderKind::TrailingStop {
offset: Decimal::from(25),
offset_type: TrailingOffsetType::BasisPoints,
activation_price: None,
};
let hold = |known: &SharedKnownLiveOrders| {
known.lock().live(
&margin_key("BTCUSDT", "trailing"),
Decimal::TWO,
trailing_stop,
&Open::new(
VenueOrderId::Assigned(OrderId::new("7")),
Utc::now(),
Decimal::ZERO,
),
);
};
let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
hold(&known);
let mut unchecked = UncheckedOrders::default();
unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
let (tx, mut rx) = mpsc::unbounded_channel();
recover_margin_ended_orders(
&margin_rest_at(&server),
&Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
&known,
&mut unchecked,
&UnrecoveredFills::default(),
&tx,
false,
)
.await;
let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
let [event] = sent.as_slice() else {
panic!("the trailing stop ended: {sent:?}");
};
let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
panic!("an order snapshot: {event:?}");
};
assert_eq!(snapshot.0.kind, trailing_stop);
assert!(matches!(
snapshot.0.state,
OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(_))
));
assert!(!known.lock().contains(&ClientOrderId::new("trailing")));
let client = margin_client_at(&server, false);
hold(&client.known_live);
let ended = client
.fetch_ended_orders(&[margin_key("BTCUSDT", "trailing")])
.await
.expect("lookup");
let [order] = ended.as_slice() else {
panic!("the trailing stop ended: {ended:?}");
};
assert_eq!(order.kind, trailing_stop);
}
#[tokio::test]
async fn a_margin_reconnect_reports_how_an_unlisted_order_ended_in_the_configured_account() {
for (is_isolated, flag) in [(false, "FALSE"), (true, "TRUE")] {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
.and(wiremock::matchers::query_param("symbol", "BTCUSDT"))
.respond_with(
wiremock::ResponseTemplate::new(200)
.set_body_json(serde_json::json!([margin_order_row("live", "NEW")])),
)
.mount(&server)
.await;
mount_margin_order(&server, "gone", margin_row_response("gone", "CANCELED")).await;
mount_margin_order(&server, "unknown", margin_venue_error(-2013)).await;
let known = held_margin_orders();
let mut unchecked = UncheckedOrders::default();
unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
let (tx, mut rx) = mpsc::unbounded_channel();
recover_margin_ended_orders(
&margin_rest_at(&server),
&Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
&known,
&mut unchecked,
&UnrecoveredFills::default(),
&tx,
is_isolated,
)
.await;
let sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
let [event] = sent.as_slice() else {
panic!("one order ended: {sent:?}");
};
assert_eq!(event.exchange, ExchangeId::BinanceMargin);
let AccountEventKind::OrderSnapshot(snapshot) = &event.kind else {
panic!("an order snapshot: {event:?}");
};
assert_eq!(snapshot.0.key.cid, ClientOrderId::new("gone"));
assert_eq!(snapshot.0.key.strategy, StrategyId::unknown());
assert!(matches!(
snapshot.0.state,
OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(_))
));
let mut requests = margin_requests(&server).await;
requests.sort();
let expected = |path: &str| (path.to_owned(), Some(flag.to_owned()));
assert_eq!(
requests,
[
expected("/sapi/v1/margin/openOrders"),
expected("/sapi/v1/margin/order"),
expected("/sapi/v1/margin/order"),
],
"one listing, and a lookup of each unlisted order only"
);
let known = known.lock();
assert!(known.contains(&ClientOrderId::new("live")));
assert!(!known.contains(&ClientOrderId::new("gone")));
assert!(
!known.contains(&ClientOrderId::new("unknown")),
"asking again cannot help"
);
assert!(unchecked.is_empty());
}
}
fn margin_open_request<'a>(
instrument: &'a InstrumentNameExchange,
cid: &str,
) -> OrderRequestOpen<ExchangeId, &'a InstrumentNameExchange> {
OrderRequestOpen {
key: OrderKey::new(
ExchangeId::BinanceMargin,
instrument,
StrategyId::new("strategy"),
ClientOrderId::new(cid),
),
state: crate::order::request::RequestOpen::new(
Side::Buy,
Some(Decimal::from(100)),
Decimal::TWO,
OrderKind::Limit,
TimeInForce::GoodUntilCancelled { post_only: false },
),
}
}
#[tokio::test]
async fn a_margin_trailing_stop_is_sent_with_trailing_delta() {
let server = wiremock::MockServer::start().await;
let sent_as_trailing_stop = |request: &wiremock::Request| {
let mut body_as_query = request.url.clone();
body_as_query.set_query(Some(&String::from_utf8_lossy(&request.body)));
let params: Vec<(String, String)> = request
.url
.query_pairs()
.chain(body_as_query.query_pairs())
.map(|(k, v)| (k.into_owned(), v.into_owned()))
.collect();
let has = |key: &str, value: &str| params.iter().any(|(k, v)| k == key && v == value);
has("type", "STOP_LOSS")
&& has("trailingDelta", "25")
&& has("sideEffectType", "NO_SIDE_EFFECT")
&& !params.iter().any(|(k, _)| k == "stopPrice" || k == "price")
};
wiremock::Mock::given(wiremock::matchers::method("POST"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.and(sent_as_trailing_stop)
.respond_with(
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "trailing",
"transactTime": 1_700_000_000_000_i64, "price": "0", "origQty": "2",
"executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
"timeInForce": "GTC", "type": "STOP_LOSS", "side": "SELL",
})),
)
.expect(1)
.mount(&server)
.await;
let client = margin_client_at(&server, false);
let btcusdt = InstrumentNameExchange::new("BTCUSDT");
let trailing_request = |cid, offset| {
let mut request = margin_open_request(&btcusdt, cid);
request.state.side = Side::Sell;
request.state.price = None;
request.state.kind = OrderKind::TrailingStop {
offset,
offset_type: TrailingOffsetType::BasisPoints,
activation_price: None,
};
request
};
let placed = client
.open_order(trailing_request("trailing", Decimal::from(25)))
.await;
assert!(
matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
"{placed:?}"
);
let refused = client
.open_order(trailing_request("fractional", Decimal::new(25, 1)))
.await;
assert!(
matches!(
refused.state,
OrderState::Inactive(crate::order::state::InactiveOrderState::OpenFailed(
OrderError::InvalidPrecision(_)
))
),
"{refused:?}"
);
}
#[tokio::test]
async fn a_request_margin_effect_picks_the_side_effect_type_sent() {
let server = wiremock::MockServer::start().await;
let sends = |wire: &'static str, auto_repay_at_cancel: Option<&'static str>| {
move |request: &wiremock::Request| {
let mut body_as_query = request.url.clone();
body_as_query.set_query(Some(&String::from_utf8_lossy(&request.body)));
let params: Vec<(String, String)> = request
.url
.query_pairs()
.chain(body_as_query.query_pairs())
.map(|(k, v)| (k.into_owned(), v.into_owned()))
.collect();
let value = |key: &str| {
params
.iter()
.find(|(k, _)| k == key)
.map(|(_, v)| v.as_str())
};
value("sideEffectType") == Some(wire)
&& value("autoRepayAtCancel") == auto_repay_at_cancel
}
};
for (matcher, cid) in [
(sends("MARGIN_BUY", Some("true")), "borrow"),
(sends("AUTO_REPAY", None), "repay"),
(sends("NO_SIDE_EFFECT", None), "default"),
] {
wiremock::Mock::given(wiremock::matchers::method("POST"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.and(matcher)
.respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
serde_json::json!({
"symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid,
"transactTime": 1_700_000_000_000_i64, "price": "50000", "origQty": "2",
"executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
"timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
}),
))
.expect(1)
.mount(&server)
.await;
}
let client = margin_client_at(&server, false);
let btcusdt = InstrumentNameExchange::new("BTCUSDT");
for (cid, effect) in [
("borrow", Some(MarginEffect::Borrow)),
("repay", Some(MarginEffect::Repay)),
("default", None),
] {
let mut request = margin_open_request(&btcusdt, cid);
request.state.margin_effect = effect;
let placed = client.open_order(request).await;
assert!(
matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
"{cid}: {placed:?}"
);
}
}
use crate::error::{Bound, BoundsViolation, OrderField, PrecisionLimit, PrecisionViolation};
use rust_decimal_macros::dec;
fn btcusdt_exchange_info() -> wiremock::ResponseTemplate {
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"timezone": "UTC", "serverTime": 1_700_000_000_000_i64,
"symbols": [{
"symbol": "BTCUSDT", "status": "TRADING",
"filters": [
{"filterType": "PRICE_FILTER", "minPrice": "0.01", "maxPrice": "0", "tickSize": "0.01"},
{"filterType": "LOT_SIZE", "minQty": "0.001", "maxQty": "100", "stepSize": "0.001"},
{"filterType": "NOTIONAL", "minNotional": "5", "applyMinToMarket": true,
"maxNotional": "9000000", "applyMaxToMarket": false, "avgPriceMins": 5},
],
}],
}))
}
async fn mount_order_accepted(server: &wiremock::MockServer, expected: u64) {
wiremock::Mock::given(wiremock::matchers::method("POST"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.respond_with(
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "c",
"transactTime": 1_700_000_000_000_i64, "price": "50000", "origQty": "2",
"executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
"timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
})),
)
.expect(expected)
.mount(server)
.await;
}
#[tokio::test]
async fn an_order_breaking_the_symbols_rules_is_refused_unsent() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/api/v3/exchangeInfo"))
.and(wiremock::matchers::query_param("symbols", r#"["BTCUSDT"]"#))
.respond_with(btcusdt_exchange_info())
.expect(1)
.mount(&server)
.await;
mount_order_accepted(&server, 1).await;
let client = margin_client_at(&server, false);
let btcusdt = InstrumentNameExchange::new("BTCUSDT");
let order = |cid: &str, price: Decimal, quantity: Decimal| {
let mut request = margin_open_request(&btcusdt, cid);
request.state.price = Some(price);
request.state.quantity = quantity;
request
};
let off_tick = client.open_order(order("a", dec!(100.001), dec!(1))).await;
assert_eq!(
off_tick.state,
OrderState::inactive(OrderError::InvalidPrecision(PrecisionViolation::new(
OrderField::Price,
dec!(100.001),
PrecisionLimit::NotOnGrid { step: dec!(0.01) }
)))
);
let dust = client.open_order(order("b", dec!(100), dec!(0.001))).await;
assert_eq!(
dust.state,
OrderState::inactive(OrderError::OutOfBounds(BoundsViolation::new(
OrderField::Notional,
Bound::Min,
dec!(5),
dec!(0.100)
)))
);
let placed = client.open_order(order("c", dec!(100), dec!(1))).await;
assert!(
matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
"{placed:?}"
);
}
#[tokio::test]
async fn with_the_rules_off_every_order_is_sent_and_no_rules_are_read() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::path("/api/v3/exchangeInfo"))
.respond_with(btcusdt_exchange_info())
.expect(0)
.mount(&server)
.await;
mount_order_accepted(&server, 1).await;
let mut client = margin_client_at(&server, false);
client.rules = rules_at(&server, PreSendRules::Off);
let btcusdt = InstrumentNameExchange::new("BTCUSDT");
let mut request = margin_open_request(&btcusdt, "x");
request.state.price = Some(dec!(100.001));
let placed = client.open_order(request).await;
assert!(
matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
"{placed:?}"
);
}
#[tokio::test]
async fn an_order_whose_rules_cannot_be_loaded_is_sent_unchecked() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::path("/api/v3/exchangeInfo"))
.respond_with(wiremock::ResponseTemplate::new(500))
.mount(&server)
.await;
mount_order_accepted(&server, 1).await;
let client = margin_client_at(&server, false);
let btcusdt = InstrumentNameExchange::new("BTCUSDT");
let mut request = margin_open_request(&btcusdt, "x");
request.state.price = Some(dec!(100.001));
let placed = client.open_order(request).await;
assert!(
matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
"{placed:?}"
);
}
#[tokio::test]
async fn load_rules_returns_the_rules_and_the_first_order_reuses_them() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::path("/api/v3/exchangeInfo"))
.respond_with(btcusdt_exchange_info())
.expect(1)
.mount(&server)
.await;
mount_order_accepted(&server, 0).await;
let client = margin_client_at(&server, false);
let btcusdt = InstrumentNameExchange::new("BTCUSDT");
let loaded = client.load_rules(std::slice::from_ref(&btcusdt)).await;
let rules = loaded.ok().and_then(|mut loaded| loaded.remove(&btcusdt));
assert_eq!(
rules.map(|rules| (rules.price.step, rules.quantity.step, rules.notional.min)),
Some((Some(dec!(0.01)), Some(dec!(0.001)), Some(dec!(5))))
);
let mut request = margin_open_request(&btcusdt, "x");
request.state.price = Some(dec!(100.001));
let refused = client.open_order(request).await;
assert!(
matches!(refused.state, OrderState::Inactive(_)),
"{refused:?}"
);
}
#[tokio::test]
async fn load_rules_of_no_symbols_sends_no_request() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::path("/api/v3/exchangeInfo"))
.respond_with(btcusdt_exchange_info())
.expect(0)
.mount(&server)
.await;
let client = margin_client_at(&server, false);
assert_eq!(client.load_rules(&[]).await, Ok(HashMap::new()));
}
#[tokio::test]
async fn a_cross_refusal_of_a_trailing_stop_is_an_unsupported_order_type() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("POST"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.respond_with(
wiremock::ResponseTemplate::new(400).set_body_json(serde_json::json!({
"code": -3111,
"msg": "Trailing stop order only supports limit type.",
})),
)
.expect(1)
.mount(&server)
.await;
let client = margin_client_at(&server, false);
let btcusdt = InstrumentNameExchange::new("BTCUSDT");
let mut request = margin_open_request(&btcusdt, "trailing");
request.state.side = Side::Sell;
request.state.price = None;
request.state.kind = OrderKind::TrailingStop {
offset: Decimal::from(25),
offset_type: TrailingOffsetType::BasisPoints,
activation_price: None,
};
let refused = client.open_order(request).await;
let OrderState::Inactive(crate::order::state::InactiveOrderState::OpenFailed(
OrderError::UnsupportedOrderType(msg),
)) = &refused.state
else {
panic!("expected UnsupportedOrderType: {refused:?}");
};
assert!(
msg.contains("-3111") && msg.contains("only supports limit type"),
"{msg}"
);
}
#[tokio::test]
async fn placing_listing_and_cancelling_margin_orders_feed_the_known_live_set() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("POST"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.respond_with(
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "placed",
"transactTime": 1_700_000_000_000_i64, "price": "100", "origQty": "2",
"executedQty": "0", "cummulativeQuoteQty": "0", "status": "NEW",
"timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
})),
)
.mount(&server)
.await;
wiremock::Mock::given(wiremock::matchers::method("DELETE"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.respond_with(
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"symbol": "BTCUSDT", "orderId": "7", "origClientOrderId": "placed",
"clientOrderId": "cancel", "price": "100", "origQty": "2", "executedQty": "0",
"cummulativeQuoteQty": "0", "status": "CANCELED", "timeInForce": "GTC",
"type": "LIMIT", "side": "BUY",
})),
)
.mount(&server)
.await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/openOrders"))
.respond_with(
wiremock::ResponseTemplate::new(200)
.set_body_json(serde_json::json!([margin_order_row("listed", "NEW")])),
)
.mount(&server)
.await;
let client = margin_client_at(&server, false);
let btcusdt = InstrumentNameExchange::new("BTCUSDT");
let placed = client
.open_order(margin_open_request(&btcusdt, "placed"))
.await;
assert!(
matches!(placed.state, OrderState::Active(ActiveOrderState::Open(_))),
"{placed:?}"
);
assert!(
client
.known_live
.lock()
.contains(&ClientOrderId::new("placed"))
);
client.fetch_open_orders(&[]).await.unwrap();
assert!(
client
.known_live
.lock()
.contains(&ClientOrderId::new("listed"))
);
let cancelled = client
.cancel_order(OrderRequestCancel {
key: OrderKey::new(
ExchangeId::BinanceMargin,
&btcusdt,
StrategyId::new("strategy"),
ClientOrderId::new("placed"),
),
state: crate::order::request::RequestCancel {
id: Some(VenueOrderId::Assigned(OrderId::new("7"))),
},
})
.await;
assert!(cancelled.state.is_ok(), "{cancelled:?}");
assert!(
!client
.known_live
.lock()
.contains(&ClientOrderId::new("placed"))
);
}
#[tokio::test]
async fn a_cancel_response_without_executed_qty_reports_the_fill_unknown() {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("DELETE"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.respond_with(
wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({
"symbol": "BTCUSDT", "orderId": "7", "origClientOrderId": "placed",
"clientOrderId": "cancel", "price": "100", "origQty": "2",
"status": "CANCELED", "timeInForce": "GTC", "type": "LIMIT", "side": "BUY",
})),
)
.mount(&server)
.await;
let client = margin_client_at(&server, false);
let cancelled = client
.cancel_order(OrderRequestCancel {
key: OrderKey::new(
ExchangeId::BinanceMargin,
&InstrumentNameExchange::new("BTCUSDT"),
StrategyId::new("strategy"),
ClientOrderId::new("placed"),
),
state: crate::order::request::RequestCancel {
id: Some(VenueOrderId::Assigned(OrderId::new("7"))),
},
})
.await;
let Ok(cancelled) = cancelled.state else {
panic!("expected Ok, got {cancelled:?}");
};
assert_eq!(cancelled.filled_quantity, None);
}
#[test]
fn a_reconnect_opens_a_gap_on_each_stream_instrument_and_checks_those_with_orders_held() {
let [btc, eth, xrp] = ["BTCUSDT", "ETHUSDT", "XRPUSDT"].map(InstrumentNameExchange::new);
let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
for (instrument, cid) in [(&btc, "btc"), (&xrp, "xrp")] {
known.lock().live(
&margin_key(instrument.name(), cid),
Decimal::TWO,
OrderKind::Limit,
&Open::new(
VenueOrderId::Assigned(OrderId::new(cid)),
Utc::now(),
Decimal::ZERO,
),
);
}
let mut unrecovered = UnrecoveredFills::default();
let mut unchecked = UncheckedOrders::default();
open_reconnect_recovery(
&mut unrecovered,
&mut unchecked,
&known,
&[btc.clone(), eth.clone()],
Utc::now() - chrono::Duration::minutes(1),
);
assert!(unrecovered.covers(&btc) && unrecovered.covers(ð));
assert!(!unrecovered.covers(&xrp), "not the stream's");
assert!(unchecked.contains(&btc));
assert!(!unchecked.contains(ð), "no order held");
assert!(!unchecked.contains(&xrp), "its fills are not recovered");
}
#[tokio::test]
async fn a_recovered_margin_fill_that_completes_a_held_order_ends_it() {
let disconnect = Utc::now() - chrono::Duration::minutes(10);
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/sapi/v1/margin/myTrades"))
.respond_with(venue(vec![
(
1,
(disconnect + chrono::Duration::minutes(1)).timestamp_millis(),
),
(
2,
(disconnect + chrono::Duration::minutes(2)).timestamp_millis(),
),
]))
.mount(&server)
.await;
let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
for (cid, order_id, quantity) in [("one", "1", "0.01"), ("two", "2", "0.02")] {
known.lock().live(
&margin_key("BTCUSDT", cid),
Decimal::from_str(quantity).unwrap(),
OrderKind::Limit,
&Open::new(
VenueOrderId::Assigned(OrderId::new(order_id)),
Utc::now(),
Decimal::ZERO,
),
);
}
let mut unrecovered = UnrecoveredFills::default();
unrecovered.open(
&[InstrumentNameExchange::new("BTCUSDT")],
disconnect,
Utc::now(),
);
let (tx, mut rx) = mpsc::unbounded_channel();
recover_margin_fills(
&margin_rest_at(&server),
&Arc::new(RateLimitTracker::new(WeightPool::Sapi)),
&mut unrecovered,
&tx,
&new_dedup_cache(),
&known,
false,
)
.await;
assert_eq!(
std::iter::from_fn(|| rx.try_recv().ok()).count(),
2,
"both fills"
);
let known = known.lock();
assert!(!known.contains(&ClientOrderId::new("one")), "completed");
assert!(known.contains(&ClientOrderId::new("two")), "part-filled");
}
#[tokio::test]
async fn a_margin_order_that_expired_on_placement_is_not_reported_or_held_open() {
use rust_decimal_macros::dec;
for (cid, filled, quote) in [("unfilled", "0", "0"), ("partly", "1", "100")] {
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("POST"))
.and(wiremock::matchers::path("/sapi/v1/margin/order"))
.respond_with(wiremock::ResponseTemplate::new(200).set_body_json(
serde_json::json!({
"symbol": "BTCUSDT", "orderId": 7, "clientOrderId": cid,
"transactTime": 1_700_000_000_000_i64, "price": "100", "origQty": "2",
"executedQty": filled, "cummulativeQuoteQty": quote,
"status": "EXPIRED", "timeInForce": "IOC", "type": "LIMIT", "side": "BUY",
}),
))
.mount(&server)
.await;
let client = margin_client_at(&server, false);
let btcusdt = InstrumentNameExchange::new("BTCUSDT");
let placed = client.open_order(margin_open_request(&btcusdt, cid)).await;
let OrderState::Inactive(crate::order::state::InactiveOrderState::Expired(expired)) =
&placed.state
else {
panic!("{cid}: expired, not open: {placed:?}");
};
assert_eq!(expired.filled_quantity, Decimal::from_str(filled).ok());
let late = Open::new(
VenueOrderId::Assigned(OrderId::new("7")),
Utc::now(),
Decimal::ZERO,
);
let mut known = client.known_live.lock();
known.live(&placed.key, dec!(2), OrderKind::Limit, &late);
assert!(
!known.contains(&ClientOrderId::new(cid)),
"{cid}: not held, even after a late live report"
);
}
}
async fn mount_borrow_terms(
server: &wiremock::MockServer,
pair: serde_json::Value,
asset: serde_json::Value,
rate: serde_json::Value,
max_borrow: serde_json::Value,
) {
use wiremock::{
Mock, ResponseTemplate,
matchers::{method, path},
};
for (route, body) in [
("/sapi/v1/margin/allPairs", pair.clone()),
("/sapi/v1/margin/isolated/allPairs", pair),
("/sapi/v1/margin/allAssets", asset),
("/sapi/v1/margin/next-hourly-interest-rate", rate),
("/sapi/v1/margin/maxBorrowable", max_borrow),
("/sapi/v1/margin/delist-schedule", serde_json::json!([])),
] {
Mock::given(method("GET"))
.and(path(route))
.respond_with(ResponseTemplate::new(200).set_body_json(body))
.mount(server)
.await;
}
}
async fn mount_delist_schedule(server: &wiremock::MockServer, schedule: serde_json::Value) {
use wiremock::{
Mock, ResponseTemplate,
matchers::{method, path},
};
Mock::given(method("GET"))
.and(path("/sapi/v1/margin/delist-schedule"))
.respond_with(ResponseTemplate::new(200).set_body_json(schedule))
.with_priority(1)
.mount(server)
.await;
}
fn btcusdt_pair(is_margin_trade: bool, is_sell_allowed: bool) -> serde_json::Value {
serde_json::json!([{
"base": "BTC", "quote": "USDT", "symbol": "BTCUSDT",
"isMarginTrade": is_margin_trade, "isSellAllowed": is_sell_allowed, "isBuyAllowed": true,
}])
}
fn btc_asset(is_borrowable: bool) -> serde_json::Value {
serde_json::json!([{ "assetName": "BTC", "isBorrowable": is_borrowable }])
}
fn btc_rate(rate: &str) -> serde_json::Value {
serde_json::json!([{ "asset": "BTC", "nextHourlyInterestRate": rate }])
}
async fn query_value(server: &wiremock::MockServer, route: &str, key: &str) -> Option<String> {
server
.received_requests()
.await
.unwrap()
.iter()
.find(|request| request.url.path() == route)
.and_then(|request| {
request
.url
.query_pairs()
.find(|(name, _)| name == key)
.map(|(_, value)| value.into_owned())
})
}
#[tokio::test]
async fn margin_shortability_reads_the_pair_the_asset_and_the_annualised_rate() {
use crate::client::ShortabilityClient;
use rust_decimal_macros::dec;
for (is_isolated, pairs_route, flag) in [
(false, "/sapi/v1/margin/allPairs", "FALSE"),
(true, "/sapi/v1/margin/isolated/allPairs", "TRUE"),
] {
let server = wiremock::MockServer::start().await;
mount_borrow_terms(
&server,
btcusdt_pair(true, true),
btc_asset(true),
btc_rate("0.00000571"),
serde_json::json!({}),
)
.await;
let shortability = margin_client_at(&server, is_isolated)
.fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
.await
.unwrap();
assert_eq!(
shortability,
Shortability::new()
.with_shortable(true)
.with_fee_rate(dec!(0.00000571) * dec!(8760)),
"isolated: {is_isolated}"
);
assert_eq!(
query_value(&server, pairs_route, "symbol").await.as_deref(),
Some("BTCUSDT"),
"the pair list is asked for the one symbol"
);
assert_eq!(
query_value(&server, "/sapi/v1/margin/allAssets", "asset")
.await
.as_deref(),
Some("BTC"),
"the base asset is the one borrowed"
);
let rate = "/sapi/v1/margin/next-hourly-interest-rate";
assert_eq!(
query_value(&server, rate, "assets").await.as_deref(),
Some("BTC")
);
assert_eq!(
query_value(&server, rate, "isIsolated").await.as_deref(),
Some(flag)
);
}
}
#[tokio::test]
async fn margin_shortability_is_refused_by_any_flag_and_unknown_without_them() {
use crate::client::ShortabilityClient;
let cases = [
(btcusdt_pair(false, true), btc_asset(true), Some(false)),
(btcusdt_pair(true, false), btc_asset(true), Some(false)),
(btcusdt_pair(true, true), btc_asset(false), Some(false)),
(btcusdt_pair(true, true), serde_json::json!([]), None),
];
for (pair, asset, shortable) in cases {
let server = wiremock::MockServer::start().await;
mount_borrow_terms(
&server,
pair.clone(),
asset.clone(),
serde_json::json!([]),
serde_json::json!({}),
)
.await;
let shortability = margin_client_at(&server, false)
.fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
.await
.unwrap();
assert_eq!(shortability.shortable, shortable, "{pair} {asset}");
assert_eq!(
shortability.fee_rate, None,
"no rate returned is an unknown fee"
);
assert_eq!(shortability.easy_to_borrow, None);
assert_eq!(shortability.available, None);
}
}
#[tokio::test]
async fn margin_shortability_reads_the_delisting_schedule_for_this_margin() {
use crate::client::ShortabilityClient;
let past = Utc::now() - chrono::Duration::hours(1);
let soon = Utc::now() + chrono::Duration::days(3);
let later = soon + chrono::Duration::days(7);
let entry = |time: DateTime<Utc>, assets: &[&str], symbols: &[&str]| {
serde_json::json!({
"delistTime": time.timestamp_millis(),
"crossMarginAssets": assets,
"isolatedMarginSymbols": symbols,
})
};
let millis = |time: DateTime<Utc>| DateTime::from_timestamp_millis(time.timestamp_millis());
let cases = [
(
false,
vec![entry(soon, &["BTC"], &[])],
millis(soon),
Some(true),
),
(
false,
vec![entry(soon, &["USDT"], &[])],
millis(soon),
Some(true),
),
(
false,
vec![
entry(later, &["BTC"], &[]),
entry(soon, &["ETH", "USDT"], &[]),
],
millis(soon),
Some(true),
),
(
false,
vec![entry(soon, &["ETH"], &["BTCUSDT"])],
None,
Some(true),
),
(
false,
vec![entry(past, &["BTC"], &[])],
millis(past),
Some(false),
),
(
true,
vec![entry(soon, &[], &["BTCUSDT"])],
millis(soon),
Some(true),
),
(
true,
vec![entry(soon, &["BTC", "USDT"], &["ETHUSDT"])],
None,
Some(true),
),
(
true,
vec![entry(past, &[], &["BTCUSDT"])],
millis(past),
Some(false),
),
];
for (is_isolated, schedule, delisting_at, shortable) in cases {
let server = wiremock::MockServer::start().await;
mount_borrow_terms(
&server,
btcusdt_pair(true, true),
btc_asset(true),
serde_json::json!([]),
serde_json::json!({}),
)
.await;
let schedule = serde_json::Value::from(schedule);
mount_delist_schedule(&server, schedule.clone()).await;
let shortability = margin_client_at(&server, is_isolated)
.fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
.await
.unwrap();
let case = format!("isolated: {is_isolated}, {schedule}");
assert_eq!(shortability.delisting_at, delisting_at, "{case}");
assert_eq!(shortability.shortable, shortable, "{case}");
}
}
#[tokio::test]
async fn margin_shortability_of_a_scheduled_pair_without_a_time_is_an_error() {
use crate::client::ShortabilityClient;
let server = wiremock::MockServer::start().await;
mount_borrow_terms(
&server,
btcusdt_pair(true, true),
btc_asset(true),
serde_json::json!([]),
serde_json::json!({}),
)
.await;
mount_delist_schedule(
&server,
serde_json::json!([{ "crossMarginAssets": ["BTC"] }]),
)
.await;
let result = margin_client_at(&server, false)
.fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
.await;
assert!(
matches!(result, Err(UnindexedClientError::Internal(ref message))
if message.contains("delistTime")),
"{result:?}"
);
}
#[tokio::test]
async fn margin_shortability_of_an_unlisted_pair_is_an_invalid_instrument() {
use crate::client::ShortabilityClient;
let server = wiremock::MockServer::start().await;
mount_borrow_terms(
&server,
serde_json::json!([]),
btc_asset(true),
btc_rate("0.00000571"),
serde_json::json!({}),
)
.await;
let result = margin_client_at(&server, false)
.fetch_shortability(&InstrumentNameExchange::new("BTCUSDT"))
.await;
assert!(
matches!(
result,
Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(ref instrument, _)))
if instrument.name().as_str() == "BTCUSDT"
),
"{result:?}"
);
}
#[tokio::test]
async fn margin_borrow_capacity_asks_for_the_asset_the_side_borrows() {
use crate::client::BorrowCapacityClient;
use rust_decimal_macros::dec;
for (is_isolated, side, asset, isolated_symbol) in [
(false, Side::Sell, "BTC", None),
(false, Side::Buy, "USDT", None),
(true, Side::Sell, "BTC", Some("BTCUSDT")),
] {
let server = wiremock::MockServer::start().await;
mount_borrow_terms(
&server,
btcusdt_pair(true, true),
btc_asset(true),
btc_rate("0.00000571"),
serde_json::json!({ "amount": "1.69248805", "borrowLimit": "60" }),
)
.await;
let capacity = margin_client_at(&server, is_isolated)
.fetch_borrow_capacity(&InstrumentNameExchange::new("BTCUSDT"), side)
.await
.unwrap();
assert_eq!(
capacity,
BorrowCapacity::new(AssetNameExchange::new(asset))
.with_borrowable_now(dec!(1.69248805))
.with_account_limit(dec!(60)),
"{side:?} isolated: {is_isolated}"
);
let max_borrow = "/sapi/v1/margin/maxBorrowable";
assert_eq!(
query_value(&server, max_borrow, "asset").await.as_deref(),
Some(asset)
);
assert_eq!(
query_value(&server, max_borrow, "isolatedSymbol")
.await
.as_deref(),
isolated_symbol,
"only isolated margin names the pair's account"
);
}
}
#[tokio::test]
async fn margin_borrow_capacity_keeps_an_omitted_amount_unknown_and_fails_a_malformed_one() {
use crate::client::BorrowCapacityClient;
use rust_decimal_macros::dec;
let instrument = InstrumentNameExchange::new("BTCUSDT");
let server = wiremock::MockServer::start().await;
mount_borrow_terms(
&server,
btcusdt_pair(true, true),
btc_asset(true),
btc_rate("0"),
serde_json::json!({ "borrowLimit": "60" }),
)
.await;
let capacity = margin_client_at(&server, false)
.fetch_borrow_capacity(&instrument, Side::Sell)
.await
.unwrap();
assert_eq!(capacity.borrowable_now, None);
assert_eq!(capacity.account_limit, Some(dec!(60)));
let server = wiremock::MockServer::start().await;
mount_borrow_terms(
&server,
btcusdt_pair(true, true),
btc_asset(true),
btc_rate("0"),
serde_json::json!({ "amount": "lots", "borrowLimit": "60" }),
)
.await;
let result = margin_client_at(&server, false)
.fetch_borrow_capacity(&instrument, Side::Sell)
.await;
assert!(
matches!(result, Err(UnindexedClientError::Internal(ref message)) if message.contains("amount")),
"{result:?}"
);
}
#[tokio::test]
async fn margin_borrow_capacity_without_an_isolated_account_names_the_pair() {
use crate::client::BorrowCapacityClient;
use wiremock::{
Mock,
matchers::{method, path},
};
let server = wiremock::MockServer::start().await;
Mock::given(method("GET"))
.and(path("/sapi/v1/margin/maxBorrowable"))
.respond_with(
wiremock::ResponseTemplate::new(400).set_body_json(serde_json::json!({
"code": -11001, "msg": "Isolated margin account does not exist."
})),
)
.mount(&server)
.await;
mount_borrow_terms(
&server,
btcusdt_pair(true, true),
btc_asset(true),
btc_rate("0"),
serde_json::json!({}),
)
.await;
let err = margin_client_at(&server, true)
.fetch_borrow_capacity(&InstrumentNameExchange::new("BTCUSDT"), Side::Sell)
.await
.unwrap_err();
assert_eq!(
err.to_string(),
"API: request rejected: BTCUSDT: -11001 Isolated margin account does not exist."
);
assert!(!err.is_transient());
}
}