use std::{
future::Future,
sync::{Arc, Mutex},
time::Duration,
};
use ahash::{AHashMap, AHashSet};
use anyhow::Context;
use async_trait::async_trait;
use nautilus_common::{
cache::fifo::FifoCache,
clients::ExecutionClient,
live::{get_runtime, runner::get_exec_event_sender, task::TaskHandles},
messages::execution::{
BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
GenerateFillReportsBuilder, GenerateOrderStatusReport, GenerateOrderStatusReports,
GenerateOrderStatusReportsBuilder, GeneratePositionStatusReports,
GeneratePositionStatusReportsBuilder, ModifyOrder, QueryAccount, QueryOrder, SubmitOrder,
SubmitOrderList,
},
};
use nautilus_core::{
MUTEX_POISONED, UUID4, UnixNanos,
datetime::{NANOSECONDS_IN_MILLISECOND, checked_mins_to_nanos},
time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_live::{ExecutionClientCore, ExecutionEventEmitter};
use nautilus_model::{
accounts::AccountAny,
enums::{ContingencyType, LiquiditySide, OmsType, OrderStatus, OrderType, TimeInForce},
events::{
AccountState, OrderAccepted, OrderCancelRejected, OrderCanceled, OrderEventAny,
OrderExpired, OrderFilled, OrderModifyRejected, OrderRejected, OrderUpdated,
},
identifiers::{
AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, Venue,
VenueOrderId,
},
instruments::{Instrument, InstrumentAny},
orders::{Order, OrderAny},
reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
types::{AccountBalance, Currency, MarginBalance, Money, Price, Quantity},
};
use rust_decimal::Decimal;
use tokio::task::JoinHandle;
use ustr::Ustr;
use super::websocket::trading::{
client::BinanceSpotWsTradingClient,
messages::BinanceSpotWsTradingMessage,
parse::{
parse_spot_account_position, parse_spot_exec_report_to_fill,
parse_spot_exec_report_to_order_status,
},
user_data::{BinanceSpotExecutionReport, BinanceSpotExecutionType},
};
use crate::{
common::{
consts::{
BINANCE_GTX_ORDER_REJECT_CODE, BINANCE_NAUTILUS_SPOT_BROKER_ID,
BINANCE_NEW_ORDER_REJECTED_CODE, BINANCE_SPOT_POST_ONLY_REJECT_MSG,
BINANCE_STATUS_UNKNOWN_CODE, BINANCE_UNEXPECTED_RESPONSE_CODE, BINANCE_VENUE,
},
credential::resolve_credentials,
dispatch::{
OrderIdentity, PendingOperation, PendingRequest, WsDispatchState,
ensure_accepted_emitted,
},
encoder::{decode_client_order_id, encode_broker_id},
enums::{BinanceSide, BinanceTimeInForce},
parse::{
parse_millis_or_init, parse_required_decimal, parse_required_price_at_precision,
parse_required_quantity_at_precision,
},
urls::{get_http_base_url_with_us, get_spot_user_stream_url},
},
config::BinanceExecClientConfig,
spot::{
enums::{
BinanceCancelReplaceMode, BinanceOrderResponseType, BinanceSpotOrderType,
order_type_to_binance_spot, time_in_force_to_binance_spot,
},
http::{
client::BinanceSpotHttpClient,
error::BinanceSpotHttpError,
models::BatchCancelResult,
query::{
BatchCancelItem, CancelOrderParams, CancelReplaceOrderParams,
NewOcoOrderListParams, NewOrderParams,
},
},
},
};
const ACCOUNT_TRADES_MAX_INTERVAL_MS: i64 = 24 * 60 * 60 * 1_000;
const ACCOUNT_TRADES_PAGE_LIMIT: u32 = 1_000;
#[derive(Debug)]
pub struct BinanceSpotExecutionClient {
core: ExecutionClientCore,
clock: &'static AtomicTime,
config: BinanceExecClientConfig,
emitter: ExecutionEventEmitter,
dispatch_state: Arc<WsDispatchState>,
http_client: BinanceSpotHttpClient,
ws_trading_client: Option<BinanceSpotWsTradingClient>,
ws_trading_handle: Option<JoinHandle<()>>,
ws_user_data_client: Option<BinanceSpotWsTradingClient>,
ws_user_data_handle: Option<JoinHandle<()>>,
listen_key_keepalive_handle: Option<JoinHandle<()>>,
listen_key: Option<String>,
us_credentials: Option<(String, String)>,
ws_authenticated: Arc<tokio::sync::Notify>,
ws_user_data_subscribed: Arc<tokio::sync::Notify>,
pending_tasks: TaskHandles,
}
impl BinanceSpotExecutionClient {
pub fn new(core: ExecutionClientCore, config: BinanceExecClientConfig) -> anyhow::Result<Self> {
config.validate()?;
let (api_key, api_secret) = resolve_credentials(
config.api_key.clone(),
config.api_secret.clone(),
config.environment,
config.product_type,
)?;
let clock = get_atomic_clock_realtime();
let base_url_http = config.base_url_http.clone().or_else(|| {
config.us.then(|| {
get_http_base_url_with_us(config.product_type, config.environment, true).to_string()
})
});
let http_client = BinanceSpotHttpClient::new_with_json_responses(
config.environment,
clock,
Some(api_key.clone()),
Some(api_secret.clone()),
base_url_http,
Some(config.recv_window_ms),
None, config.proxy_url.clone(),
config.us,
)
.context("failed to construct Binance Spot HTTP client")?;
let emitter = ExecutionEventEmitter::new(
clock,
core.trader_id,
core.account_id,
core.account_type,
core.base_currency,
);
let ws_trading_client = if config.use_ws_trading && !config.us {
Some(
BinanceSpotWsTradingClient::new(
config.base_url_ws_trading.clone(),
api_key.clone(),
api_secret.clone(),
None, config.transport_backend,
)
.with_proxy(config.proxy_url.clone())
.with_recv_window(Some(config.recv_window_ms)),
)
} else {
None
};
let us_credentials = config.us.then_some((api_key, api_secret));
Ok(Self {
core,
clock,
config,
emitter,
dispatch_state: Arc::new(WsDispatchState::default()),
http_client,
ws_trading_client,
ws_trading_handle: None,
ws_user_data_client: None,
ws_user_data_handle: None,
listen_key_keepalive_handle: None,
listen_key: None,
us_credentials,
ws_authenticated: Arc::new(tokio::sync::Notify::new()),
ws_user_data_subscribed: Arc::new(tokio::sync::Notify::new()),
pending_tasks: TaskHandles::default(),
})
}
async fn refresh_account_state(&self) -> anyhow::Result<AccountState> {
self.http_client
.request_account_state(self.core.account_id)
.await
}
fn update_account_state(&self) {
let http_client = self.http_client.clone();
let account_id = self.core.account_id;
let emitter = self.emitter.clone();
let clock = self.clock;
self.spawn_task("query_account", async move {
let account_state = http_client.request_account_state(account_id).await?;
let ts_now = clock.get_time_ns();
emitter.emit_account_state(
account_state.balances.clone(),
account_state.margins.clone(),
account_state.is_reported,
ts_now,
);
Ok(())
});
}
fn ws_trading_active(&self) -> bool {
let dispatch_running = self
.ws_trading_handle
.as_ref()
.is_some_and(|handle| !handle.is_finished());
self.ws_trading_client
.as_ref()
.is_some_and(|client| client.is_active())
&& dispatch_running
}
fn submit_order_internal(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;
let event_emitter = self.emitter.clone();
let trader_id = self.core.trader_id;
let account_id = self.core.account_id;
let client_order_id = order.client_order_id();
let strategy_id = order.strategy_id();
let instrument_id = order.instrument_id();
let order_side = order.order_side();
let order_type = order.order_type();
let quantity = order.quantity();
let time_in_force = order.time_in_force();
let price = order.price();
let trigger_price = order.trigger_price();
let is_post_only = order.is_post_only();
let is_quote_quantity = order.is_quote_quantity();
let display_qty = order.display_qty();
let use_gtd = self.config.use_gtd;
let clock = self.clock;
let ts_init = self.clock.get_time_ns();
self.dispatch_state.order_identities.insert(
client_order_id,
OrderIdentity {
instrument_id,
strategy_id,
order_side,
order_type,
price,
quantity,
},
);
if self.ws_trading_active() {
let ws_client = self.ws_trading_client.as_ref().unwrap().clone();
let dispatch_state = self.dispatch_state.clone();
let params = build_new_order_params(
&order,
client_order_id,
is_post_only,
is_quote_quantity,
use_gtd,
)?;
let request_id = ws_client.next_request_id();
dispatch_state.pending_requests.insert(
request_id.clone(),
PendingRequest {
client_order_id,
venue_order_id: None,
operation: PendingOperation::Place,
},
);
self.spawn_task("submit_order_ws", async move {
if let Err(e) = ws_client
.place_order_with_id(request_id.clone(), params)
.await
{
dispatch_state.pending_requests.remove(&request_id);
log::warn!(
"WS submit request failed for {client_order_id}, awaiting reconciliation: {e}"
);
anyhow::bail!("WS submit order failed: {e}");
}
Ok(())
});
} else {
let http_client = self.http_client.clone();
let dispatch_state = self.dispatch_state.clone();
log::debug!("WS trading not active, falling back to HTTP for submit_order");
self.spawn_task("submit_order_http", async move {
let result = http_client
.submit_order(
account_id,
instrument_id,
client_order_id,
order_side,
order_type,
quantity,
time_in_force,
price,
trigger_price,
is_post_only,
is_quote_quantity,
display_qty,
use_gtd,
)
.await;
match result {
Ok(report) => {
dispatch_state.insert_accepted(client_order_id);
let accepted = OrderAccepted::new(
trader_id,
strategy_id,
instrument_id,
client_order_id,
report.venue_order_id,
account_id,
UUID4::new(),
ts_init,
ts_init,
false,
);
event_emitter.send_order_event(OrderEventAny::Accepted(accepted));
}
Err(e) => {
if is_ambiguous_submit_error(&e) {
log::warn!(
"Ambiguous submit failure for {client_order_id}, awaiting reconciliation: {e}"
);
} else if is_structured_venue_rejection(&e)
|| is_local_command_failure(&e)
{
let due_post_only = e
.downcast_ref::<BinanceSpotHttpError>()
.is_some_and(is_spot_post_only_rejection);
dispatch_state.cleanup_terminal(client_order_id);
let rejected = OrderRejected::new(
trader_id,
strategy_id,
instrument_id,
client_order_id,
account_id,
format!("submit-order-error: {e}").into(),
UUID4::new(),
ts_init,
clock.get_time_ns(),
false,
due_post_only,
);
event_emitter.send_order_event(OrderEventAny::Rejected(rejected));
} else {
log::warn!(
"Ambiguous submit failure for {client_order_id}, awaiting reconciliation: {e}"
);
}
return Err(e);
}
}
Ok(())
});
}
Ok(())
}
fn cancel_order_internal(&self, cmd: &CancelOrder) {
let event_emitter = self.emitter.clone();
let trader_id = self.core.trader_id;
let account_id = self.core.account_id;
let clock = self.clock;
let command = cmd.clone();
if self.ws_trading_active() {
let ws_client = self.ws_trading_client.as_ref().unwrap().clone();
let dispatch_state = self.dispatch_state.clone();
let params = build_cancel_order_params(&command);
let request_id = ws_client.next_request_id();
dispatch_state.pending_requests.insert(
request_id.clone(),
PendingRequest {
client_order_id: command.client_order_id,
venue_order_id: command.venue_order_id,
operation: PendingOperation::Cancel,
},
);
self.spawn_task("cancel_order_ws", async move {
if let Err(e) = ws_client
.cancel_order_with_id(request_id.clone(), params)
.await
{
dispatch_state.pending_requests.remove(&request_id);
log::warn!(
"WS cancel request failed for {}, awaiting reconciliation: {e}",
command.client_order_id
);
anyhow::bail!("WS cancel order failed: {e}");
}
Ok(())
});
} else {
let http_client = self.http_client.clone();
let dispatch_state = self.dispatch_state.clone();
log::debug!("WS trading not active, falling back to HTTP for cancel_order");
self.spawn_task("cancel_order_http", async move {
let result = http_client
.cancel_order(
command.instrument_id,
command.venue_order_id,
Some(command.client_order_id),
)
.await;
match result {
Ok(venue_order_id) => {
dispatch_state.cleanup_terminal(command.client_order_id);
let ts_now = clock.get_time_ns();
let canceled_event = OrderCanceled::new(
trader_id,
command.strategy_id,
command.instrument_id,
command.client_order_id,
UUID4::new(),
ts_now,
ts_now,
false,
Some(venue_order_id),
Some(account_id),
);
event_emitter.send_order_event(OrderEventAny::Canceled(canceled_event));
}
Err(e) => {
if is_structured_venue_rejection(&e) {
let ts_now = clock.get_time_ns();
let rejected_event = OrderCancelRejected::new(
trader_id,
command.strategy_id,
command.instrument_id,
command.client_order_id,
format!("cancel-order-error: {e}").into(),
UUID4::new(),
ts_now,
ts_now,
false,
command.venue_order_id,
Some(account_id),
);
event_emitter
.send_order_event(OrderEventAny::CancelRejected(rejected_event));
} else if is_local_command_failure(&e) {
log::warn!(
"Cancel command failed local validation for {}: {e}",
command.client_order_id
);
} else {
log::warn!(
"Ambiguous cancel failure for {}, awaiting reconciliation: {e}",
command.client_order_id
);
}
return Err(e);
}
}
Ok(())
});
}
}
fn spawn_task<F>(&self, description: &'static str, fut: F)
where
F: Future<Output = anyhow::Result<()>> + Send + 'static,
{
crate::common::execution::spawn_task(&self.pending_tasks, description, fut);
}
fn abort_pending_tasks(&self) {
crate::common::execution::abort_pending_tasks(&self.pending_tasks);
}
async fn enter_http_only_execution_mode(
&mut self,
mut ws_trading: BinanceSpotWsTradingClient,
reason: &str,
) {
log::error!(
"{reason}; entering Spot HTTP-only execution mode. Order commands use HTTP responses; execution reconciliation requires explicit queries until WS trading is re-enabled"
);
if let Some(handle) = self.ws_trading_handle.take() {
handle.abort();
}
ws_trading.disconnect().await;
self.ws_trading_client = Some(ws_trading);
}
async fn connect_us_user_data(&mut self) -> anyhow::Result<()> {
let (api_key, api_secret) = self
.us_credentials
.clone()
.context("Binance US user data credentials are unavailable")?;
let listen_key = self
.http_client
.inner()
.create_listen_key()
.await
.context("failed to create Binance US listen key")?
.listen_key;
let url = get_spot_user_stream_url(self.config.base_url_ws.as_deref(), &listen_key);
let mut ws_user_data = BinanceSpotWsTradingClient::new(
Some(url),
api_key,
api_secret,
Some(20),
self.config.transport_backend,
)
.with_proxy(self.config.proxy_url.clone());
ws_user_data
.connect()
.await
.context("failed to connect Binance US user data stream")?;
let ws_clone = ws_user_data.clone();
let emitter = self.emitter.clone();
let account_id = self.core.account_id;
let clock = self.clock;
let http_client = self.http_client.clone();
let dispatch_state = self.dispatch_state.clone();
let treat_expired_as_canceled = self.config.treat_expired_as_canceled;
let ws_authenticated = self.ws_authenticated.clone();
let ws_user_data_subscribed = self.ws_user_data_subscribed.clone();
let (setup_error_tx, _setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
let handle = get_runtime().spawn(async move {
while let Some(message) = ws_clone.recv().await {
dispatch_ws_trading_message(
message,
&emitter,
&http_client,
account_id,
treat_expired_as_canceled,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&setup_error_tx,
&seen_trade_ids,
);
}
log::warn!("Binance US user data dispatch loop ended");
});
self.ws_user_data_handle = Some(handle);
let keepalive_http = self.http_client.clone();
let keepalive_key = listen_key.clone();
let keepalive = get_runtime().spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(30 * 60));
interval.tick().await;
loop {
interval.tick().await;
if let Err(e) = keepalive_http
.inner()
.extend_listen_key(&keepalive_key)
.await
{
log::warn!("Binance US listen key keepalive failed: {e}");
}
}
});
self.listen_key_keepalive_handle = Some(keepalive);
self.listen_key = Some(listen_key);
self.ws_user_data_client = Some(ws_user_data);
Ok(())
}
async fn disconnect_us_user_data(&mut self) {
if let Some(handle) = self.ws_user_data_handle.take() {
handle.abort();
}
if let Some(handle) = self.listen_key_keepalive_handle.take() {
handle.abort();
}
if let Some(client) = self.ws_user_data_client.as_mut() {
client.disconnect().await;
}
if let Some(listen_key) = self.listen_key.take()
&& let Err(e) = self.http_client.inner().close_listen_key(&listen_key).await
{
log::warn!("Failed to close Binance US listen key: {e}");
}
self.ws_user_data_client = None;
}
}
#[async_trait(?Send)]
impl ExecutionClient for BinanceSpotExecutionClient {
fn is_connected(&self) -> bool {
self.core.is_connected()
}
fn client_id(&self) -> ClientId {
self.core.client_id
}
fn account_id(&self) -> AccountId {
self.core.account_id
}
fn venue(&self) -> Venue {
*BINANCE_VENUE
}
fn oms_type(&self) -> OmsType {
self.core.oms_type
}
fn get_account(&self) -> Option<AccountAny> {
self.core.cache().account_owned(&self.core.account_id)
}
async fn connect(&mut self) -> anyhow::Result<()> {
if self.core.is_connected() {
return Ok(());
}
let ws_setup_timeout = Duration::from_millis(self.config.ws_trading_setup_timeout_ms);
if !self.core.instruments_initialized() {
let instruments = self
.http_client
.request_instruments_with_config(&self.config.instrument_provider, self.config.us)
.await
.context("failed to request Binance Spot instruments")?;
if instruments.is_empty() {
log::warn!("No instruments returned for Binance Spot");
} else {
log::debug!("Loaded {} Spot instruments", instruments.len());
self.http_client.cache_instruments(instruments);
}
self.core.set_instruments_initialized();
}
let account_state = self
.refresh_account_state()
.await
.context("failed to request Binance account state")?;
if !account_state.balances.is_empty() {
log::debug!(
"Received account state with {} balance(s)",
account_state.balances.len()
);
}
self.emitter.send_account_state(account_state);
crate::common::execution::await_account_registered(&self.core, self.core.account_id, 30.0)
.await?;
if self.config.us {
self.connect_us_user_data().await?;
}
if let Some(mut ws_trading) = self.ws_trading_client.take() {
match ws_trading.connect().await {
Ok(()) => {
log::debug!("Connected to Binance Spot WS trading API");
let ws_trading_clone = ws_trading.clone();
let emitter = self.emitter.clone();
let account_id = self.core.account_id;
let clock = self.clock;
let http_client = self.http_client.clone();
let dispatch_state = self.dispatch_state.clone();
let treat_expired_as_canceled = self.config.treat_expired_as_canceled;
let ws_authenticated = self.ws_authenticated.clone();
let ws_user_data_subscribed = self.ws_user_data_subscribed.clone();
let (ws_setup_error_tx, mut ws_setup_error_rx) =
tokio::sync::mpsc::unbounded_channel();
let seen_trade_ids = std::sync::Arc::new(Mutex::new(FifoCache::new()));
let handle = get_runtime().spawn(async move {
loop {
match ws_trading_clone.recv().await {
Some(msg) => {
dispatch_ws_trading_message(
msg,
&emitter,
&http_client,
account_id,
treat_expired_as_canceled,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&ws_setup_error_tx,
&seen_trade_ids,
);
}
None => {
log::warn!("WS trading dispatch loop ended");
break;
}
}
}
});
self.ws_trading_handle = Some(handle);
if let Err(e) = ws_trading.session_logon().await {
let reason = format!("WS session logon failed: {e}");
self.enter_http_only_execution_mode(ws_trading, &reason)
.await;
} else {
let auth_result = wait_for_ws_setup_response(
ws_setup_timeout,
self.ws_authenticated.notified(),
&mut ws_setup_error_rx,
"WS session authentication timed out",
)
.await;
if let Err(e) = auth_result {
self.enter_http_only_execution_mode(ws_trading, &e.to_string())
.await;
} else if let Err(e) = ws_trading.subscribe_user_data().await {
let reason = format!("WS user data subscribe failed: {e}");
self.enter_http_only_execution_mode(ws_trading, &reason)
.await;
} else {
let subscribe_result = wait_for_ws_setup_response(
ws_setup_timeout,
self.ws_user_data_subscribed.notified(),
&mut ws_setup_error_rx,
"WS user data subscription timed out",
)
.await;
if let Err(e) = subscribe_result {
self.enter_http_only_execution_mode(ws_trading, &e.to_string())
.await;
} else {
self.ws_trading_client = Some(ws_trading);
}
}
}
}
Err(e) => {
let reason = format!("Failed to connect WS trading API: {e}");
self.enter_http_only_execution_mode(ws_trading, &reason)
.await;
}
}
}
let refresh_secs = self.config.instrument_refresh_interval_secs;
if refresh_secs > 0 {
let http_client = self.http_client.clone();
let provider = self.config.instrument_provider.clone();
let us = self.config.us;
self.spawn_task("instrument_refresh", async move {
let mut interval = tokio::time::interval(Duration::from_secs(refresh_secs));
interval.tick().await;
loop {
interval.tick().await;
match http_client
.request_instruments_with_config(&provider, us)
.await
{
Ok(instruments) => log::debug!(
"Refreshed Binance Spot execution instruments: count={}",
instruments.len()
),
Err(e) => {
log::warn!("Binance Spot execution instrument refresh failed: {e}");
}
}
}
});
}
self.core.set_connected();
log::info!("Connected: client_id={}", self.core.client_id);
Ok(())
}
async fn disconnect(&mut self) -> anyhow::Result<()> {
if self.core.is_disconnected() {
return Ok(());
}
if let Some(handle) = self.ws_trading_handle.take() {
handle.abort();
}
if let Some(ref mut ws_trading) = self.ws_trading_client {
ws_trading.disconnect().await;
}
self.disconnect_us_user_data().await;
self.abort_pending_tasks();
self.core.set_disconnected();
log::info!("Disconnected: client_id={}", self.core.client_id);
Ok(())
}
fn query_account(&self, _cmd: QueryAccount) -> anyhow::Result<()> {
self.update_account_state();
Ok(())
}
fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
log::debug!("query_order: client_order_id={}", cmd.client_order_id);
let http_client = self.http_client.clone();
let command = cmd;
let event_emitter = self.emitter.clone();
let account_id = self.core.account_id;
let treat_expired_as_canceled = self.config.treat_expired_as_canceled;
self.spawn_task("query_order", async move {
let result = http_client
.request_order_status_report(
account_id,
command.instrument_id,
command.venue_order_id,
Some(command.client_order_id),
)
.await;
match result {
Ok(Some(mut report)) => {
normalize_spot_order_status_report(&mut report, treat_expired_as_canceled);
event_emitter.send_order_status_report(report);
}
Ok(None) => log::debug!(
"No order status report returned: client_order_id={}",
command.client_order_id
),
Err(e) => log::warn!("Failed to query order status: {e}"),
}
Ok(())
});
Ok(())
}
fn generate_account_state(
&self,
balances: Vec<AccountBalance>,
margins: Vec<MarginBalance>,
reported: bool,
ts_event: UnixNanos,
) -> anyhow::Result<()> {
self.emitter
.emit_account_state(balances, margins, reported, ts_event);
Ok(())
}
fn start(&mut self) -> anyhow::Result<()> {
if self.core.is_started() {
return Ok(());
}
self.emitter.set_sender(get_exec_event_sender());
self.core.set_started();
let http_client = self.http_client.clone();
let provider = self.config.instrument_provider.clone();
let us = self.config.us;
get_runtime().spawn(async move {
match http_client
.request_instruments_with_config(&provider, us)
.await
{
Ok(instruments) => {
if instruments.is_empty() {
log::warn!("No instruments returned for Binance Spot");
} else {
http_client.cache_instruments(instruments);
log::debug!("Instruments initialized");
}
}
Err(e) => {
log::error!("Failed to request Binance Spot instruments: {e}");
}
}
});
log::info!(
"Started: client_id={}, account_id={}, account_type={:?}, environment={:?}, product_type={:?}",
self.core.client_id,
self.core.account_id,
self.core.account_type,
self.config.environment,
self.config.product_type,
);
Ok(())
}
fn stop(&mut self) -> anyhow::Result<()> {
if self.core.is_stopped() {
return Ok(());
}
if let Some(handle) = self.ws_trading_handle.take() {
handle.abort();
}
if let Some(handle) = self.ws_user_data_handle.take() {
handle.abort();
}
if let Some(handle) = self.listen_key_keepalive_handle.take() {
handle.abort();
}
self.core.set_stopped();
self.core.set_disconnected();
self.abort_pending_tasks();
log::info!("Stopped: client_id={}", self.core.client_id);
Ok(())
}
async fn generate_order_status_report(
&self,
cmd: &GenerateOrderStatusReport,
) -> anyhow::Result<Option<OrderStatusReport>> {
let Some(instrument_id) = cmd.instrument_id else {
log::warn!("generate_order_status_report requires instrument_id: {cmd:?}");
return Ok(None);
};
let venue_order_id = cmd
.venue_order_id
.as_ref()
.map(|id| VenueOrderId::new(id.inner()));
let report = self
.http_client
.request_order_status_report(
self.core.account_id,
instrument_id,
venue_order_id,
cmd.client_order_id,
)
.await?;
Ok(report.map(|mut report| {
normalize_spot_order_status_report(&mut report, self.config.treat_expired_as_canceled);
report
}))
}
async fn generate_order_status_reports(
&self,
cmd: &GenerateOrderStatusReports,
) -> anyhow::Result<Vec<OrderStatusReport>> {
let start_dt = cmd.start.map(|nanos| nanos.to_datetime_utc());
let end_dt = cmd.end.map(|nanos| nanos.to_datetime_utc());
let mut reports = self
.http_client
.request_order_status_reports(
self.core.account_id,
cmd.instrument_id,
start_dt,
end_dt,
cmd.open_only,
None, )
.await?;
normalize_spot_order_status_reports(&mut reports, self.config.treat_expired_as_canceled);
Ok(reports)
}
async fn generate_fill_reports(
&self,
cmd: GenerateFillReports,
) -> anyhow::Result<Vec<FillReport>> {
let Some(instrument_id) = cmd.instrument_id else {
log::warn!("generate_fill_reports requires instrument_id for Binance Spot");
return Ok(Vec::new());
};
let venue_order_id = cmd
.venue_order_id
.as_ref()
.map(|id| VenueOrderId::new(id.inner()));
let requested_start_time = cmd
.start
.map(|start| start.as_i64() / NANOSECONDS_IN_MILLISECOND as i64);
let requested_end_time = cmd
.end
.map(|end| end.as_i64() / NANOSECONDS_IN_MILLISECOND as i64);
if let (Some(start), Some(end)) = (requested_start_time, requested_end_time) {
anyhow::ensure!(
start <= end,
"fill report start time must not exceed end time"
);
}
let mut reports = Vec::new();
let mut seen_trade_ids = AHashSet::new();
if venue_order_id.is_some() {
let mut from_id = 0;
loop {
let page = self
.http_client
.request_fill_reports_with_cursor(
self.core.account_id,
instrument_id,
venue_order_id,
None,
None,
Some(from_id),
Some(ACCOUNT_TRADES_PAGE_LIMIT),
)
.await?;
if page.is_empty() {
break;
}
let page_len = page.len();
let max_trade_id = max_trade_id(&page)?;
let passed_end = requested_end_time.is_some_and(|end_time| {
page.iter().any(|report| report_time_ms(report) > end_time)
});
reports.extend(page.into_iter().filter(|report| {
requested_start_time
.is_none_or(|start_time| report_time_ms(report) >= start_time)
&& requested_end_time
.is_none_or(|end_time| report_time_ms(report) <= end_time)
&& seen_trade_ids.insert(report.trade_id)
}));
if page_len < ACCOUNT_TRADES_PAGE_LIMIT as usize || passed_end {
break;
}
let next_from_id = max_trade_id
.checked_add(1)
.context("Binance Spot trade ID overflow during pagination")?;
anyhow::ensure!(
next_from_id > from_id,
"Binance Spot account-trades pagination made no progress"
);
from_id = next_from_id;
}
} else if let Some(query_start_time) = requested_start_time {
let query_end_time = requested_end_time.unwrap_or_else(|| {
self.clock.get_time_ns().as_i64() / NANOSECONDS_IN_MILLISECOND as i64
});
anyhow::ensure!(
query_start_time <= query_end_time,
"fill report start time must not exceed end time"
);
let mut window_start = query_start_time;
loop {
let window_end = window_start
.saturating_add(ACCOUNT_TRADES_MAX_INTERVAL_MS)
.min(query_end_time);
let mut from_id = None;
loop {
let start = if from_id.is_none() {
Some(
chrono::DateTime::from_timestamp_millis(window_start)
.context("invalid Binance Spot account-trades start time")?,
)
} else {
None
};
let end = if from_id.is_none() {
Some(
chrono::DateTime::from_timestamp_millis(window_end)
.context("invalid Binance Spot account-trades end time")?,
)
} else {
None
};
let page = self
.http_client
.request_fill_reports_with_cursor(
self.core.account_id,
instrument_id,
None,
start,
end,
from_id,
Some(ACCOUNT_TRADES_PAGE_LIMIT),
)
.await?;
if page.is_empty() {
break;
}
let page_len = page.len();
let max_trade_id = max_trade_id(&page)?;
let passed_window_end = page
.iter()
.any(|report| report_time_ms(report) > window_end);
reports.extend(page.into_iter().filter(|report| {
let report_time = report_time_ms(report);
report_time >= window_start
&& report_time <= window_end
&& seen_trade_ids.insert(report.trade_id)
}));
if page_len < ACCOUNT_TRADES_PAGE_LIMIT as usize || passed_window_end {
break;
}
let next_from_id = max_trade_id
.checked_add(1)
.context("Binance Spot trade ID overflow during pagination")?;
anyhow::ensure!(
from_id.is_none_or(|cursor| next_from_id > cursor),
"Binance Spot account-trades pagination made no progress"
);
from_id = Some(next_from_id);
}
if window_end >= query_end_time {
break;
}
window_start = window_end.saturating_add(1);
}
} else {
let mut from_id = 0;
loop {
let page = self
.http_client
.request_fill_reports_with_cursor(
self.core.account_id,
instrument_id,
None,
None,
None,
Some(from_id),
Some(ACCOUNT_TRADES_PAGE_LIMIT),
)
.await?;
if page.is_empty() {
break;
}
let page_len = page.len();
let max_trade_id = max_trade_id(&page)?;
let passed_end = requested_end_time.is_some_and(|end_time| {
page.iter().any(|report| report_time_ms(report) > end_time)
});
reports.extend(page.into_iter().filter(|report| {
requested_end_time.is_none_or(|end_time| report_time_ms(report) <= end_time)
&& seen_trade_ids.insert(report.trade_id)
}));
if page_len < ACCOUNT_TRADES_PAGE_LIMIT as usize || passed_end {
break;
}
let next_from_id = max_trade_id
.checked_add(1)
.context("Binance Spot trade ID overflow during pagination")?;
anyhow::ensure!(
next_from_id > from_id,
"Binance Spot account-trades pagination made no progress"
);
from_id = next_from_id;
}
}
let mut reports_with_trade_ids = reports
.into_iter()
.map(|report| parse_trade_id(&report).map(|trade_id| (report, trade_id)))
.collect::<anyhow::Result<Vec<_>>>()?;
reports_with_trade_ids
.sort_unstable_by_key(|(report, trade_id)| (report.ts_event, *trade_id));
Ok(reports_with_trade_ids
.into_iter()
.map(|(report, _)| report)
.collect())
}
async fn generate_position_status_reports(
&self,
_cmd: &GeneratePositionStatusReports,
) -> anyhow::Result<Vec<PositionStatusReport>> {
Ok(Vec::new())
}
async fn generate_mass_status(
&self,
lookback_mins: Option<u64>,
) -> anyhow::Result<Option<ExecutionMassStatus>> {
log::info!("Generating ExecutionMassStatus (lookback_mins={lookback_mins:?})");
let ts_now = self.clock.get_time_ns();
let start = if let Some(mins) = lookback_mins {
let lookback_ns = checked_mins_to_nanos(mins)
.context("lookback minutes exceed the nanosecond range")?;
Some(UnixNanos::from(ts_now.as_u64().saturating_sub(lookback_ns)))
} else {
None
};
let order_cmd = GenerateOrderStatusReportsBuilder::default()
.ts_init(ts_now)
.open_only(true)
.start(start)
.build()
.map_err(|e| anyhow::anyhow!("{e}"))?;
let position_cmd = GeneratePositionStatusReportsBuilder::default()
.ts_init(ts_now)
.start(start)
.build()
.map_err(|e| anyhow::anyhow!("{e}"))?;
let (order_reports, position_reports) = tokio::try_join!(
self.generate_order_status_reports(&order_cmd),
self.generate_position_status_reports(&position_cmd),
)?;
let mut instrument_ids: Vec<_> = order_reports
.iter()
.map(|report| report.instrument_id)
.collect();
{
let cache = self.core.cache();
instrument_ids.extend(
cache
.orders_open(
Some(&BINANCE_VENUE),
None,
None,
Some(&self.core.account_id),
None,
)
.into_iter()
.chain(cache.orders_inflight(
Some(&BINANCE_VENUE),
None,
None,
Some(&self.core.account_id),
None,
))
.map(|order| order.instrument_id())
.filter(|instrument_id| {
matches!(
cache.instrument(instrument_id),
Some(InstrumentAny::CurrencyPair(_))
)
}),
);
}
instrument_ids.sort_unstable();
instrument_ids.dedup();
let mut fill_reports = Vec::new();
for instrument_id in instrument_ids {
let fill_cmd = GenerateFillReportsBuilder::default()
.ts_init(ts_now)
.instrument_id(Some(instrument_id))
.start(start)
.build()
.map_err(|e| anyhow::anyhow!("{e}"))?;
fill_reports.extend(self.generate_fill_reports(fill_cmd).await?);
}
log::info!("Received {} OrderStatusReports", order_reports.len());
log::info!("Received {} FillReports", fill_reports.len());
log::info!("Received {} PositionReports", position_reports.len());
let mut mass_status = ExecutionMassStatus::new(
self.core.client_id,
self.core.account_id,
*BINANCE_VENUE,
ts_now,
None,
);
mass_status.add_order_reports(order_reports);
mass_status.add_fill_reports(fill_reports);
mass_status.add_position_reports(position_reports);
Ok(Some(mass_status))
}
fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;
if order.is_closed() {
let client_order_id = order.client_order_id();
log::warn!("Cannot submit closed order {client_order_id}");
return Ok(());
}
if order.time_in_force() == TimeInForce::Gtd && self.config.use_gtd {
time_in_force_to_binance_spot(order.time_in_force(), self.config.use_gtd)?;
}
log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
self.emitter.emit_order_submitted(&order);
self.submit_order_internal(&cmd)
}
fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
if cmd.order_list.client_order_ids.is_empty() {
log::debug!("submit_order_list called with empty order list");
return Ok(());
}
let orders = self.core.get_orders_for_list(&cmd.order_list)?;
if let Some(order) = orders.iter().find(|order| order.is_closed()) {
let reason = format!("Cannot submit closed order {}", order.client_order_id());
for order in &orders {
self.emitter.emit_order_denied(order, &reason);
}
return Ok(());
}
let params = match build_spot_order_list_params(
cmd.order_list.id.as_ref(),
&orders,
self.config.use_gtd,
) {
Ok(request) => request,
Err(reason) => {
for order in &orders {
self.emitter.emit_order_denied(order, &reason);
}
return Ok(());
}
};
for order in &orders {
self.dispatch_state.order_identities.insert(
order.client_order_id(),
OrderIdentity {
instrument_id: order.instrument_id(),
strategy_id: order.strategy_id(),
order_side: order.order_side(),
order_type: order.order_type(),
price: order.price(),
quantity: order.quantity(),
},
);
self.emitter.emit_order_submitted(order);
}
let event_emitter = self.emitter.clone();
let trader_id = self.core.trader_id;
let account_id = self.core.account_id;
let clock = self.clock;
let ts_init = self.clock.get_time_ns();
let http_client = self.http_client.clone();
let dispatch_state = self.dispatch_state.clone();
self.spawn_task("submit_order_list_http", async move {
let response = http_client.submit_oco_order_list(¶ms).await;
match response {
Ok(response) => {
for order in &orders {
let encoded_client_order_id = encode_broker_id(
&order.client_order_id(),
BINANCE_NAUTILUS_SPOT_BROKER_ID,
);
let Some(report) = response
.orders
.iter()
.find(|report| report.client_order_id == encoded_client_order_id)
else {
log::warn!(
"OCO response missing leg for {}, awaiting reconciliation",
order.client_order_id()
);
continue;
};
let client_order_id = order.client_order_id();
dispatch_state.insert_accepted(client_order_id);
let accepted = OrderAccepted::new(
trader_id,
order.strategy_id(),
order.instrument_id(),
client_order_id,
VenueOrderId::from(report.order_id.to_string().as_str()),
account_id,
UUID4::new(),
ts_init,
clock.get_time_ns(),
false,
);
event_emitter.send_order_event(OrderEventAny::Accepted(accepted));
}
}
Err(e) => {
handle_spot_order_list_submit_error(
&event_emitter,
&dispatch_state,
trader_id,
account_id,
clock,
&orders,
e,
)?;
}
}
Ok(())
});
Ok(())
}
fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
let order = self
.core
.cache()
.order(&cmd.client_order_id)
.map(|o| o.clone());
let Some(order) = order else {
log::warn!(
"Cannot modify order {}: not found in cache",
cmd.client_order_id
);
let ts_init = self.clock.get_time_ns();
let rejected_event = OrderModifyRejected::new(
self.core.trader_id,
cmd.strategy_id,
cmd.instrument_id,
cmd.client_order_id,
"Order not found in cache for modify".into(),
UUID4::new(),
ts_init, ts_init,
false,
cmd.venue_order_id,
Some(self.core.account_id),
);
self.emitter
.send_order_event(OrderEventAny::ModifyRejected(rejected_event));
return Ok(());
};
let event_emitter = self.emitter.clone();
let trader_id = self.core.trader_id;
let account_id = self.core.account_id;
let clock = self.clock;
let order_side = order.order_side();
let order_type = order.order_type();
let time_in_force = order.time_in_force();
let quantity = cmd.quantity.unwrap_or_else(|| order.quantity());
let use_gtd = self.config.use_gtd;
if self.ws_trading_active() {
let command = cmd;
let ws_client = self.ws_trading_client.as_ref().unwrap().clone();
let dispatch_state = self.dispatch_state.clone();
let params = build_cancel_replace_params(&command, &order, quantity, use_gtd)?;
let request_id = ws_client.next_request_id();
dispatch_state.pending_requests.insert(
request_id.clone(),
PendingRequest {
client_order_id: command.client_order_id,
venue_order_id: command.venue_order_id,
operation: PendingOperation::Modify,
},
);
self.spawn_task("modify_order_ws", async move {
if let Err(e) = ws_client
.cancel_replace_order_with_id(request_id.clone(), params)
.await
{
dispatch_state.pending_requests.remove(&request_id);
log::warn!(
"WS modify request failed for {}, awaiting reconciliation: {e}",
command.client_order_id
);
anyhow::bail!("WS modify order failed: {e}");
}
Ok(())
});
} else {
let command = cmd;
let http_client = self.http_client.clone();
log::debug!("WS trading not active, falling back to HTTP for modify_order");
self.spawn_task("modify_order_http", async move {
let result = match command.venue_order_id {
Some(venue_order_id) => {
http_client
.modify_order(
account_id,
command.instrument_id,
venue_order_id,
command.client_order_id,
order_side,
order_type,
quantity,
time_in_force,
command.price,
use_gtd,
)
.await
}
None => Err(anyhow::anyhow!(BinanceSpotHttpError::ValidationError(
"venue_order_id required for modify".to_string()
))),
};
match result {
Ok(report) => {
let ts_now = clock.get_time_ns();
let updated_event = OrderUpdated::new(
trader_id,
command.strategy_id,
command.instrument_id,
command.client_order_id,
report.quantity,
UUID4::new(),
ts_now,
ts_now,
false,
Some(report.venue_order_id),
Some(account_id),
report.price,
None, None, false, );
event_emitter.send_order_event(OrderEventAny::Updated(updated_event));
}
Err(e) => {
if is_structured_venue_rejection(&e) || is_local_command_failure(&e) {
let ts_now = clock.get_time_ns();
let rejected_event = OrderModifyRejected::new(
trader_id,
command.strategy_id,
command.instrument_id,
command.client_order_id,
format!("modify-order-error: {e}").into(),
UUID4::new(),
ts_now,
ts_now,
false,
command.venue_order_id,
Some(account_id),
);
event_emitter
.send_order_event(OrderEventAny::ModifyRejected(rejected_event));
} else {
log::warn!(
"Ambiguous modify failure for {}, awaiting reconciliation: {e}",
command.client_order_id
);
}
return Err(e);
}
}
Ok(())
});
}
Ok(())
}
fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
self.cancel_order_internal(&cmd);
Ok(())
}
fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
let event_emitter = self.emitter.clone();
let trader_id = self.core.trader_id;
let account_id = self.core.account_id;
let clock = self.clock;
if self.ws_trading_active() {
let ws_client = self.ws_trading_client.as_ref().unwrap().clone();
let symbol = cmd.instrument_id.symbol.to_string();
self.spawn_task("cancel_all_orders_ws", async move {
if let Err(e) = ws_client.cancel_all_orders(symbol).await {
log::error!("WS cancel_all_orders failed: {e}");
}
Ok(())
});
return Ok(());
}
log::debug!("WS trading not active, falling back to HTTP for cancel_all_orders");
let http_client = self.http_client.clone();
let strategy_lookup: AHashMap<ClientOrderId, StrategyId> = {
let cache = self.core.cache();
cache
.orders_open(None, Some(&cmd.instrument_id), None, None, None)
.into_iter()
.map(|order| (order.client_order_id(), order.strategy_id()))
.collect()
};
let command = cmd;
self.spawn_task("cancel_all_orders_http", async move {
let canceled_orders = http_client.cancel_all_orders(command.instrument_id).await?;
for (venue_order_id, client_order_id) in canceled_orders {
let strategy_id = strategy_lookup
.get(&client_order_id)
.copied()
.unwrap_or(command.strategy_id);
let canceled_event = OrderCanceled::new(
trader_id,
strategy_id,
command.instrument_id,
client_order_id,
UUID4::new(),
command.ts_init,
clock.get_time_ns(),
false,
Some(venue_order_id),
Some(account_id),
);
event_emitter.send_order_event(OrderEventAny::Canceled(canceled_event));
}
Ok(())
});
Ok(())
}
fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
const BATCH_SIZE: usize = 5;
if cmd.cancels.is_empty() {
return Ok(());
}
let http_client = self.http_client.clone();
let command = cmd;
let event_emitter = self.emitter.clone();
let trader_id = self.core.trader_id;
let account_id = self.core.account_id;
let clock = self.clock;
self.spawn_task("batch_cancel_orders", async move {
for chunk in command.cancels.chunks(BATCH_SIZE) {
let batch_items: Vec<BatchCancelItem> = chunk
.iter()
.map(|cancel| {
if let Some(venue_order_id) = cancel.venue_order_id {
let order_id = venue_order_id.inner().parse::<i64>().unwrap_or(0);
if order_id != 0 {
BatchCancelItem::by_order_id(
command.instrument_id.symbol.to_string(),
order_id,
)
} else {
BatchCancelItem::by_client_order_id(
command.instrument_id.symbol.to_string(),
encode_broker_id(
&cancel.client_order_id,
BINANCE_NAUTILUS_SPOT_BROKER_ID,
),
)
}
} else {
BatchCancelItem::by_client_order_id(
command.instrument_id.symbol.to_string(),
encode_broker_id(
&cancel.client_order_id,
BINANCE_NAUTILUS_SPOT_BROKER_ID,
),
)
}
})
.collect();
match http_client.batch_cancel_orders(&batch_items).await {
Ok(results) => {
for (i, result) in results.iter().enumerate() {
let cancel = &chunk[i];
match result {
BatchCancelResult::Success(success) => {
let venue_order_id =
VenueOrderId::new(success.order_id.to_string());
let canceled_event = OrderCanceled::new(
trader_id,
cancel.strategy_id,
cancel.instrument_id,
cancel.client_order_id,
UUID4::new(),
cancel.ts_init,
clock.get_time_ns(),
false,
Some(venue_order_id),
Some(account_id),
);
event_emitter
.send_order_event(OrderEventAny::Canceled(canceled_event));
}
BatchCancelResult::Error(error) => {
let rejected_event = OrderCancelRejected::new(
trader_id,
cancel.strategy_id,
cancel.instrument_id,
cancel.client_order_id,
format!(
"batch-cancel-error: code={}, msg={}",
error.code, error.msg
)
.into(),
UUID4::new(),
clock.get_time_ns(),
cancel.ts_init,
false,
cancel.venue_order_id,
Some(account_id),
);
event_emitter.send_order_event(OrderEventAny::CancelRejected(
rejected_event,
));
}
}
}
}
Err(e) => {
if is_local_http_command_failure(&e) {
log::warn!(
"Batch cancel command failed local validation for {} orders: {e}",
chunk.len()
);
} else {
log::warn!(
"Ambiguous batch cancel failure for {} orders, awaiting reconciliation: {e}",
chunk.len()
);
}
}
}
}
Ok(())
});
Ok(())
}
}
fn max_trade_id(reports: &[FillReport]) -> anyhow::Result<i64> {
let mut max_trade_id = None;
for report in reports {
let trade_id = parse_trade_id(report)?;
max_trade_id = Some(max_trade_id.map_or(trade_id, |current: i64| current.max(trade_id)));
}
max_trade_id.context("Binance Spot account-trades page was empty")
}
fn parse_trade_id(report: &FillReport) -> anyhow::Result<i64> {
report
.trade_id
.to_string()
.parse::<i64>()
.with_context(|| format!("invalid Binance Spot trade ID {}", report.trade_id))
}
fn report_time_ms(report: &FillReport) -> i64 {
report.ts_event.as_i64() / NANOSECONDS_IN_MILLISECOND as i64
}
fn normalize_spot_order_status_report(
report: &mut OrderStatusReport,
treat_expired_as_canceled: bool,
) {
if treat_expired_as_canceled && report.order_status == OrderStatus::Expired {
report.order_status = OrderStatus::Canceled;
}
}
fn normalize_spot_order_status_reports(
reports: &mut [OrderStatusReport],
treat_expired_as_canceled: bool,
) {
for report in reports {
normalize_spot_order_status_report(report, treat_expired_as_canceled);
}
}
async fn wait_for_ws_setup_response(
timeout: Duration,
success: impl Future<Output = ()>,
setup_errors: &mut tokio::sync::mpsc::UnboundedReceiver<String>,
timeout_message: &'static str,
) -> anyhow::Result<()> {
tokio::pin!(success);
let result = tokio::time::timeout(timeout, async {
tokio::select! {
() = &mut success => Ok(()),
err = setup_errors.recv() => {
anyhow::bail!(
"{}",
err.unwrap_or_else(|| "WS setup error channel closed".to_string()),
)
}
}
})
.await;
result.map_err(|_| anyhow::anyhow!(timeout_message))?
}
#[expect(clippy::too_many_arguments)]
fn dispatch_ws_trading_message(
msg: BinanceSpotWsTradingMessage,
emitter: &ExecutionEventEmitter,
http_client: &BinanceSpotHttpClient,
account_id: AccountId,
treat_expired_as_canceled: bool,
clock: &'static AtomicTime,
dispatch_state: &WsDispatchState,
ws_authenticated: &tokio::sync::Notify,
ws_user_data_subscribed: &tokio::sync::Notify,
ws_setup_error_tx: &tokio::sync::mpsc::UnboundedSender<String>,
seen_trade_ids: &std::sync::Arc<Mutex<FifoCache<(Ustr, i64), 10_000>>>,
) {
match msg {
BinanceSpotWsTradingMessage::OrderAccepted {
request_id,
response,
} => {
dispatch_state.pending_requests.remove(&request_id);
log::debug!(
"WS order accepted: request_id={request_id}, order_id={}",
response.order_id
);
}
BinanceSpotWsTradingMessage::OrderRejected {
request_id,
code,
msg,
} => {
log::debug!("WS order rejected: request_id={request_id}, code={code}, msg={msg}");
if let Some((_, pending)) = dispatch_state.pending_requests.remove(&request_id) {
let code_i64 = i64::from(code);
if matches!(
code_i64,
BINANCE_UNEXPECTED_RESPONSE_CODE | BINANCE_STATUS_UNKNOWN_CODE
) {
log::warn!(
"Ambiguous WS submit failure for {}, awaiting reconciliation: code={code}, msg={msg}",
pending.client_order_id,
);
return;
}
let identity = dispatch_state
.order_identities
.get(&pending.client_order_id)
.map(|r| r.clone());
if let Some(identity) = identity {
let due_post_only = code_i64 == BINANCE_GTX_ORDER_REJECT_CODE
|| (code_i64 == BINANCE_NEW_ORDER_REJECTED_CODE
&& msg == BINANCE_SPOT_POST_ONLY_REJECT_MSG);
let ts_now = clock.get_time_ns();
let rejected = OrderRejected::new(
emitter.trader_id(),
identity.strategy_id,
identity.instrument_id,
pending.client_order_id,
account_id,
Ustr::from(&format!("code={code}: {msg}")),
UUID4::new(),
ts_now,
ts_now,
false,
due_post_only,
);
dispatch_state.cleanup_terminal(pending.client_order_id);
emitter.send_order_event(OrderEventAny::Rejected(rejected));
} else {
log::warn!(
"No order identity for {}, cannot emit OrderRejected",
pending.client_order_id
);
}
} else {
log::warn!("No pending request for {request_id}, cannot emit OrderRejected");
}
}
BinanceSpotWsTradingMessage::OrderCanceled {
request_id,
response,
} => {
dispatch_state.pending_requests.remove(&request_id);
log::debug!(
"WS order canceled: request_id={request_id}, order_id={}",
response.order_id
);
}
BinanceSpotWsTradingMessage::CancelRejected {
request_id,
code,
msg,
} => {
log::warn!("WS cancel rejected: request_id={request_id}, code={code}, msg={msg}");
if let Some((_, pending)) = dispatch_state.pending_requests.remove(&request_id)
&& let Some(identity) = dispatch_state
.order_identities
.get(&pending.client_order_id)
{
let ts_now = clock.get_time_ns();
let rejected = OrderCancelRejected::new(
emitter.trader_id(),
identity.strategy_id,
identity.instrument_id,
pending.client_order_id,
Ustr::from(&format!("code={code}: {msg}")),
UUID4::new(),
ts_now,
ts_now,
false,
pending.venue_order_id,
Some(account_id),
);
emitter.send_order_event(OrderEventAny::CancelRejected(rejected));
}
}
BinanceSpotWsTradingMessage::CancelReplaceAccepted {
request_id,
cancel_response,
new_order_response,
} => {
dispatch_state.pending_requests.remove(&request_id);
log::debug!(
"WS cancel-replace accepted: request_id={request_id}, \
canceled_id={}, new_id={}",
cancel_response.order_id,
new_order_response.order_id,
);
}
BinanceSpotWsTradingMessage::CancelReplaceRejected {
request_id,
code,
msg,
} => {
log::warn!(
"WS cancel-replace rejected: request_id={request_id}, code={code}, msg={msg}"
);
if let Some((_, pending)) = dispatch_state.pending_requests.remove(&request_id)
&& let Some(identity) = dispatch_state
.order_identities
.get(&pending.client_order_id)
{
let ts_now = clock.get_time_ns();
let rejected = OrderModifyRejected::new(
emitter.trader_id(),
identity.strategy_id,
identity.instrument_id,
pending.client_order_id,
Ustr::from(&format!("code={code}: {msg}")),
UUID4::new(),
ts_now,
ts_now,
false,
pending.venue_order_id,
Some(account_id),
);
emitter.send_order_event(OrderEventAny::ModifyRejected(rejected));
}
}
BinanceSpotWsTradingMessage::RequestFailed { request_id, msg } => {
dispatch_state.pending_requests.remove(&request_id);
log::error!(
"WS trading request failed without structured venue response: request_id={request_id}, {msg}"
);
}
BinanceSpotWsTradingMessage::AllOrdersCanceled {
request_id,
responses,
} => {
dispatch_state.pending_requests.remove(&request_id);
log::debug!(
"WS all orders canceled: request_id={request_id}, count={}",
responses.len()
);
}
BinanceSpotWsTradingMessage::UserDataSubscribed { subscription_id } => {
log::debug!("User data stream subscribed: id={subscription_id}");
ws_user_data_subscribed.notify_one();
}
BinanceSpotWsTradingMessage::ExecutionReport(report) => {
let ts_init = clock.get_time_ns();
dispatch_execution_report(
&report,
emitter,
http_client,
account_id,
treat_expired_as_canceled,
dispatch_state,
seen_trade_ids,
ts_init,
);
}
BinanceSpotWsTradingMessage::AccountPosition(position) => {
let ts_init = clock.get_time_ns();
let state = parse_spot_account_position(&position, account_id, ts_init);
emitter.send_account_state(state);
}
BinanceSpotWsTradingMessage::BalanceUpdate(update) => {
log::debug!(
"Balance update: asset={}, delta={}",
update.asset,
update.delta,
);
let http_client = http_client.clone();
let emitter = emitter.clone();
get_runtime().spawn(async move {
match http_client.request_account_state(account_id).await {
Ok(state) => emitter.send_account_state(state),
Err(e) => {
log::error!("Failed to refresh account state after balance update: {e}");
}
}
});
}
BinanceSpotWsTradingMessage::Connected => {
log::debug!("WS trading API connected");
}
BinanceSpotWsTradingMessage::Authenticated => {
log::debug!("WS trading API authenticated");
ws_authenticated.notify_one();
}
BinanceSpotWsTradingMessage::Reconnected => {
log::info!("WS trading API reconnected");
}
BinanceSpotWsTradingMessage::ServerShutdown { event_time } => {
log::warn!(
"WS trading API server shutdown notice (event_time={event_time}); reconnect expected within ~10 minutes"
);
}
BinanceSpotWsTradingMessage::Error(err) => {
log::error!("WS trading API error: {err}");
let _ = ws_setup_error_tx.send(err);
}
}
}
fn build_new_order_params(
order: &impl Order,
client_order_id: ClientOrderId,
is_post_only: bool,
is_quote_quantity: bool,
use_gtd: bool,
) -> anyhow::Result<NewOrderParams> {
let binance_side = BinanceSide::try_from(order.order_side())?;
let binance_order_type = order_type_to_binance_spot(order.order_type(), is_post_only)?;
let requires_trigger = matches!(
order.order_type(),
OrderType::StopMarket
| OrderType::StopLimit
| OrderType::MarketIfTouched
| OrderType::LimitIfTouched
);
if requires_trigger && order.trigger_price().is_none() {
anyhow::bail!("Conditional orders require a trigger price");
}
let supports_tif = matches!(
binance_order_type,
BinanceSpotOrderType::Limit
| BinanceSpotOrderType::StopLossLimit
| BinanceSpotOrderType::TakeProfitLimit
);
let binance_tif = time_in_force_to_binance_spot(order.time_in_force(), use_gtd)?;
let binance_tif = if supports_tif {
Some(binance_tif)
} else {
None
};
let qty_str = order.quantity().to_string();
let (base_qty, quote_qty) = if is_quote_quantity {
(None, Some(qty_str))
} else {
(Some(qty_str), None)
};
let client_id_str = encode_broker_id(&client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID);
Ok(NewOrderParams {
symbol: order.instrument_id().symbol.to_string(),
side: binance_side,
order_type: binance_order_type,
time_in_force: binance_tif,
quantity: base_qty,
quote_order_qty: quote_qty,
price: order.price().map(|p| p.to_string()),
new_client_order_id: Some(client_id_str),
stop_price: order.trigger_price().map(|p| p.to_string()),
trailing_delta: None,
iceberg_qty: order.display_qty().map(|q| q.to_string()),
new_order_resp_type: Some(BinanceOrderResponseType::Full),
self_trade_prevention_mode: None,
strategy_id: None,
strategy_type: None,
})
}
fn build_spot_order_list_params(
order_list_id: &str,
orders: &[OrderAny],
use_gtd: bool,
) -> Result<NewOcoOrderListParams, String> {
let has_grouped_order = orders.iter().any(is_grouped_order);
if has_grouped_order {
return build_spot_oco_order_list_params(order_list_id, orders, use_gtd);
}
Err("Binance Spot order-list submission currently supports only OCO lists".to_string())
}
fn build_spot_oco_order_list_params(
order_list_id: &str,
orders: &[OrderAny],
use_gtd: bool,
) -> Result<NewOcoOrderListParams, String> {
if orders.len() != 2 {
return Err(format!(
"Binance Spot OCO order-list submission requires exactly 2 orders, was {}",
orders.len()
));
}
if orders
.iter()
.any(|order| order.contingency_type() != Some(ContingencyType::Oco))
{
return Err(
"Binance Spot grouped order-list submission currently supports only OCO lists"
.to_string(),
);
}
let first = &orders[0];
let second = &orders[1];
if first.instrument_id() != second.instrument_id() {
return Err("Binance Spot OCO order-list legs must use the same instrument".to_string());
}
if first.order_side() != second.order_side() {
return Err("Binance Spot OCO order-list legs must use the same side".to_string());
}
if first.quantity() != second.quantity() {
return Err("Binance Spot OCO order-list legs must use the same quantity".to_string());
}
if first.is_quote_quantity() || second.is_quote_quantity() {
return Err("Binance Spot OCO order-list legs do not support quote quantity".to_string());
}
let mut above = None;
let mut below = None;
for order in orders {
let params = build_new_order_params(
order,
order.client_order_id(),
order.is_post_only(),
false,
use_gtd,
)
.map_err(|e| e.to_string())?;
match spot_oco_leg_position(params.side, params.order_type)? {
SpotOcoLegPosition::Above => {
if above.replace(params).is_some() {
return Err(
"Binance Spot OCO order-list resolved more than one above leg".to_string(),
);
}
}
SpotOcoLegPosition::Below => {
if below.replace(params).is_some() {
return Err(
"Binance Spot OCO order-list resolved more than one below leg".to_string(),
);
}
}
}
}
let above = above.ok_or_else(|| "Binance Spot OCO order-list missing above leg".to_string())?;
let below = below.ok_or_else(|| "Binance Spot OCO order-list missing below leg".to_string())?;
let quantity = above
.quantity
.clone()
.ok_or_else(|| "Binance Spot OCO order-list requires base quantity".to_string())?;
Ok(NewOcoOrderListParams {
symbol: first.instrument_id().symbol.to_string(),
list_client_order_id: Some(order_list_id.to_string()),
side: above.side,
quantity,
above_type: above.order_type,
above_client_order_id: above.new_client_order_id,
above_iceberg_qty: above.iceberg_qty,
above_price: above.price,
above_stop_price: above.stop_price,
above_time_in_force: above.time_in_force,
below_type: below.order_type,
below_client_order_id: below.new_client_order_id,
below_iceberg_qty: below.iceberg_qty,
below_price: below.price,
below_stop_price: below.stop_price,
below_time_in_force: below.time_in_force,
new_order_resp_type: Some(BinanceOrderResponseType::Full),
self_trade_prevention_mode: None,
})
}
enum SpotOcoLegPosition {
Above,
Below,
}
fn spot_oco_leg_position(
side: BinanceSide,
order_type: BinanceSpotOrderType,
) -> Result<SpotOcoLegPosition, String> {
match (side, order_type) {
(
BinanceSide::Sell,
BinanceSpotOrderType::LimitMaker
| BinanceSpotOrderType::TakeProfit
| BinanceSpotOrderType::TakeProfitLimit,
)
| (
BinanceSide::Buy,
BinanceSpotOrderType::StopLoss | BinanceSpotOrderType::StopLossLimit,
) => Ok(SpotOcoLegPosition::Above),
(
BinanceSide::Sell,
BinanceSpotOrderType::StopLoss | BinanceSpotOrderType::StopLossLimit,
)
| (
BinanceSide::Buy,
BinanceSpotOrderType::LimitMaker
| BinanceSpotOrderType::TakeProfit
| BinanceSpotOrderType::TakeProfitLimit,
) => Ok(SpotOcoLegPosition::Below),
(_, unsupported) => Err(format!(
"Unsupported Binance Spot OCO leg order type: {unsupported:?}"
)),
}
}
fn is_grouped_order(order: &OrderAny) -> bool {
matches!(
order.contingency_type(),
Some(contingency_type) if contingency_type != ContingencyType::NoContingency
) || order
.linked_order_ids()
.is_some_and(|linked_order_ids| !linked_order_ids.is_empty())
}
fn handle_spot_order_list_submit_error(
event_emitter: &ExecutionEventEmitter,
dispatch_state: &WsDispatchState,
trader_id: TraderId,
account_id: AccountId,
clock: &'static AtomicTime,
orders: &[OrderAny],
error: BinanceSpotHttpError,
) -> anyhow::Result<()> {
let ambiguous = matches!(
error,
BinanceSpotHttpError::BinanceError {
code: BINANCE_UNEXPECTED_RESPONSE_CODE | BINANCE_STATUS_UNKNOWN_CODE,
..
}
);
if ambiguous {
log::error!("Ambiguous order-list submit failure, awaiting reconciliation: {error}");
return Err(error.into());
}
let reject_orders = matches!(
error,
BinanceSpotHttpError::BinanceError { .. }
| BinanceSpotHttpError::MissingCredentials
| BinanceSpotHttpError::ValidationError(_)
);
if reject_orders {
let ts_now = clock.get_time_ns();
let reason = format!("submit-order-list-error: {error}");
for order in orders {
let client_order_id = order.client_order_id();
dispatch_state.cleanup_terminal(client_order_id);
let rejected = OrderRejected::new(
trader_id,
order.strategy_id(),
order.instrument_id(),
client_order_id,
account_id,
reason.clone().into(),
UUID4::new(),
ts_now,
ts_now,
false,
false,
);
event_emitter.send_order_event(OrderEventAny::Rejected(rejected));
}
} else {
log::error!("Order-list submit failed, awaiting reconciliation: {error}");
}
Err(error.into())
}
fn build_cancel_order_params(cmd: &CancelOrder) -> CancelOrderParams {
let order_id = cmd
.venue_order_id
.and_then(|id| id.inner().parse::<i64>().ok());
if let Some(order_id) = order_id {
CancelOrderParams::by_order_id(cmd.instrument_id.symbol.to_string(), order_id)
} else {
let client_id_str = encode_broker_id(&cmd.client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID);
CancelOrderParams::by_client_order_id(cmd.instrument_id.symbol.to_string(), client_id_str)
}
}
fn build_cancel_replace_params(
cmd: &ModifyOrder,
order: &impl Order,
quantity: Quantity,
use_gtd: bool,
) -> anyhow::Result<CancelReplaceOrderParams> {
let binance_side = BinanceSide::try_from(order.order_side())?;
let binance_order_type = order_type_to_binance_spot(order.order_type(), false)?;
let binance_tif = time_in_force_to_binance_spot(order.time_in_force(), use_gtd)?;
let cancel_order_id: Option<i64> = cmd
.venue_order_id
.map(|id| {
id.inner()
.parse::<i64>()
.map_err(|_| anyhow::anyhow!("Invalid venue order ID: {id}"))
})
.transpose()?;
let client_id_str = encode_broker_id(&cmd.client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID);
Ok(CancelReplaceOrderParams {
symbol: cmd.instrument_id.symbol.to_string(),
side: binance_side,
order_type: binance_order_type,
cancel_replace_mode: BinanceCancelReplaceMode::StopOnFailure,
time_in_force: Some(binance_tif),
quantity: Some(quantity.to_string()),
quote_order_qty: None,
price: cmd.price.map(|p| p.to_string()),
cancel_order_id,
cancel_orig_client_order_id: if cancel_order_id.is_none() {
Some(client_id_str.clone())
} else {
None
},
new_client_order_id: Some(client_id_str),
stop_price: None,
trailing_delta: None,
iceberg_qty: None,
new_order_resp_type: Some(BinanceOrderResponseType::Full),
self_trade_prevention_mode: None,
})
}
#[expect(clippy::too_many_arguments)]
fn dispatch_execution_report(
report: &BinanceSpotExecutionReport,
emitter: &ExecutionEventEmitter,
http_client: &BinanceSpotHttpClient,
account_id: AccountId,
treat_expired_as_canceled: bool,
dispatch_state: &WsDispatchState,
seen_trade_ids: &std::sync::Arc<Mutex<FifoCache<(Ustr, i64), 10_000>>>,
ts_init: UnixNanos,
) {
let symbol = report.symbol;
let instrument_id = InstrumentId::new(symbol.into(), *BINANCE_VENUE);
let (price_precision, size_precision) = http_client
.get_instrument(&symbol)
.map_or((8, 8), |i| (i.price_precision(), i.size_precision()));
let client_order_id =
match decode_client_order_id(&report.client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID) {
Ok(client_order_id) => client_order_id,
Err(e) => {
log::warn!("Skipping Spot execution report with invalid client order ID: {e}");
return;
}
};
let identity = dispatch_state
.order_identities
.get(&client_order_id)
.map(|r| r.clone());
if let Some(identity) = identity {
dispatch_tracked_execution_report(
report,
emitter,
account_id,
treat_expired_as_canceled,
dispatch_state,
seen_trade_ids,
client_order_id,
&identity,
instrument_id,
price_precision,
size_precision,
ts_init,
);
} else {
dispatch_untracked_execution_report(
report,
emitter,
http_client,
account_id,
treat_expired_as_canceled,
seen_trade_ids,
instrument_id,
price_precision,
size_precision,
ts_init,
);
}
}
#[expect(clippy::too_many_arguments)]
fn dispatch_tracked_execution_report(
report: &BinanceSpotExecutionReport,
emitter: &ExecutionEventEmitter,
account_id: AccountId,
treat_expired_as_canceled: bool,
state: &WsDispatchState,
seen_trade_ids: &std::sync::Arc<Mutex<FifoCache<(Ustr, i64), 10_000>>>,
client_order_id: ClientOrderId,
identity: &OrderIdentity,
instrument_id: InstrumentId,
price_precision: u8,
size_precision: u8,
ts_init: UnixNanos,
) {
let venue_order_id = VenueOrderId::new(report.order_id.to_string());
let ts_event = parse_millis_or_init(report.event_time, "Spot execution event time", ts_init);
match report.execution_type {
BinanceSpotExecutionType::New => {
if state.has_filled(&client_order_id) {
log::debug!("Skipping New for already-filled {client_order_id}");
return;
}
if state.has_emitted_accepted(&client_order_id) {
let Some(price) = parse_spot_execution_report_price(
report,
&report.price,
price_precision,
"price",
) else {
return;
};
let Some(quantity) = parse_spot_execution_report_quantity(
report,
&report.original_qty,
size_precision,
"original_qty",
) else {
return;
};
let Some(stop_price) =
parse_spot_execution_report_decimal(report, &report.stop_price, "stop_price")
else {
return;
};
let trigger = if stop_price > Decimal::ZERO {
let Some(trigger_price) = parse_spot_execution_report_price(
report,
&report.stop_price,
price_precision,
"stop_price",
) else {
return;
};
Some(trigger_price)
} else {
None
};
let updated = OrderUpdated::new(
emitter.trader_id(),
identity.strategy_id,
identity.instrument_id,
client_order_id,
quantity,
UUID4::new(),
ts_event,
ts_init,
false,
Some(venue_order_id),
Some(account_id),
Some(price),
trigger,
None, false, );
emitter.send_order_event(OrderEventAny::Updated(updated));
return;
}
state.insert_accepted(client_order_id);
let accepted = OrderAccepted::new(
emitter.trader_id(),
identity.strategy_id,
identity.instrument_id,
client_order_id,
venue_order_id,
account_id,
UUID4::new(),
ts_event,
ts_init,
false,
);
emitter.send_order_event(OrderEventAny::Accepted(accepted));
}
BinanceSpotExecutionType::Trade => {
let dedup_key = (report.symbol, report.trade_id);
let mut guard = seen_trade_ids.lock().expect(MUTEX_POISONED);
let is_duplicate = guard.contains(&dedup_key);
guard.add(dedup_key);
drop(guard);
if is_duplicate {
log::debug!(
"Duplicate trade_id={} for {}, skipping",
report.trade_id,
report.symbol
);
return;
}
ensure_accepted_emitted(
client_order_id,
account_id,
venue_order_id,
identity,
emitter,
state,
ts_init,
);
let Some(last_qty) = parse_spot_execution_report_quantity(
report,
&report.last_filled_qty,
size_precision,
"last_filled_qty",
) else {
return;
};
let Some(last_px) = parse_spot_execution_report_price(
report,
&report.last_filled_price,
price_precision,
"last_filled_price",
) else {
return;
};
let Some(commission) =
parse_spot_execution_report_decimal(report, &report.commission, "commission")
else {
return;
};
let commission_currency = report
.commission_asset
.as_ref()
.map_or_else(Currency::USDT, |a| {
Currency::get_or_create_crypto(a.as_str())
});
let commission_money = match Money::from_decimal(commission, commission_currency) {
Ok(money) => money,
Err(e) => {
log::warn!(
"Failed to build Spot commission money for symbol={}, order_id={}, \
trade_id={}: {e}",
report.symbol,
report.order_id,
report.trade_id,
);
return;
}
};
let liquidity_side = if report.is_maker {
LiquiditySide::Maker
} else {
LiquiditySide::Taker
};
let filled = OrderFilled::new(
emitter.trader_id(),
identity.strategy_id,
instrument_id,
client_order_id,
venue_order_id,
account_id,
TradeId::new(report.trade_id.to_string()),
identity.order_side,
identity.order_type,
last_qty,
last_px,
commission_currency,
liquidity_side,
UUID4::new(),
ts_event,
ts_init,
false,
None,
Some(commission_money),
None,
);
state.insert_filled(client_order_id);
emitter.send_order_event(OrderEventAny::Filled(filled));
let cumulative_qty = parse_spot_execution_report_decimal(
report,
&report.cumulative_filled_qty,
"cumulative_filled_qty",
);
let original_qty =
parse_spot_execution_report_decimal(report, &report.original_qty, "original_qty");
if let (Some(original_qty), Some(cumulative_qty)) = (original_qty, cumulative_qty)
&& original_qty <= cumulative_qty
{
state.cleanup_terminal(client_order_id);
}
}
BinanceSpotExecutionType::Replaced => {
log::debug!(
"Order replaced: client_order_id={client_order_id}, venue_order_id={venue_order_id}"
);
}
BinanceSpotExecutionType::Canceled | BinanceSpotExecutionType::TradePrevention => {
ensure_accepted_emitted(
client_order_id,
account_id,
venue_order_id,
identity,
emitter,
state,
ts_init,
);
let canceled = OrderCanceled::new(
emitter.trader_id(),
identity.strategy_id,
identity.instrument_id,
client_order_id,
UUID4::new(),
ts_event,
ts_init,
false,
Some(venue_order_id),
Some(account_id),
);
state.cleanup_terminal(client_order_id);
emitter.send_order_event(OrderEventAny::Canceled(canceled));
}
BinanceSpotExecutionType::Expired => {
ensure_accepted_emitted(
client_order_id,
account_id,
venue_order_id,
identity,
emitter,
state,
ts_init,
);
state.cleanup_terminal(client_order_id);
if treat_expired_as_canceled {
let canceled = OrderCanceled::new(
emitter.trader_id(),
identity.strategy_id,
identity.instrument_id,
client_order_id,
UUID4::new(),
ts_event,
ts_init,
false,
Some(venue_order_id),
Some(account_id),
);
emitter.send_order_event(OrderEventAny::Canceled(canceled));
} else {
let expired = OrderExpired::new(
emitter.trader_id(),
identity.strategy_id,
identity.instrument_id,
client_order_id,
UUID4::new(),
ts_event,
ts_init,
false,
Some(venue_order_id),
Some(account_id),
);
emitter.send_order_event(OrderEventAny::Expired(expired));
}
}
BinanceSpotExecutionType::Rejected => {
let reason = if report.reject_reason.is_empty() {
Ustr::from("Order rejected by venue")
} else {
Ustr::from(&report.reject_reason)
};
let due_post_only = report.time_in_force == BinanceTimeInForce::Gtx
|| (report.order_type == "LIMIT_MAKER"
&& (report.reject_reason.is_empty() || report.reject_reason == "NONE"));
state.cleanup_terminal(client_order_id);
emitter.emit_order_rejected_event(
identity.strategy_id,
identity.instrument_id,
client_order_id,
reason.as_str(),
ts_init,
due_post_only,
);
}
}
}
fn parse_spot_execution_report_quantity(
report: &BinanceSpotExecutionReport,
raw: &str,
precision: u8,
field: &str,
) -> Option<Quantity> {
match parse_required_quantity_at_precision(raw, precision, field) {
Ok(value) => Some(value),
Err(e) => {
warn_invalid_spot_execution_report_field(report, field, &e);
None
}
}
}
fn parse_spot_execution_report_price(
report: &BinanceSpotExecutionReport,
raw: &str,
precision: u8,
field: &str,
) -> Option<Price> {
match parse_required_price_at_precision(raw, precision, field) {
Ok(value) => Some(value),
Err(e) => {
warn_invalid_spot_execution_report_field(report, field, &e);
None
}
}
}
fn parse_spot_execution_report_decimal(
report: &BinanceSpotExecutionReport,
raw: &str,
field: &str,
) -> Option<Decimal> {
match parse_required_decimal(raw, field) {
Ok(value) => Some(value),
Err(e) => {
warn_invalid_spot_execution_report_field(report, field, &e);
None
}
}
}
fn warn_invalid_spot_execution_report_field(
report: &BinanceSpotExecutionReport,
field: &str,
error: &anyhow::Error,
) {
log::warn!(
"Failed to parse Spot execution report {field} for symbol={}, order_id={}, \
trade_id={}, client_order_id={}: {error}",
report.symbol,
report.order_id,
report.trade_id,
report.client_order_id,
);
}
#[expect(clippy::too_many_arguments)]
fn dispatch_untracked_execution_report(
report: &BinanceSpotExecutionReport,
emitter: &ExecutionEventEmitter,
_http_client: &BinanceSpotHttpClient,
account_id: AccountId,
treat_expired_as_canceled: bool,
seen_trade_ids: &std::sync::Arc<Mutex<FifoCache<(Ustr, i64), 10_000>>>,
instrument_id: InstrumentId,
price_precision: u8,
size_precision: u8,
ts_init: UnixNanos,
) {
match report.execution_type {
BinanceSpotExecutionType::Trade => {
let dedup_key = (report.symbol, report.trade_id);
let mut guard = seen_trade_ids.lock().expect(MUTEX_POISONED);
let is_duplicate = guard.contains(&dedup_key);
guard.add(dedup_key);
drop(guard);
if is_duplicate {
log::debug!(
"Duplicate trade_id={} for {}, skipping",
report.trade_id,
report.symbol
);
return;
}
match parse_spot_exec_report_to_order_status(
report,
instrument_id,
price_precision,
size_precision,
account_id,
treat_expired_as_canceled,
ts_init,
) {
Ok(status) => emitter.send_order_status_report(status),
Err(e) => log::error!("Failed to parse order status report: {e}"),
}
match parse_spot_exec_report_to_fill(
report,
instrument_id,
price_precision,
size_precision,
account_id,
ts_init,
) {
Ok(fill) => emitter.send_fill_report(fill),
Err(e) => log::error!("Failed to parse fill report: {e}"),
}
}
BinanceSpotExecutionType::New
| BinanceSpotExecutionType::Canceled
| BinanceSpotExecutionType::Replaced
| BinanceSpotExecutionType::Rejected
| BinanceSpotExecutionType::Expired
| BinanceSpotExecutionType::TradePrevention => {
match parse_spot_exec_report_to_order_status(
report,
instrument_id,
price_precision,
size_precision,
account_id,
treat_expired_as_canceled,
ts_init,
) {
Ok(status) => emitter.send_order_status_report(status),
Err(e) => log::error!("Failed to parse order status report: {e}"),
}
}
}
}
fn is_spot_post_only_rejection(error: &BinanceSpotHttpError) -> bool {
match error {
BinanceSpotHttpError::BinanceError { code, message } => {
*code == BINANCE_GTX_ORDER_REJECT_CODE
|| (*code == BINANCE_NEW_ORDER_REJECTED_CODE
&& message == BINANCE_SPOT_POST_ONLY_REJECT_MSG)
}
_ => false,
}
}
fn is_structured_venue_rejection(err: &anyhow::Error) -> bool {
err.downcast_ref::<BinanceSpotHttpError>()
.is_some_and(|be| matches!(be, BinanceSpotHttpError::BinanceError { .. }))
}
fn is_ambiguous_submit_error(err: &anyhow::Error) -> bool {
err.downcast_ref::<BinanceSpotHttpError>()
.is_some_and(|be| {
matches!(
be,
BinanceSpotHttpError::BinanceError {
code: BINANCE_UNEXPECTED_RESPONSE_CODE | BINANCE_STATUS_UNKNOWN_CODE,
..
}
)
})
}
fn is_local_command_failure(err: &anyhow::Error) -> bool {
err.downcast_ref::<BinanceSpotHttpError>()
.is_some_and(is_local_http_command_failure)
}
fn is_local_http_command_failure(err: &BinanceSpotHttpError) -> bool {
matches!(
err,
BinanceSpotHttpError::MissingCredentials | BinanceSpotHttpError::ValidationError(_)
)
}
#[cfg(test)]
mod tests {
use nautilus_common::messages::ExecutionEvent;
use nautilus_core::time::get_atomic_clock_realtime;
use nautilus_model::{
enums::{AccountType, LiquiditySide, OrderSide},
identifiers::{StrategyId, TraderId},
};
use rstest::rstest;
use super::*;
use crate::common::enums::BinanceEnvironment;
#[rstest]
fn test_dispatch_ws_trading_message_emits_cancel_rejected_and_clears_pending_request() {
let clock = get_atomic_clock_realtime();
let (emitter, mut rx) = create_test_emitter(clock);
let http_client = create_test_http_client(clock);
let dispatch_state = create_tracked_dispatch_state(
ClientOrderId::from("TEST"),
InstrumentId::from("BTCUSDT.BINANCE"),
);
let ws_authenticated = tokio::sync::Notify::new();
let ws_user_data_subscribed = tokio::sync::Notify::new();
let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
dispatch_state.pending_requests.insert(
"req-cancel".to_string(),
PendingRequest {
client_order_id: ClientOrderId::from("TEST"),
venue_order_id: Some(VenueOrderId::from("12345")),
operation: PendingOperation::Cancel,
},
);
dispatch_ws_trading_message(
BinanceSpotWsTradingMessage::CancelRejected {
request_id: "req-cancel".to_string(),
code: -2011,
msg: "Unknown order sent".to_string(),
},
&emitter,
&http_client,
AccountId::from("BINANCE-001"),
false,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&ws_setup_error_tx,
&seen_trade_ids,
);
assert!(dispatch_state.pending_requests.get("req-cancel").is_none());
match rx
.try_recv()
.expect("Cancel rejection event should be emitted")
{
ExecutionEvent::Order(OrderEventAny::CancelRejected(event)) => {
assert_eq!(event.client_order_id, ClientOrderId::from("TEST"));
assert_eq!(event.account_id, Some(AccountId::from("BINANCE-001")));
assert!(event.reason.as_str().contains("code=-2011"));
}
other => panic!("Expected CancelRejected event, was {other:?}"),
}
}
#[rstest]
#[case(
BINANCE_UNEXPECTED_RESPONSE_CODE,
"An unexpected response was received from the message bus"
)]
#[case(
BINANCE_STATUS_UNKNOWN_CODE,
"Timeout waiting for response from backend server"
)]
fn test_dispatch_ws_trading_message_unknown_status_keeps_order_registered(
#[case] code: i64,
#[case] msg: &str,
) {
let clock = get_atomic_clock_realtime();
let (emitter, mut rx) = create_test_emitter(clock);
let http_client = create_test_http_client(clock);
let client_order_id = ClientOrderId::from("TEST");
let dispatch_state =
create_tracked_dispatch_state(client_order_id, InstrumentId::from("BTCUSDT.BINANCE"));
let ws_authenticated = tokio::sync::Notify::new();
let ws_user_data_subscribed = tokio::sync::Notify::new();
let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
dispatch_state.pending_requests.insert(
"req-submit".to_string(),
PendingRequest {
client_order_id,
venue_order_id: None,
operation: PendingOperation::Place,
},
);
dispatch_ws_trading_message(
BinanceSpotWsTradingMessage::OrderRejected {
request_id: "req-submit".to_string(),
code: code as i32,
msg: msg.to_string(),
},
&emitter,
&http_client,
AccountId::from("BINANCE-001"),
false,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&ws_setup_error_tx,
&seen_trade_ids,
);
assert!(dispatch_state.pending_requests.get("req-submit").is_none());
assert!(
dispatch_state
.order_identities
.get(&client_order_id)
.is_some()
);
assert!(rx.try_recv().is_err());
}
#[rstest]
fn test_dispatch_ws_trading_message_definite_submit_rejection_emits_order_rejected() {
let clock = get_atomic_clock_realtime();
let (emitter, mut rx) = create_test_emitter(clock);
let http_client = create_test_http_client(clock);
let client_order_id = ClientOrderId::from("TEST");
let dispatch_state =
create_tracked_dispatch_state(client_order_id, InstrumentId::from("BTCUSDT.BINANCE"));
let ws_authenticated = tokio::sync::Notify::new();
let ws_user_data_subscribed = tokio::sync::Notify::new();
let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
dispatch_state.pending_requests.insert(
"req-submit".to_string(),
PendingRequest {
client_order_id,
venue_order_id: None,
operation: PendingOperation::Place,
},
);
dispatch_ws_trading_message(
BinanceSpotWsTradingMessage::OrderRejected {
request_id: "req-submit".to_string(),
code: BINANCE_NEW_ORDER_REJECTED_CODE as i32,
msg: BINANCE_SPOT_POST_ONLY_REJECT_MSG.to_string(),
},
&emitter,
&http_client,
AccountId::from("BINANCE-001"),
false,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&ws_setup_error_tx,
&seen_trade_ids,
);
assert!(dispatch_state.pending_requests.get("req-submit").is_none());
assert!(
dispatch_state
.order_identities
.get(&client_order_id)
.is_none()
);
match rx
.try_recv()
.expect("OrderRejected event should be emitted")
{
ExecutionEvent::Order(OrderEventAny::Rejected(event)) => {
assert_eq!(event.client_order_id, client_order_id);
assert_eq!(event.account_id, AccountId::from("BINANCE-001"));
assert!(event.reason.as_str().contains("code=-2010"));
assert!(event.due_post_only);
}
other => panic!("Expected OrderRejected event, was {other:?}"),
}
}
#[rstest]
fn test_dispatch_ws_trading_message_emits_modify_rejected_and_clears_pending_request() {
let clock = get_atomic_clock_realtime();
let (emitter, mut rx) = create_test_emitter(clock);
let http_client = create_test_http_client(clock);
let dispatch_state = create_tracked_dispatch_state(
ClientOrderId::from("TEST"),
InstrumentId::from("BTCUSDT.BINANCE"),
);
let ws_authenticated = tokio::sync::Notify::new();
let ws_user_data_subscribed = tokio::sync::Notify::new();
let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
dispatch_state.pending_requests.insert(
"req-modify".to_string(),
PendingRequest {
client_order_id: ClientOrderId::from("TEST"),
venue_order_id: Some(VenueOrderId::from("12345")),
operation: PendingOperation::Modify,
},
);
dispatch_ws_trading_message(
BinanceSpotWsTradingMessage::CancelReplaceRejected {
request_id: "req-modify".to_string(),
code: -2021,
msg: "Order cancel-replace partially failed".to_string(),
},
&emitter,
&http_client,
AccountId::from("BINANCE-001"),
false,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&ws_setup_error_tx,
&seen_trade_ids,
);
assert!(dispatch_state.pending_requests.get("req-modify").is_none());
match rx
.try_recv()
.expect("Modify rejection event should be emitted")
{
ExecutionEvent::Order(OrderEventAny::ModifyRejected(event)) => {
assert_eq!(event.client_order_id, ClientOrderId::from("TEST"));
assert_eq!(event.account_id, Some(AccountId::from("BINANCE-001")));
assert!(event.reason.as_str().contains("code=-2021"));
}
other => panic!("Expected ModifyRejected event, was {other:?}"),
}
}
fn create_test_emitter(
clock: &'static AtomicTime,
) -> (
ExecutionEventEmitter,
tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
) {
let mut emitter = ExecutionEventEmitter::new(
clock,
TraderId::from("TESTER-001"),
AccountId::from("BINANCE-001"),
AccountType::Cash,
None,
);
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
emitter.set_sender(tx);
(emitter, rx)
}
fn create_test_http_client(clock: &'static AtomicTime) -> BinanceSpotHttpClient {
BinanceSpotHttpClient::new(
BinanceEnvironment::Live,
clock,
None,
None,
None,
None,
None,
None,
)
.expect("Test HTTP client should be created")
}
fn create_tracked_dispatch_state(
client_order_id: ClientOrderId,
instrument_id: InstrumentId,
) -> WsDispatchState {
let dispatch_state = WsDispatchState::default();
dispatch_state.order_identities.insert(
client_order_id,
OrderIdentity {
instrument_id,
strategy_id: StrategyId::from("TEST-STRATEGY"),
order_side: OrderSide::Buy,
order_type: OrderType::Limit,
price: None,
quantity: Quantity::from("1"),
},
);
dispatch_state
}
#[rstest]
#[case::gtx(
BinanceSpotHttpError::BinanceError {
code: BINANCE_GTX_ORDER_REJECT_CODE,
message: "Order would immediately trigger.".to_string(),
},
true,
)]
#[case::spot_post_only(
BinanceSpotHttpError::BinanceError {
code: BINANCE_NEW_ORDER_REJECTED_CODE,
message: BINANCE_SPOT_POST_ONLY_REJECT_MSG.to_string(),
},
true,
)]
#[case::new_order_rejected_other_message(
BinanceSpotHttpError::BinanceError {
code: BINANCE_NEW_ORDER_REJECTED_CODE,
message: "Insufficient balance.".to_string(),
},
false,
)]
#[case::unrelated_code(
BinanceSpotHttpError::BinanceError {
code: -2011,
message: "Unknown order sent.".to_string(),
},
false,
)]
#[case::non_binance_error(
BinanceSpotHttpError::NetworkError("connection reset".to_string()),
false,
)]
fn test_is_spot_post_only_rejection(
#[case] error: BinanceSpotHttpError,
#[case] expected: bool,
) {
assert_eq!(is_spot_post_only_rejection(&error), expected);
}
#[rstest]
#[case(BINANCE_UNEXPECTED_RESPONSE_CODE)]
#[case(BINANCE_STATUS_UNKNOWN_CODE)]
fn test_unknown_status_submit_error_is_ambiguous(#[case] code: i64) {
let err = anyhow::Error::new(BinanceSpotHttpError::BinanceError {
code,
message: "test error".to_string(),
});
assert!(is_ambiguous_submit_error(&err));
assert!(is_structured_venue_rejection(&err));
}
#[rstest]
fn test_other_structured_submit_error_is_not_ambiguous() {
let err = anyhow::Error::new(BinanceSpotHttpError::BinanceError {
code: BINANCE_GTX_ORDER_REJECT_CODE,
message: "test error".to_string(),
});
assert!(!is_ambiguous_submit_error(&err));
assert!(is_structured_venue_rejection(&err));
}
#[rstest]
fn test_dispatch_tracked_execution_report_trade_dedup() {
let clock = get_atomic_clock_realtime();
let (emitter, mut rx) = create_test_emitter(clock);
let http_client = create_test_http_client(clock);
let client_order_id = ClientOrderId::from("x-TD67BGP9-T0000000000000");
let dispatch_state = create_tracked_dispatch_state(
ClientOrderId::from("O-20200101-000000-000-000-0"),
InstrumentId::from("ETHUSDT.BINANCE"),
);
let ws_authenticated = tokio::sync::Notify::new();
let ws_user_data_subscribed = tokio::sync::Notify::new();
let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
let trade_json = crate::common::testing::load_fixture_string(
"spot/user_data_json/execution_report_trade.json",
);
let report: BinanceSpotExecutionReport = serde_json::from_str(&trade_json).unwrap();
dispatch_ws_trading_message(
BinanceSpotWsTradingMessage::ExecutionReport(Box::new(report.clone())),
&emitter,
&http_client,
AccountId::from("BINANCE-001"),
false,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&ws_setup_error_tx,
&seen_trade_ids,
);
dispatch_ws_trading_message(
BinanceSpotWsTradingMessage::ExecutionReport(Box::new(report)),
&emitter,
&http_client,
AccountId::from("BINANCE-001"),
false,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&ws_setup_error_tx,
&seen_trade_ids,
);
let mut events = Vec::new();
while let Ok(event) = rx.try_recv() {
events.push(event);
}
let fills: Vec<_> = events
.iter()
.filter(|e| matches!(e, ExecutionEvent::Order(OrderEventAny::Filled(_))))
.collect();
assert_eq!(fills.len(), 1, "duplicate trade should be deduped");
match fills[0] {
ExecutionEvent::Order(OrderEventAny::Filled(fill)) => {
assert_eq!(
fill.client_order_id,
ClientOrderId::from("O-20200101-000000-000-000-0"),
);
assert_eq!(fill.trade_id, TradeId::new("98765432"));
assert_eq!(fill.liquidity_side, LiquiditySide::Maker);
}
_ => unreachable!(),
}
let _ = client_order_id;
}
#[rstest]
fn test_dispatch_tracked_execution_report_invalid_fill_qty_skips_filled_event() {
let clock = get_atomic_clock_realtime();
let (emitter, mut rx) = create_test_emitter(clock);
let http_client = create_test_http_client(clock);
let dispatch_state = create_tracked_dispatch_state(
ClientOrderId::from("O-20200101-000000-000-000-0"),
InstrumentId::from("ETHUSDT.BINANCE"),
);
let ws_authenticated = tokio::sync::Notify::new();
let ws_user_data_subscribed = tokio::sync::Notify::new();
let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
let trade_json = crate::common::testing::load_fixture_string(
"spot/user_data_json/execution_report_trade.json",
);
let mut report: BinanceSpotExecutionReport = serde_json::from_str(&trade_json).unwrap();
report.last_filled_qty = "not-a-number".to_string();
dispatch_ws_trading_message(
BinanceSpotWsTradingMessage::ExecutionReport(Box::new(report)),
&emitter,
&http_client,
AccountId::from("BINANCE-001"),
false,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&ws_setup_error_tx,
&seen_trade_ids,
);
let mut events = Vec::new();
while let Ok(event) = rx.try_recv() {
events.push(event);
}
assert!(
events
.iter()
.all(|e| !matches!(e, ExecutionEvent::Order(OrderEventAny::Filled(_)))),
"invalid fill quantity must not emit OrderFilled",
);
}
#[rstest]
fn test_dispatch_execution_report_invalid_client_order_id_emits_nothing() {
let clock = get_atomic_clock_realtime();
let (emitter, mut rx) = create_test_emitter(clock);
let http_client = create_test_http_client(clock);
let dispatch_state = WsDispatchState::default();
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
let json = crate::common::testing::load_fixture_string(
"spot/user_data_json/execution_report_new.json",
);
let mut report: BinanceSpotExecutionReport = serde_json::from_str(&json).unwrap();
report.client_order_id = "x-TD67BGP9-R".to_string();
dispatch_execution_report(
&report,
&emitter,
&http_client,
AccountId::from("BINANCE-001"),
false,
&dispatch_state,
&seen_trade_ids,
clock.get_time_ns(),
);
assert!(rx.try_recv().is_err());
assert!(dispatch_state.order_identities.is_empty());
}
#[rstest]
#[case::as_expired(false, OrderStatus::Expired)]
#[case::as_canceled(true, OrderStatus::Canceled)]
fn test_normalize_spot_order_status_report_expired_respects_config(
#[case] treat_expired_as_canceled: bool,
#[case] expected: OrderStatus,
) {
let clock = get_atomic_clock_realtime();
let json = crate::common::testing::load_fixture_string(
"spot/user_data_json/execution_report_expired.json",
);
let msg: BinanceSpotExecutionReport = serde_json::from_str(&json).unwrap();
let mut report = parse_spot_exec_report_to_order_status(
&msg,
InstrumentId::from("ETHUSDT.BINANCE"),
2,
5,
AccountId::from("BINANCE-001"),
false,
clock.get_time_ns(),
)
.unwrap();
let mut reports = vec![report.clone()];
normalize_spot_order_status_report(&mut report, treat_expired_as_canceled);
normalize_spot_order_status_reports(&mut reports, treat_expired_as_canceled);
assert_eq!(report.order_status, expected);
assert_eq!(reports[0].order_status, expected);
}
#[rstest]
#[case::as_expired(false)]
#[case::as_canceled(true)]
fn test_dispatch_tracked_execution_report_expired_respects_config(
#[case] treat_expired_as_canceled: bool,
) {
let clock = get_atomic_clock_realtime();
let (emitter, mut rx) = create_test_emitter(clock);
let client_order_id = ClientOrderId::from("O-20200101-000000-000-000-0");
let instrument_id = InstrumentId::from("ETHUSDT.BINANCE");
let dispatch_state = WsDispatchState::default();
dispatch_state.insert_accepted(client_order_id);
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
let identity = OrderIdentity {
instrument_id,
strategy_id: StrategyId::from("TEST-STRATEGY"),
order_side: OrderSide::Buy,
order_type: OrderType::Limit,
price: None,
quantity: Quantity::from("1"),
};
let json = crate::common::testing::load_fixture_string(
"spot/user_data_json/execution_report_expired.json",
);
let report: BinanceSpotExecutionReport = serde_json::from_str(&json).unwrap();
dispatch_tracked_execution_report(
&report,
&emitter,
AccountId::from("BINANCE-001"),
treat_expired_as_canceled,
&dispatch_state,
&seen_trade_ids,
client_order_id,
&identity,
instrument_id,
2,
5,
clock.get_time_ns(),
);
let event = rx.try_recv().expect("terminal order event expected");
match (treat_expired_as_canceled, event) {
(true, ExecutionEvent::Order(OrderEventAny::Canceled(event))) => {
assert_eq!(event.client_order_id, client_order_id);
}
(false, ExecutionEvent::Order(OrderEventAny::Expired(event))) => {
assert_eq!(event.client_order_id, client_order_id);
}
(_, other) => panic!("Expected terminal expired/canceled event, was {other:?}"),
}
assert!(rx.try_recv().is_err());
}
#[rstest]
fn test_dispatch_tracked_execution_report_rejected_gtx_sets_post_only() {
let clock = get_atomic_clock_realtime();
let (emitter, mut rx) = create_test_emitter(clock);
let http_client = create_test_http_client(clock);
let client_order_id = ClientOrderId::from("O-20200101-000000-000-000-1");
let dispatch_state =
create_tracked_dispatch_state(client_order_id, InstrumentId::from("ETHUSDT.BINANCE"));
let ws_authenticated = tokio::sync::Notify::new();
let ws_user_data_subscribed = tokio::sync::Notify::new();
let (ws_setup_error_tx, _ws_setup_error_rx) = tokio::sync::mpsc::unbounded_channel();
let seen_trade_ids = Arc::new(Mutex::new(FifoCache::new()));
let encoded = encode_broker_id(&client_order_id, BINANCE_NAUTILUS_SPOT_BROKER_ID);
let report_json = format!(
r#"{{
"e":"executionReport","E":1709654400000,"s":"ETHUSDT",
"c":"{encoded}","S":"BUY","o":"LIMIT","f":"GTX",
"q":"1.00000000","p":"2500.00000000","P":"0.00000000",
"x":"REJECTED","X":"REJECTED","r":"NONE","i":12345678,
"l":"0.00000000","z":"0.00000000","L":"0.00000000",
"n":"0","N":null,"T":1709654400000,"t":-1,"w":false,"m":false,
"O":1709654400000,"Z":"0.00000000","C":""
}}"#,
);
let report: BinanceSpotExecutionReport = serde_json::from_str(&report_json).unwrap();
dispatch_ws_trading_message(
BinanceSpotWsTradingMessage::ExecutionReport(Box::new(report)),
&emitter,
&http_client,
AccountId::from("BINANCE-001"),
false,
clock,
&dispatch_state,
&ws_authenticated,
&ws_user_data_subscribed,
&ws_setup_error_tx,
&seen_trade_ids,
);
match rx.try_recv().expect("OrderRejected event expected") {
ExecutionEvent::Order(OrderEventAny::Rejected(event)) => {
assert_eq!(event.client_order_id, client_order_id);
assert_eq!(event.account_id, AccountId::from("BINANCE-001"));
assert!(event.due_post_only);
}
other => panic!("Expected OrderRejected event, was {other:?}"),
}
}
}