use super::shared::{
AbortOnDropStream, BINANCE_MAX_TRADES, BinanceOrderType, BinanceTimeInForce,
CONNECT_TIMEOUT_SECS, ExecutionReport, ExponentialBackoff, FILL_RECOVERY_TIMEOUT_SECS,
HEARTBEAT_TIMEOUT_SECS, MyTradesFrom, ORDER_EXECUTIONS_BUDGET, OpenOrderListing,
PlacementResponse, RateLimitTracker, RequestKind, SIGNAL_RECOVERY_LOOKBACK_MS,
SharedDedupCache, TrailingDeltaError, UnrecoveredFills, 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, drop_after,
frame_excerpt, gap_failed, gap_time, is_duplicate, is_unknown_order, log_unrecognised_frame,
new_dedup_cache, parse_user_data_frame, placed_order_state, recovered_order_totals,
response_decode_error, rest_call_with_retry, trailing_delta_basis_points,
};
use crate::{
AccountEventKind, AccountSnapshot, FillRecoveryFailure, InstrumentAccountSnapshot,
InstrumentBalanceUpdate, IsolatedInstrumentState, IsolatedMarginRisk, UnindexedAccountEvent,
UnindexedAccountSnapshot,
balance::{AssetBalance, AssetBalanceUpdate, Balance, BalanceUpdate},
client::{
ExecutionClient, OrderStatusClient,
order_recovery::{
KnownLiveOrders, OpenListing, OrderLookup, SharedKnownLiveOrders, UncheckedOrders,
fetch_ended_by_key, recover_ended_orders,
},
},
emit_stream_terminated,
error::{
ApiError, ConnectivityError, OrderError, StreamTerminationReason, UnindexedClientError,
},
order::{
Order, OrderKey, OrderKind, TimeInForce, UnindexedInactiveOrder, UnindexedOrderKey,
id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
request::{OrderRequestCancel, OrderRequestOpen, UnindexedOrderResponseCancel},
state::{Cancelled, Open, OrderState, UnindexedOrderState},
},
position::PositionReport,
trade::{AssetFees, Trade, TradeId, TradesRead},
};
use binance_sdk::{
common::{
config::{ConfigurationRestApi, ConfigurationWebsocketApi},
constants::{MARGIN_TRADING_REST_API_PROD_URL, SPOT_WS_API_PROD_URL},
models::WebsocketEvent,
websocket::{
SendWebsocketMessageResult, Subscription, WebsocketApi as WsApiBase,
WebsocketMessageSendOptions,
},
},
margin_trading::{
MarginTradingRestApi,
rest_api::{
MarginAccountCancelOrderIsIsolatedEnum, MarginAccountCancelOrderParams,
MarginAccountNewOrderIsIsolatedEnum, MarginAccountNewOrderNewOrderRespTypeEnum,
MarginAccountNewOrderParams, MarginAccountNewOrderSideEffectTypeEnum,
MarginAccountNewOrderSideEnum, MarginAccountNewOrderTimeInForceEnum,
MarginAccountNewOrderTypeEnum, QueryCrossMarginAccountDetailsParams,
QueryCrossMarginAccountDetailsResponseUserAssetsInner,
QueryIsolatedMarginAccountInfoParams,
QueryIsolatedMarginAccountInfoResponseAssetsInner,
QueryMarginAccountsOpenOrdersIsIsolatedEnum, QueryMarginAccountsOpenOrdersParams,
QueryMarginAccountsOrderIsIsolatedEnum, QueryMarginAccountsOrderParams,
QueryMarginAccountsTradeListIsIsolatedEnum, QueryMarginAccountsTradeListParams,
QueryMarginAccountsTradeListResponseInner, 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, Serialize};
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};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MarginSideEffect {
#[default]
AutoBorrowRepay,
NoBorrow,
}
impl MarginSideEffect {
pub fn as_binance_str(self) -> &'static str {
match self {
MarginSideEffect::AutoBorrowRepay => "AUTO_BORROW_REPAY",
MarginSideEffect::NoBorrow => "NO_SIDE_EFFECT",
}
}
fn as_sdk_enum(self) -> MarginAccountNewOrderSideEffectTypeEnum {
match self {
MarginSideEffect::AutoBorrowRepay => {
MarginAccountNewOrderSideEffectTypeEnum::AutoBorrowRepay
}
MarginSideEffect::NoBorrow => MarginAccountNewOrderSideEffectTypeEnum::NoSideEffect,
}
}
}
#[derive(Clone, Deserialize)]
pub struct BinanceMarginConfig {
api_key: String,
secret_key: String,
pub testnet: bool,
#[serde(default)]
pub is_isolated: bool,
#[serde(default)]
pub isolated_symbols: Vec<InstrumentNameExchange>,
#[serde(default)]
pub side_effect: MarginSideEffect,
}
impl std::fmt::Debug for BinanceMarginConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BinanceMarginConfig")
.field("api_key", &"***")
.field("secret_key", &"***")
.field("testnet", &self.testnet)
.field("is_isolated", &self.is_isolated)
.field("isolated_symbols", &self.isolated_symbols)
.field("side_effect", &self.side_effect)
.finish()
}
}
impl BinanceMarginConfig {
pub fn new(
api_key: String,
secret_key: String,
testnet: bool,
is_isolated: bool,
isolated_symbols: Vec<InstrumentNameExchange>,
side_effect: MarginSideEffect,
) -> Self {
Self {
api_key,
secret_key,
testnet,
is_isolated,
isolated_symbols,
side_effect,
}
}
pub fn cross_margin(api_key: String, secret_key: String) -> Self {
Self::new(
api_key,
secret_key,
false,
false,
Vec::new(),
MarginSideEffect::default(),
)
}
pub fn isolated(
api_key: String,
secret_key: String,
symbols: Vec<InstrumentNameExchange>,
) -> Self {
Self::new(
api_key,
secret_key,
false,
true,
symbols,
MarginSideEffect::default(),
)
}
pub fn api_key(&self) -> &str {
&self.api_key
}
}
#[derive(Clone)]
pub struct BinanceMargin {
config: Arc<BinanceMarginConfig>,
rest: Arc<RestApi>,
rest_config: Arc<ConfigurationRestApi>,
ws_config: ConfigurationWebsocketApi,
rate_limiter: Arc<RateLimitTracker>,
known_live: SharedKnownLiveOrders,
}
impl std::fmt::Debug for BinanceMargin {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BinanceMargin")
.field("testnet", &self.config.testnet)
.field("is_isolated", &self.config.is_isolated)
.field("side_effect", &self.config.side_effect)
.finish_non_exhaustive()
}
}
impl BinanceMargin {
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 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),
}
}
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(),
self.config.side_effect,
self.config.is_isolated,
) {
Ok(params) => params,
Err(BuildOrderError::Unsupported) => {
return inactive(OrderState::inactive(OrderError::UnsupportedOrderType(
format!(
"BinanceMargin does not support OrderKind::{kind:?} with {time_in_force:?}"
),
)));
}
Err(BuildOrderError::TrailingDelta(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),
)));
}
};
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?;
let live = isolated_connect_and_subscribe(
&ws_config,
&tx,
&dedup,
&known_live,
&base_quote,
&tokens,
)
.await
.map_err(|e| {
UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
})?;
tokio::spawn(isolated_connection_manager(
tx,
dedup,
ws_config,
rest_config,
rest,
rate_limiter,
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?;
let initial_ws = connect_margin_ws(&ws_config).await.map_err(|e| {
UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
})?;
tokio::spawn(margin_connection_manager(
tx,
dedup,
ws_config,
rest_config,
rest,
rate_limiter,
known_live,
instruments,
Some((initial_ws, initial_token)),
))
};
let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
let guarded_stream = AbortOnDropStream::new(rx_stream, cm_handle);
Ok(futures::StreamExt::boxed(guarded_stream))
}
}
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 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<Arc<WsApiBase>> {
let ws = WsApiBase::new(ws_config.clone(), vec![]);
tokio::time::timeout(
Duration::from_secs(CONNECT_TIMEOUT_SECS),
ws.clone().connect(),
)
.await
.map_err(|_| {
anyhow::anyhow!("BinanceMargin WS connect timed out after {CONNECT_TIMEOUT_SECS}s")
})??;
Ok(ws)
}
async fn subscribe_listen_token(ws: &Arc<WsApiBase>, token: &str) -> anyhow::Result<()> {
let mut payload = BTreeMap::new();
payload.insert(
"listenToken".to_string(),
serde_json::Value::String(token.to_string()),
);
ws.send_message::<serde_json::Value>(
"userDataStream.subscribe.listenToken",
payload,
WebsocketMessageSendOptions::new(),
)
.await
.map_err(|e| anyhow::anyhow!("userDataStream.subscribe.listenToken failed: {e}"))?;
Ok(())
}
async fn subscribe_listen_token_capture(
ws: &Arc<WsApiBase>,
token: &str,
) -> anyhow::Result<Option<i64>> {
#[derive(Deserialize)]
struct SubscribeAck {
#[serde(rename = "subscriptionId", default)]
subscription_id: Option<i64>,
}
let mut payload = BTreeMap::new();
payload.insert(
"listenToken".to_string(),
serde_json::Value::String(token.to_string()),
);
let result = ws
.send_message::<SubscribeAck>(
"userDataStream.subscribe.listenToken",
payload,
WebsocketMessageSendOptions::new(),
)
.await
.map_err(|e| anyhow::anyhow!("userDataStream.subscribe.listenToken failed: {e}"))?;
let ack = match result {
SendWebsocketMessageResult::Single(resp) => resp.data()?,
SendWebsocketMessageResult::Multiple(responses) => responses
.into_iter()
.next()
.ok_or_else(|| anyhow::anyhow!("empty subscribe ack"))?
.data()?,
};
Ok(ack.subscription_id)
}
fn cross_account_position_handler(
position: OutboundAccountPosition,
_subscription_id: Option<i64>,
buf: &mut Vec<UnindexedAccountEvent>,
) {
convert_margin_account_position(position, buf);
}
#[cfg(test)]
fn convert_margin_user_data_events(frame: &str, buf: &mut Vec<UnindexedAccountEvent>) -> bool {
convert_margin_user_data_events_with(frame, buf, &mut cross_account_position_handler)
}
fn convert_margin_user_data_events_with(
frame: &str,
buf: &mut Vec<UnindexedAccountEvent>,
handle_position: &mut impl FnMut(
OutboundAccountPosition,
Option<i64>,
&mut Vec<UnindexedAccountEvent>,
),
) -> bool {
let (subscription_id, event_type, event_raw) = match parse_user_data_frame(frame) {
UserDataFrame::Response => return false,
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 false;
}
};
match event_type {
"executionReport" => {
match serde_json::from_str::<ExecutionReport>(event_raw) {
Ok(report) => 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
}
"userLiabilityChange" => {
match serde_json::from_str::<UserLiabilityChange>(event_raw) {
Ok(c) => {
if tracing::enabled!(tracing::Level::INFO) {
info!(
asset = c.a.as_deref().unwrap_or("?"),
kind = c.t.as_deref().unwrap_or("?"),
principal = c.p.as_deref().unwrap_or("?"),
interest = c.i.as_deref().unwrap_or("?"),
"BinanceMargin userLiabilityChange (observable; not applied to balance state)"
);
}
}
Err(e) => warn!(
error = %e,
frame = frame_excerpt(event_raw),
"BinanceMargin: undeserializable userLiabilityChange, dropping"
),
}
false
}
"marginLevelStatusChange" => {
if tracing::enabled!(tracing::Level::WARN) {
match serde_json::from_str::<MarginLevelStatusChange>(event_raw) {
Ok(c) => warn!(
margin_level = c.l.as_deref().unwrap_or("?"),
status = c.s.as_deref().unwrap_or("?"),
"BinanceMargin marginLevelStatusChange (liquidation risk; observable, no policy)"
),
Err(e) => warn!(
error = %e,
frame = frame_excerpt(event_raw),
"BinanceMargin: undeserializable marginLevelStatusChange, dropping"
),
}
}
false
}
"eventStreamTerminated" => {
warn!("BinanceMargin user data stream terminated by exchange, signalling reconnect");
true
}
other => {
trace!(
event_type = other,
"BinanceMargin ignoring unhandled user data event"
);
false
}
}
}
fn convert_margin_account_position(
position: OutboundAccountPosition,
buf: &mut Vec<UnindexedAccountEvent>,
) {
let time_exchange = position
.u
.and_then(|ms| Utc.timestamp_millis_opt(ms).single())
.unwrap_or_else(Utc::now);
for b in position.b_uppercase.unwrap_or_default() {
let asset = match b.a {
Some(a) => AssetNameExchange::new(a),
None => {
warn!("BinanceMargin account position entry missing asset name");
continue;
}
};
let free = match b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%asset, "BinanceMargin account position missing/unparseable 'free' field");
continue;
}
};
let locked = match b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()) {
Some(v) => v,
None => {
warn!(%asset, "BinanceMargin account position missing/unparseable 'locked' field");
continue;
}
};
let update =
AssetBalanceUpdate::new(asset, BalanceUpdate::new(free, locked), time_exchange);
buf.push(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::BalanceStreamUpdate(
rustrade_integration::collection::snapshot::Snapshot::new(update),
),
));
}
}
fn route_isolated_account_position(
position: OutboundAccountPosition,
subscription_id: Option<i64>,
sub_map: &Mutex<HashMap<i64, InstrumentNameExchange>>,
base_quote: &HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>,
buf: &mut Vec<UnindexedAccountEvent>,
) {
let Some(sub_id) = subscription_id else {
warn!(
"BinanceMargin isolated: outboundAccountPosition without subscriptionId — dropping \
(per-instrument balance routing requires it)"
);
return;
};
let instrument = {
let map = sub_map.lock().unwrap_or_else(|p| p.into_inner());
match map.get(&sub_id) {
Some(inst) => inst.clone(),
None => {
warn!(
subscription_id = sub_id,
"BinanceMargin isolated: outboundAccountPosition for unmapped subscriptionId — dropping"
);
return;
}
}
};
let Some((base_asset, quote_asset)) = base_quote.get(&instrument) else {
warn!(
instrument = %instrument.name(),
"BinanceMargin isolated: no base/quote known for instrument — dropping balance frame"
);
return;
};
let time_exchange = position
.u
.and_then(|ms| Utc.timestamp_millis_opt(ms).single())
.unwrap_or_else(Utc::now);
let mut base_update: Option<AssetBalanceUpdate<AssetNameExchange>> = None;
let mut quote_update: Option<AssetBalanceUpdate<AssetNameExchange>> = None;
for b in position.b_uppercase.unwrap_or_default() {
let Some(asset) = b.a.as_deref() else {
warn!(instrument = %instrument.name(), "BinanceMargin isolated account position entry missing asset name");
continue;
};
let side = if asset == base_asset.name().as_str() {
&mut base_update
} else if asset == quote_asset.name().as_str() {
&mut quote_update
} else {
continue;
};
let (Some(free), Some(locked)) = (
b.f.as_deref().and_then(|s| Decimal::from_str(s).ok()),
b.l.as_deref().and_then(|s| Decimal::from_str(s).ok()),
) else {
warn!(%asset, instrument = %instrument.name(), "BinanceMargin isolated account position missing/unparseable free/locked");
continue;
};
*side = Some(AssetBalanceUpdate::new(
AssetNameExchange::new(asset),
BalanceUpdate::new(free, locked),
time_exchange,
));
}
let (Some(base), Some(quote)) = (base_update, quote_update) else {
warn!(
instrument = %instrument.name(),
"BinanceMargin isolated: outboundAccountPosition missing base or quote side — dropping (no partial InstrumentBalanceUpdate)"
);
return;
};
buf.push(UnindexedAccountEvent::new(
ExchangeId::BinanceMargin,
AccountEventKind::InstrumentBalanceUpdate(InstrumentBalanceUpdate::new(
instrument, base, quote,
)),
));
}
fn register_user_data_listener(
ws: &Arc<WsApiBase>,
tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: SharedDedupCache,
known: SharedKnownLiveOrders,
heartbeat_flag: Arc<AtomicBool>,
signal_tx: oneshot::Sender<()>,
mut handle_position: impl FnMut(
OutboundAccountPosition,
Option<i64>,
&mut Vec<UnindexedAccountEvent>,
) + Send
+ 'static,
) -> Subscription {
let mut signal_tx_opt = Some(signal_tx);
let mut event_tx = Some(tx);
let mut event_buf = Vec::with_capacity(32);
ws.common.events.subscribe(move |event| {
let Some(ref sender) = event_tx else { return };
match event {
WebsocketEvent::Message(json_str) => {
heartbeat_flag.store(true, Ordering::Release);
let terminated = convert_margin_user_data_events_with(
&json_str,
&mut event_buf,
&mut handle_position,
);
for ev in event_buf.drain(..) {
if let Some(key) = dedup_key_from_event(&ev)
&& is_duplicate(&dedup, key)
{
trace!("BinanceMargin dedup: skipping duplicate event");
continue;
}
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 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());
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<(Arc<WsApiBase>, UserListenToken)>,
) {
enum DisconnectReason {
Signal,
HeartbeatTimeout,
TokenRefresh,
ConsumerDropped,
}
let mut backoff = ExponentialBackoff::new();
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 => match acquire_user_listen_token(&rest_config, &rate_limiter, None).await {
Ok(t) => t,
Err(e) => {
error!(%e, "BinanceMargin userListenToken acquisition failed");
if !backoff.wait().await {
error!("BinanceMargin max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
},
};
let ws = match current_ws.take() {
Some(ws) => ws,
None => match connect_margin_ws(&ws_config).await {
Ok(ws) => ws,
Err(e) => {
error!(%e, "BinanceMargin WS connect failed");
if !backoff.wait().await {
error!("BinanceMargin max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
},
};
let (signal_tx, signal_rx) = oneshot::channel::<()>();
let heartbeat_flag = Arc::new(AtomicBool::new(true));
let subscription = register_user_data_listener(
&ws,
tx.clone(),
dedup.clone(),
known.clone(),
heartbeat_flag.clone(),
signal_tx,
cross_account_position_handler,
);
if let Err(e) = subscribe_listen_token(&ws, &token.token).await {
warn!(%e, "BinanceMargin user-data subscribe failed, cleaning up and retrying");
subscription.unsubscribe();
if let Err(de) = ws.disconnect().await {
warn!(%de, "BinanceMargin failed to disconnect after subscribe failure");
}
if !backoff.wait().await {
error!("BinanceMargin max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
info!("BinanceMargin account_stream connected and subscribed");
backoff.reset();
let token_deadline =
tokio::time::Instant::now() + token_renew_after(token.expiration_time_ms);
if let Some(dt) = disconnect_time.take() {
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);
loop {
tokio::select! {
biased;
_ = tx.closed() => {
debug!("BinanceMargin account_stream consumer dropped, terminating");
break DisconnectReason::ConsumerDropped;
}
_ = &mut signal_rx => {
warn!("BinanceMargin WS disconnected, will attempt reconnect");
break DisconnectReason::Signal;
}
() = tokio::time::sleep_until(token_deadline) => {
info!("BinanceMargin userListenToken nearing expiry, renewing");
break DisconnectReason::TokenRefresh;
}
() = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
if heartbeat_flag.swap(false, Ordering::AcqRel) {
continue;
}
warn!("BinanceMargin heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
break DisconnectReason::HeartbeatTimeout;
}
() = &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) && !backoff.wait().await {
error!("BinanceMargin max reconnect attempts exhausted, stream terminating");
let last_error = match reason {
DisconnectReason::HeartbeatTimeout => {
format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
}
_ => "WebSocket disconnected (server close/error)".to_string(),
};
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error,
},
);
break;
}
}
}
fn build_base_quote_map(
assets: &[QueryIsolatedMarginAccountInfoResponseAssetsInner],
) -> HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)> {
let mut map = HashMap::with_capacity(assets.len());
for entry in assets {
let (Some(symbol), Some(base), Some(quote)) = (
entry.symbol.as_deref(),
entry.base_asset.as_deref().and_then(|b| b.asset.as_deref()),
entry
.quote_asset
.as_deref()
.and_then(|q| q.asset.as_deref()),
) else {
warn!(
"BinanceMargin isolated: account-info entry missing symbol/base/quote asset — \
skipping base/quote map entry"
);
continue;
};
map.insert(
InstrumentNameExchange::new(symbol),
(AssetNameExchange::new(base), AssetNameExchange::new(quote)),
);
}
map
}
async fn acquire_all_isolated_tokens(
rest_config: &Arc<ConfigurationRestApi>,
rate_limiter: &Arc<RateLimitTracker>,
symbols: &[InstrumentNameExchange],
) -> Result<Vec<(InstrumentNameExchange, UserListenToken)>, UnindexedClientError> {
use futures::{StreamExt as _, TryStreamExt as _};
futures::stream::iter(symbols.iter().cloned().map(|sym| {
let rest_config = rest_config.clone();
let rate_limiter = rate_limiter.clone();
async move {
let token = acquire_user_listen_token(&rest_config, &rate_limiter, Some(&sym)).await?;
Ok::<_, UnindexedClientError>((sym, token))
}
}))
.buffer_unordered(8)
.try_collect()
.await
}
fn earliest_token_expiry_ms(tokens: &[(InstrumentNameExchange, UserListenToken)]) -> i64 {
tokens
.iter()
.map(|(_, t)| t.expiration_time_ms)
.min()
.unwrap_or(i64::MAX)
}
struct IsolatedLiveConn {
ws: Arc<WsApiBase>,
subscription: Subscription,
signal_rx: oneshot::Receiver<()>,
heartbeat_flag: Arc<AtomicBool>,
earliest_expiry_ms: i64,
}
async fn isolated_connect_and_subscribe(
ws_config: &ConfigurationWebsocketApi,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: &SharedDedupCache,
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(true));
let handle_position = {
let sub_map = sub_map.clone();
let base_quote = base_quote.clone();
move |position, subscription_id, buf: &mut Vec<UnindexedAccountEvent>| {
route_isolated_account_position(position, subscription_id, &sub_map, &base_quote, buf);
}
};
let subscription = register_user_data_listener(
&ws,
tx.clone(),
dedup.clone(),
known.clone(),
heartbeat_flag.clone(),
signal_tx,
handle_position,
);
let earliest_expiry_ms = earliest_token_expiry_ms(tokens);
for (sym, token) in tokens {
match subscribe_listen_token_capture(&ws, &token.token).await {
Ok(Some(id)) => {
sub_map
.lock()
.unwrap_or_else(|p| p.into_inner())
.insert(id, sym.clone());
}
Ok(None) => warn!(
symbol = %sym.name(),
"BinanceMargin isolated: subscribe ack carried no subscriptionId — live per-pair \
balances unavailable for this symbol (fills/orders unaffected; balances via snapshot)"
),
Err(e) => {
warn!(%e, symbol = %sym.name(), "BinanceMargin isolated subscribe failed");
subscription.unsubscribe();
if let Err(de) = ws.disconnect().await {
warn!(%de, "BinanceMargin isolated: disconnect after subscribe failure failed");
}
return Err(e);
}
}
}
Ok(IsolatedLiveConn {
ws,
subscription,
signal_rx,
heartbeat_flag,
earliest_expiry_ms,
})
}
#[allow(
clippy::cognitive_complexity,
reason = "inherent reconnect-loop complexity (tokens + connect + subscribe + monitor + cleanup \
+ backoff); mirrors the cross manager, not worth splitting further"
)]
async fn isolated_connection_manager(
tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: SharedDedupCache,
ws_config: ConfigurationWebsocketApi,
rest_config: Arc<ConfigurationRestApi>,
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
known: SharedKnownLiveOrders,
symbols: Vec<InstrumentNameExchange>,
base_quote: Arc<HashMap<InstrumentNameExchange, (AssetNameExchange, AssetNameExchange)>>,
initial: Option<IsolatedLiveConn>,
) {
enum DisconnectReason {
Signal,
HeartbeatTimeout,
TokenRefresh,
ConsumerDropped,
}
let mut backoff = ExponentialBackoff::new();
let mut disconnect_time: Option<DateTime<Utc>> = None;
let mut 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 => {
let tokens = match acquire_all_isolated_tokens(
&rest_config,
&rate_limiter,
&symbols,
)
.await
{
Ok(t) => t,
Err(e) => {
error!(%e, "BinanceMargin isolated userListenToken acquisition failed");
if !backoff.wait().await {
error!("BinanceMargin isolated max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
};
match isolated_connect_and_subscribe(
&ws_config,
&tx,
&dedup,
&known,
&base_quote,
&tokens,
)
.await
{
Ok(live) => live,
Err(e) => {
error!(%e, "BinanceMargin isolated connect/subscribe failed");
if !backoff.wait().await {
error!("BinanceMargin isolated max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
}
}
};
info!(
symbols = symbols.len(),
"BinanceMargin isolated account_stream connected and subscribed"
);
backoff.reset();
let token_deadline = tokio::time::Instant::now() + token_renew_after(earliest_expiry_ms);
if let Some(dt) = disconnect_time.take() {
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);
loop {
tokio::select! {
biased;
_ = tx.closed() => {
debug!("BinanceMargin isolated consumer dropped, terminating");
break DisconnectReason::ConsumerDropped;
}
_ = &mut signal_rx => {
warn!("BinanceMargin isolated WS disconnected, will attempt reconnect");
break DisconnectReason::Signal;
}
() = tokio::time::sleep_until(token_deadline) => {
info!("BinanceMargin isolated userListenToken nearing expiry, renewing all");
break DisconnectReason::TokenRefresh;
}
() = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS)) => {
if heartbeat_flag.swap(false, Ordering::AcqRel) {
continue;
}
warn!("BinanceMargin isolated heartbeat timeout ({}s), will attempt reconnect", HEARTBEAT_TIMEOUT_SECS);
break DisconnectReason::HeartbeatTimeout;
}
() = &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) && !backoff.wait().await {
error!("BinanceMargin isolated max reconnect attempts exhausted, stream terminating");
let last_error = match reason {
DisconnectReason::HeartbeatTimeout => {
format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
}
_ => "WebSocket disconnected (server close/error)".to_string(),
};
emit_stream_terminated(
&tx,
ExchangeId::BinanceMargin,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: backoff.attempts(),
last_error,
},
);
break;
}
}
}
async fn fetch_margin_open_orders_for_instrument(
rest: Arc<RestApi>,
rate_limiter: Arc<RateLimitTracker>,
instrument: InstrumentNameExchange,
is_isolated: bool,
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 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),
TrailingDelta(TrailingDeltaError),
}
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,
);
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,
side_effect: MarginSideEffect,
is_isolated: bool,
) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
let binance_side = match side {
Side::Buy => MarginAccountNewOrderSideEnum::Buy,
Side::Sell => MarginAccountNewOrderSideEnum::Sell,
};
let (binance_type, binance_tif) =
convert_order_kind_tif_margin(kind, time_in_force).ok_or(BuildOrderError::Unsupported)?;
let mut builder = MarginAccountNewOrderParams::builder(symbol, binance_side, binance_type)
.quantity(quantity)
.is_isolated(MarginAccountNewOrderIsIsolatedEnum::from_flag(is_isolated))
.side_effect_type(side_effect.as_sdk_enum())
.new_client_order_id(new_client_order_id)
.new_order_resp_type(MarginAccountNewOrderNewOrderRespTypeEnum::Full);
if side_effect == MarginSideEffect::AutoBorrowRepay {
builder = builder.auto_repay_at_cancel(true);
}
if let Some(tif) = binance_tif {
builder = builder.time_in_force(tif);
}
match kind {
OrderKind::Limit => {
builder = builder.price(price);
}
OrderKind::Stop { trigger_price } | OrderKind::TakeProfit { trigger_price } => {
builder = builder.stop_price(trigger_price);
}
OrderKind::StopLimit { trigger_price } | OrderKind::TakeProfitLimit { trigger_price } => {
builder = builder.price(price).stop_price(trigger_price);
}
OrderKind::TrailingStop {
offset,
offset_type,
} => {
let basis_points = trailing_delta_basis_points(offset, offset_type)
.map_err(BuildOrderError::TrailingDelta)?;
builder = builder.trailing_delta(i64::from(basis_points));
}
_ => {}
}
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;
use crate::order::state::ActiveOrderState;
use binance_sdk::margin_trading::rest_api::QueryMarginAccountsOpenOrdersResponseInner;
#[test]
fn margin_side_effect_default_is_auto_borrow_repay() {
assert_eq!(
MarginSideEffect::default(),
MarginSideEffect::AutoBorrowRepay
);
}
#[test]
fn margin_side_effect_wire_strings() {
assert_eq!(
MarginSideEffect::AutoBorrowRepay.as_binance_str(),
"AUTO_BORROW_REPAY"
);
assert_eq!(
MarginSideEffect::NoBorrow.as_binance_str(),
"NO_SIDE_EFFECT"
);
}
#[test]
fn margin_side_effect_serde_round_trip() {
assert_eq!(
serde_json::to_string(&MarginSideEffect::AutoBorrowRepay).unwrap(),
r#""auto_borrow_repay""#
);
assert_eq!(
serde_json::to_string(&MarginSideEffect::NoBorrow).unwrap(),
r#""no_borrow""#
);
assert_eq!(
serde_json::from_str::<MarginSideEffect>(r#""auto_borrow_repay""#).unwrap(),
MarginSideEffect::AutoBorrowRepay
);
assert_eq!(
serde_json::from_str::<MarginSideEffect>(r#""no_borrow""#).unwrap(),
MarginSideEffect::NoBorrow
);
}
#[test]
fn config_debug_redacts_secrets() {
let config = BinanceMarginConfig::new(
"my_api_key".to_string(),
"my_secret_key".to_string(),
false,
false,
Vec::new(),
MarginSideEffect::default(),
);
let debug = format!("{config:?}");
assert!(!debug.contains("my_api_key"));
assert!(!debug.contains("my_secret_key"));
assert!(debug.contains("***"));
}
#[test]
fn cross_margin_uses_common_case_defaults() {
let config = BinanceMarginConfig::cross_margin("k".to_string(), "s".to_string());
assert!(!config.testnet);
assert!(!config.is_isolated);
assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
assert_eq!(config.api_key(), "k");
}
#[test]
fn config_deserializes_with_defaults() {
let config: BinanceMarginConfig =
serde_json::from_str(r#"{"api_key":"k","secret_key":"s","testnet":false}"#)
.expect("deserialize");
assert!(!config.is_isolated);
assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
}
#[test]
fn config_deserializes_explicit_side_effect() {
let config: BinanceMarginConfig = serde_json::from_str(
r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true,"side_effect":"no_borrow"}"#,
)
.expect("deserialize");
assert!(config.is_isolated);
assert_eq!(config.side_effect, MarginSideEffect::NoBorrow);
}
#[test]
fn isolated_ctor_sets_symbols_and_flag() {
let symbols = vec![
InstrumentNameExchange::new("BTCUSDT"),
InstrumentNameExchange::new("ETHUSDT"),
];
let config =
BinanceMarginConfig::isolated("k".to_string(), "s".to_string(), symbols.clone());
assert!(config.is_isolated);
assert_eq!(config.isolated_symbols, symbols);
assert_eq!(config.side_effect, MarginSideEffect::AutoBorrowRepay);
}
#[test]
fn isolated_config_with_symbols_constructs() {
let config = BinanceMarginConfig::isolated(
"k".to_string(),
"s".to_string(),
vec![InstrumentNameExchange::new("BTCUSDT")],
);
let _client = BinanceMargin::new(config); }
#[test]
#[should_panic(expected = "non-empty isolated_symbols")]
fn isolated_empty_symbols_panics_at_new() {
let config = BinanceMarginConfig::new(
"k".to_string(),
"s".to_string(),
false,
true,
Vec::new(),
MarginSideEffect::default(),
);
let _ = BinanceMargin::new(config);
}
#[test]
#[should_panic(expected = "non-empty isolated_symbols")]
fn isolated_empty_symbols_panics_via_deserialize_path() {
let config: BinanceMarginConfig = serde_json::from_str(
r#"{"api_key":"k","secret_key":"s","testnet":false,"is_isolated":true}"#,
)
.expect("deserialize");
assert!(config.isolated_symbols.is_empty());
let _ = BinanceMargin::new(config);
}
#[test]
fn isolated_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,
side_effect: MarginSideEffect,
) -> Result<MarginAccountNewOrderParams, BuildOrderError> {
build_new_order_params(
"BTCUSDT".to_string(),
side,
price,
Decimal::from(2),
kind,
tif,
"cid-1".to_string(),
side_effect,
false, )
}
fn gtc() -> TimeInForce {
TimeInForce::GoodUntilCancelled { post_only: false }
}
#[test]
fn new_order_limit_maps_core_fields() {
let p = params(
Side::Buy,
Some(Decimal::from(50_000)),
OrderKind::Limit,
gtc(),
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(p.symbol, "BTCUSDT");
assert_eq!(p.side.as_str(), "BUY");
assert_eq!(p.r#type.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(),
MarginSideEffect::AutoBorrowRepay,
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(),
MarginSideEffect::AutoBorrowRepay,
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 auto = params(
Side::Buy,
Some(Decimal::from(1)),
OrderKind::Limit,
gtc(),
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(
auto.side_effect_type.as_ref().map(|s| s.as_str()),
Some("AUTO_BORROW_REPAY")
);
assert_eq!(auto.auto_repay_at_cancel, Some(true));
let no_borrow = params(
Side::Buy,
Some(Decimal::from(1)),
OrderKind::Limit,
gtc(),
MarginSideEffect::NoBorrow,
)
.expect("build");
assert_eq!(
no_borrow.side_effect_type.as_ref().map(|s| s.as_str()),
Some("NO_SIDE_EFFECT")
);
assert_eq!(no_borrow.auto_repay_at_cancel, None);
}
#[test]
fn new_order_market_has_no_price_or_tif() {
let p = params(
Side::Buy,
None,
OrderKind::Market,
gtc(),
MarginSideEffect::AutoBorrowRepay,
)
.expect("build");
assert_eq!(p.r#type.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 },
MarginSideEffect::AutoBorrowRepay,
)
.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(),
MarginSideEffect::AutoBorrowRepay,
)
.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(),
MarginSideEffect::AutoBorrowRepay,
)
.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(),
MarginSideEffect::AutoBorrowRepay,
)
.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,
},
gtc(),
MarginSideEffect::NoBorrow,
)
.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 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,
},
gtc(),
MarginSideEffect::AutoBorrowRepay,
);
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(),
MarginSideEffect::AutoBorrowRepay,
);
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,
},
gtc(),
MarginSideEffect::AutoBorrowRepay,
);
assert!(matches!(
fractional,
Err(BuildOrderError::TrailingDelta(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,
MarginSideEffect::AutoBorrowRepay,
)
.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 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()
}
#[test]
fn margin_ws_rpc_ack_yields_no_events() {
let frame = serde_json::json!({ "id": "abc", "status": 200, "result": {} }).to_string();
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
assert!(buf.is_empty());
}
#[test]
fn margin_ws_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!(!convert_margin_user_data_events(frame, &mut buf), "{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,
}
);
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));
}
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_execution_report_missing_symbol_is_dropped() {
let frame = push(serde_json::json!({
"e": "executionReport", "S": "BUY", "x": "TRADE", "i": 1_i64, "t": 1_i64,
"l": "1", "L": "100", "T": 1_700_000_000_000_i64,
}));
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
assert!(buf.is_empty());
}
#[test]
fn margin_ws_outbound_account_position_maps_to_balance_stream_updates() {
let frame = push(serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [
{ "a": "USDT", "f": "100.0", "l": "5.0" },
{ "a": "BTC", "f": "0.5", "l": "0" },
],
}));
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events(&frame, &mut buf));
assert_eq!(buf.len(), 2);
for ev in &buf {
assert!(matches!(ev.kind, AccountEventKind::BalanceStreamUpdate(_)));
}
}
#[test]
fn margin_ws_stream_terminated_signals_reconnect() {
let frame = push(serde_json::json!({ "e": "eventStreamTerminated" }));
let mut buf = Vec::new();
assert!(
convert_margin_user_data_events(&frame, &mut buf),
"eventStreamTerminated must signal reconnect"
);
assert!(buf.is_empty());
}
#[test]
fn margin_ws_margin_specific_events_are_observable_only() {
let mut buf = Vec::new();
let liability = push(serde_json::json!({
"e": "userLiabilityChange", "a": "USDT", "t": "BORROW", "p": "100", "i": "0.01",
}));
assert!(!convert_margin_user_data_events(&liability, &mut buf));
let level = push(serde_json::json!({
"e": "marginLevelStatusChange", "l": "1.5", "s": "MARGIN_LEVEL_2",
}));
assert!(!convert_margin_user_data_events(&level, &mut buf));
assert!(buf.is_empty());
}
#[test]
fn margin_trade_event_dedup_by_key() {
let frame = push(serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
"x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 4_242_i64,
"l": "1", "L": "100", "z": "1", "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 token(expiration_time_ms: i64) -> UserListenToken {
UserListenToken {
token: "t".to_string(),
expiration_time_ms,
}
}
#[test]
fn listen_token_query_cross_vs_isolated() {
assert!(build_listen_token_query(None).is_empty());
let q = build_listen_token_query(Some(&btcusdt()));
assert_eq!(q.get("isIsolated").and_then(|v| v.as_str()), Some("TRUE"));
assert_eq!(q.get("symbol").and_then(|v| v.as_str()), Some("BTCUSDT"));
assert_eq!(q.len(), 2);
}
#[test]
fn earliest_token_expiry_picks_min() {
let tokens = vec![
(btcusdt(), token(3_000)),
(InstrumentNameExchange::new("ETHUSDT"), token(1_000)),
(InstrumentNameExchange::new("BNBUSDT"), token(2_000)),
];
assert_eq!(earliest_token_expiry_ms(&tokens), 1_000);
assert_eq!(earliest_token_expiry_ms(&[]), i64::MAX);
}
#[test]
fn base_quote_map_extracts_base_and_quote() {
let entries = vec![
iso_entry(
"BTCUSDT",
iso_base("BTC", "0", "0", "0", "0"),
iso_quote("USDT", "0", "0", "0", "0"),
None,
None,
None,
),
iso_entry(
"ETHBTC",
iso_base("ETH", "0", "0", "0", "0"),
iso_quote("BTC", "0", "0", "0", "0"),
None,
None,
None,
),
];
let map = build_base_quote_map(&entries);
assert_eq!(
map.get(&btcusdt())
.map(|(b, q)| (b.name().as_str(), q.name().as_str())),
Some(("BTC", "USDT"))
);
assert_eq!(
map.get(&InstrumentNameExchange::new("ETHBTC"))
.map(|(b, q)| (b.name().as_str(), q.name().as_str())),
Some(("ETH", "BTC"))
);
}
#[test]
fn isolated_outbound_position_routes_to_instrument_balance_update() {
let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
let base_quote = Arc::new(HashMap::from([(
btcusdt(),
(
AssetNameExchange::new("BTC"),
AssetNameExchange::new("USDT"),
),
)]));
let mut handler = isolated_handler(sub_map, base_quote);
let frame = push_with_sub(
7,
serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [
{ "a": "USDT", "f": "1000.0", "l": "50.0" },
{ "a": "BTC", "f": "0.5", "l": "0.1" },
],
}),
);
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler
));
assert_eq!(buf.len(), 1, "one InstrumentBalanceUpdate for the pair");
match &buf[0].kind {
AccountEventKind::InstrumentBalanceUpdate(ibu) => {
assert_eq!(ibu.instrument.name().as_str(), "BTCUSDT");
assert_eq!(ibu.base.asset.name().as_str(), "BTC");
assert_eq!(ibu.base.update.free, Decimal::new(5, 1));
assert_eq!(ibu.base.update.locked, Decimal::new(1, 1));
assert_eq!(ibu.quote.asset.name().as_str(), "USDT");
assert_eq!(ibu.quote.update.free, Decimal::from(1000));
assert_eq!(ibu.quote.update.locked, Decimal::from(50));
}
other => panic!("expected InstrumentBalanceUpdate, got {other:?}"),
}
assert!(!matches!(
buf[0].kind,
AccountEventKind::BalanceStreamUpdate(_)
));
}
#[test]
fn isolated_outbound_position_unknown_subscription_dropped() {
let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
let base_quote = Arc::new(HashMap::from([(
btcusdt(),
(
AssetNameExchange::new("BTC"),
AssetNameExchange::new("USDT"),
),
)]));
let mut handler = isolated_handler(sub_map, base_quote);
let frame = push_with_sub(
99,
serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [{ "a": "BTC", "f": "1", "l": "0" }, { "a": "USDT", "f": "1", "l": "0" }],
}),
);
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler
));
assert!(buf.is_empty(), "unmapped subscriptionId frame is dropped");
}
#[test]
fn isolated_outbound_position_missing_side_dropped() {
let sub_map = Arc::new(Mutex::new(HashMap::from([(7_i64, btcusdt())])));
let base_quote = Arc::new(HashMap::from([(
btcusdt(),
(
AssetNameExchange::new("BTC"),
AssetNameExchange::new("USDT"),
),
)]));
let mut handler = isolated_handler(sub_map, base_quote);
let frame = push_with_sub(
7,
serde_json::json!({
"e": "outboundAccountPosition", "u": 1_700_000_000_000_i64,
"B": [{ "a": "BTC", "f": "0.5", "l": "0" }],
}),
);
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler
));
assert!(buf.is_empty(), "missing quote side → frame dropped");
}
#[test]
fn isolated_execution_report_routes_by_inner_symbol_independent_of_map() {
let sub_map = Arc::new(Mutex::new(HashMap::<i64, InstrumentNameExchange>::new()));
let base_quote = Arc::new(HashMap::new());
let mut handler = isolated_handler(sub_map, base_quote);
let frame = push_with_sub(
42,
serde_json::json!({
"e": "executionReport", "s": "BTCUSDT", "S": "BUY", "o": "LIMIT",
"x": "TRADE", "X": "FILLED", "i": 1_i64, "c": "cid", "t": 5_i64,
"l": "1", "L": "100", "z": "1", "p": "100", "q": "1", "f": "GTC",
"n": "0", "N": "USDT", "T": 1_700_000_000_000_i64,
}),
);
let mut buf = Vec::new();
assert!(!convert_margin_user_data_events_with(
&frame,
&mut buf,
&mut handler
));
assert_eq!(buf.len(), 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)
}
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");
}
}
#[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(),
))
}
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 mut client = <BinanceMargin as ExecutionClient>::new(config);
client.rest = margin_rest_at(server);
client
}
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,
};
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 {
side: Side::Buy,
price: Some(Decimal::from(100)),
quantity: Decimal::TWO,
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
position_id: None,
reduce_only: false,
market: None,
},
}
}
#[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", "AUTO_BORROW_REPAY")
&& !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,
};
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 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"
);
}
}
}