diff --git a/crates/standx-cli/src/commands/maker/cycle.rs b/crates/standx-cli/src/commands/maker/cycle.rs index 258f466..5de626b 100644 --- a/crates/standx-cli/src/commands/maker/cycle.rs +++ b/crates/standx-cli/src/commands/maker/cycle.rs @@ -115,11 +115,21 @@ pub(super) async fn maker_cycle( // Try to adopt from a recent place by side + price, // tolerating a shrunk qty from a partial fill (see // open_qty_adopts). - let matched = pending.iter().position(|p| { - p.side == o.side - && (p.price - price).abs() < tick / 2.0 - && open_qty_adopts(qty, p.qty) - }); + let matched = o + .cl_ord_id + .as_ref() + .and_then(|cl_ord_id| { + pending.iter().position(|p| p.cl_ord_id == *cl_ord_id) + }) + .or_else(|| { + // Backward-compatible fallback for orders + // created before client IDs were enabled. + pending.iter().position(|p| { + p.side == o.side + && (p.price - price).abs() < tick / 2.0 + && open_qty_adopts(qty, p.qty) + }) + }); let meta = match matched { Some(idx) => { let p = pending.remove(idx); @@ -237,10 +247,11 @@ pub(super) async fn maker_cycle( } Action::Place(q) => { if live { + let cl_ord_id = format!("sxmk-{}", uuid::Uuid::new_v4()); match client .create_order(CreateOrderParams { symbol: symbol.to_string(), - cl_ord_id: None, + cl_ord_id: Some(cl_ord_id.clone()), side: q.side, order_type: OrderType::Limit, quantity: format_decimals(q.qty, cfg.qty_decimals), @@ -255,8 +266,10 @@ pub(super) async fn maker_cycle( }) .await { - Ok(_) => { + Ok(submission) => { pending.push(PendingPlace { + request_id: submission.id, + cl_ord_id, side: q.side, price: q.price, qty: q.qty, diff --git a/crates/standx-cli/src/commands/maker/mod.rs b/crates/standx-cli/src/commands/maker/mod.rs index 868ebd1..5b1fe6e 100644 --- a/crates/standx-cli/src/commands/maker/mod.rs +++ b/crates/standx-cli/src/commands/maker/mod.rs @@ -4,6 +4,7 @@ use standx_sdk::auth::Credentials; use standx_sdk::client::StandXClient; use standx_sdk::error::Error as StandxError; use standx_sdk::models::OrderSide; +use standx_sdk::order_response::OrderResponse; use std::collections::HashMap; use std::time::Duration; use tokio::signal; @@ -30,10 +31,13 @@ enum MakerExit { FailSafe(String), } -/// Pending place awaiting order-id adoption (live mode): create_order only -/// returns a request id, so new open orders are matched back to recent -/// places by (side, price, qty) on the next cycle. +/// Pending place awaiting order-id adoption in live mode. The HTTP request is +/// correlated to its asynchronous response by request ID, then to the visible +/// open order by client order ID. Side, price, and quantity remain only as a +/// compatibility fallback for orders submitted before client IDs were added. struct PendingPlace { + request_id: String, + cl_ord_id: String, side: OrderSide, price: f64, qty: f64, @@ -42,6 +46,46 @@ struct PendingPlace { cycle: u64, } +fn apply_order_responses( + receiver: &mut tokio::sync::mpsc::Receiver, + pending: &mut Vec, + output_format: OutputFormat, + symbol: &str, + cycle: u64, + price_decimals: u32, +) { + while let Ok(response) = receiver.try_recv() { + let Some(request_id) = response.request_id.as_deref() else { + continue; + }; + let Some(index) = pending + .iter() + .position(|place| place.request_id == request_id) + else { + // Authentication acknowledgements and cancellation responses do + // not correspond to a pending maker place. + continue; + }; + + if response.accepted() { + continue; + } + + let rejected = pending.remove(index); + output::log_maker_event( + output_format, + symbol, + cycle, + "place_rejected_async", + rejected.side, + rejected.level, + rejected.price, + price_decimals, + &response.message, + ); + } +} + /// A business rejection from the venue: the exchange responded with a /// definite "no" (post-only would-cross, order not found, insufficient /// margin, …). These are expected during normal quoting and must NOT trip @@ -274,8 +318,13 @@ struct MakerRunArgs { async fn run_maker(symbol: String, args: MakerRunArgs, output_format: OutputFormat) -> Result<()> { use standx_sdk::maker::{self, MakerConfig, RestingQuote}; + use standx_sdk::order_response::OrderResponseStream; - let client = StandXClient::new()?; + let order_session_id = args.live.then(|| uuid::Uuid::new_v4().to_string()); + let client = match order_session_id.as_deref() { + Some(session_id) => StandXClient::new()?.with_session_id(session_id), + None => StandXClient::new()?, + }; // ---- Startup: symbol metadata + invariants (fail fast) ---- let infos = client.get_symbol_info().await?; @@ -354,6 +403,8 @@ async fn run_maker(symbol: String, args: MakerRunArgs, output_format: OutputForm } // ---- Live gating & clean start ---- + let mut order_responses = None; + let mut order_response_handle = None; if args.live { if std::env::var(LIVE_MAKER_ENV).ok().as_deref() != Some("1") { return Err(anyhow::anyhow!( @@ -372,6 +423,14 @@ async fn run_maker(symbol: String, args: MakerRunArgs, output_format: OutputForm "Live mode requires a private key for order signing. Run 'standx auth login' with --private-key." )); } + let stream = OrderResponseStream::new( + order_session_id + .as_deref() + .expect("live maker must have an order session"), + )?; + let (responses, handle) = stream.connect().await?; + order_responses = Some(responses); + order_response_handle = Some(handle); // Start from a clean book so reconciliation isn't confused by // leftovers from a previous run. The bot owns ALL orders on this // symbol while running. @@ -500,6 +559,17 @@ async fn run_maker(symbol: String, args: MakerRunArgs, output_format: OutputForm let mut last_src: Option<&'static str> = None; let exit = 'main: loop { + if let Some(receiver) = order_responses.as_mut() { + apply_order_responses( + receiver, + &mut pending, + output_format, + &symbol, + cycle, + cfg.price_decimals, + ); + } + // Work phase raced against Ctrl+C so a slow API call can be // interrupted (mirrors run_watch_loop). let work = async { @@ -652,6 +722,9 @@ async fn run_maker(symbol: String, args: MakerRunArgs, output_format: OutputForm if args.live { cancel_all_with_retry(&client, &symbol, 3).await?; } + if let Some(handle) = order_response_handle { + handle.abort(); + } // Notify stop on every exit path. Await delivery so the message lands // before the process exits. @@ -762,4 +835,72 @@ mod tests { // Float slop just under the placed qty is tolerated. assert!(open_qty_adopts(0.01 + 1e-9, 0.01)); } + + #[test] + fn async_rejection_removes_only_matching_pending_place() { + let (sender, mut receiver) = tokio::sync::mpsc::channel(4); + let pending_place = |request_id: &str| PendingPlace { + request_id: request_id.to_string(), + cl_ord_id: format!("client-{request_id}"), + side: OrderSide::Buy, + price: 100.0, + qty: 0.01, + level: 0, + ref_center: 100.0, + cycle: 1, + }; + let mut pending = vec![pending_place("request-1"), pending_place("request-2")]; + sender + .try_send(OrderResponse { + code: 400, + message: "alo order rejected".to_string(), + request_id: Some("request-1".to_string()), + }) + .unwrap(); + + apply_order_responses( + &mut receiver, + &mut pending, + OutputFormat::Quiet, + "BTC-USD", + 2, + 2, + ); + + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].request_id, "request-2"); + } + + #[test] + fn async_acceptance_keeps_pending_until_exchange_order_is_visible() { + let (sender, mut receiver) = tokio::sync::mpsc::channel(2); + let mut pending = vec![PendingPlace { + request_id: "request-1".to_string(), + cl_ord_id: "client-1".to_string(), + side: OrderSide::Sell, + price: 101.0, + qty: 0.01, + level: 0, + ref_center: 100.0, + cycle: 1, + }]; + sender + .try_send(OrderResponse { + code: 0, + message: "accepted".to_string(), + request_id: Some("request-1".to_string()), + }) + .unwrap(); + + apply_order_responses( + &mut receiver, + &mut pending, + OutputFormat::Quiet, + "BTC-USD", + 2, + 2, + ); + + assert_eq!(pending.len(), 1); + } } diff --git a/crates/standx-sdk/src/client/mod.rs b/crates/standx-sdk/src/client/mod.rs index 52be911..d8ee87b 100644 --- a/crates/standx-sdk/src/client/mod.rs +++ b/crates/standx-sdk/src/client/mod.rs @@ -19,6 +19,7 @@ const DEFAULT_BASE_URL: &str = "https://perps.standx.com"; pub struct StandXClient { client: Client, base_url: String, + session_id: Option, } impl StandXClient { @@ -32,6 +33,7 @@ impl StandXClient { Ok(Self { client, base_url: DEFAULT_BASE_URL.to_string(), + session_id: None, }) } @@ -42,7 +44,11 @@ impl StandXClient { .connect_timeout(Duration::from_secs(10)) .build()?; - Ok(Self { client, base_url }) + Ok(Self { + client, + base_url, + session_id: None, + }) } /// Get the base URL @@ -50,6 +56,17 @@ impl StandXClient { &self.base_url } + /// Attach a stable order-response session ID to signed order requests. + pub fn with_session_id(mut self, session_id: impl Into) -> Self { + self.session_id = Some(session_id.into()); + self + } + + /// Current order-response session ID, if configured. + pub fn session_id(&self) -> Option<&str> { + self.session_id.as_deref() + } + /// Build authenticated headers with optional request signing pub async fn build_auth_headers(&self, payload: Option<&str>) -> Result { let creds = Credentials::load()?; @@ -78,6 +95,16 @@ impl StandXClient { headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + if let Some(session_id) = &self.session_id { + headers.insert( + "x-session-id", + HeaderValue::from_str(session_id).map_err(|e| Error::Validation { + field: "session_id".to_string(), + message: e.to_string(), + })?, + ); + } + // Add request signature if private key is available if !creds.private_key.is_empty() { if let Ok(signer) = StandXSigner::from_base58(&creds.private_key) { @@ -413,6 +440,14 @@ mod tests { use super::*; use mockito::Server; + #[test] + fn session_id_builder_is_stable() { + let client = StandXClient::with_base_url("https://example.invalid".to_string()) + .unwrap() + .with_session_id("maker-session"); + assert_eq!(client.session_id(), Some("maker-session")); + } + #[tokio::test] async fn test_get_symbol_info() { let mut server = Server::new_async().await; diff --git a/crates/standx-sdk/src/lib.rs b/crates/standx-sdk/src/lib.rs index f256dfe..33b607f 100644 --- a/crates/standx-sdk/src/lib.rs +++ b/crates/standx-sdk/src/lib.rs @@ -42,6 +42,7 @@ pub mod client; pub mod error; pub mod maker; pub mod models; +pub mod order_response; pub mod websocket; pub use error::{Error, Result}; diff --git a/crates/standx-sdk/src/order_response.rs b/crates/standx-sdk/src/order_response.rs new file mode 100644 index 0000000..257c545 --- /dev/null +++ b/crates/standx-sdk/src/order_response.rs @@ -0,0 +1,158 @@ +//! Correlated asynchronous responses for HTTP order requests. + +use crate::auth::Credentials; +use crate::error::{Error, Result}; +use futures::{SinkExt, StreamExt}; +use serde::{Deserialize, Serialize}; +use tokio::sync::mpsc; +use tokio_tungstenite::{connect_async, tungstenite::Message}; + +const DEFAULT_ORDER_RESPONSE_URL: &str = "wss://perps.standx.com/ws-api/v1"; + +/// Asynchronous acceptance or rejection for an order request. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct OrderResponse { + pub code: i64, + #[serde(default)] + pub message: String, + #[serde(default)] + pub request_id: Option, +} + +impl OrderResponse { + pub fn accepted(&self) -> bool { + self.code == 0 + } +} + +/// WebSocket stream paired with the `x-session-id` used by HTTP order calls. +pub struct OrderResponseStream { + url: String, + token: String, + session_id: String, +} + +impl OrderResponseStream { + /// Construct a production stream from the currently-loaded credentials. + pub fn new(session_id: impl Into) -> Result { + let credentials = Credentials::load()?; + if credentials.is_expired() { + return Err(Error::AuthRequired { + message: "Token expired".to_string(), + resolution: "Run 'standx auth login' or set STANDX_JWT environment variable" + .to_string(), + }); + } + + Ok(Self { + url: DEFAULT_ORDER_RESPONSE_URL.to_string(), + token: credentials.token, + session_id: session_id.into(), + }) + } + + #[cfg(test)] + fn with_url_and_token( + url: impl Into, + token: impl Into, + session_id: impl Into, + ) -> Self { + Self { + url: url.into(), + token: token.into(), + session_id: session_id.into(), + } + } + + pub fn session_id(&self) -> &str { + &self.session_id + } + + /// Connect, authenticate, and return parsed asynchronous order responses. + pub async fn connect( + &self, + ) -> Result<(mpsc::Receiver, tokio::task::JoinHandle<()>)> { + let (stream, _) = connect_async(&self.url).await?; + let (mut write, mut read) = stream.split(); + let auth = auth_request(&self.session_id, &self.token); + write.send(Message::Text(auth.to_string().into())).await?; + + let (tx, rx) = mpsc::channel(256); + let handle = tokio::spawn(async move { + while let Some(message) = read.next().await { + match message { + Ok(Message::Text(text)) => { + if let Ok(response) = serde_json::from_str::(&text) { + if tx.send(response).await.is_err() { + break; + } + } + } + Ok(Message::Ping(payload)) => { + if write.send(Message::Pong(payload)).await.is_err() { + break; + } + } + Ok(Message::Close(_)) | Err(_) => break, + _ => {} + } + } + }); + + Ok((rx, handle)) + } +} + +fn auth_request(session_id: &str, token: &str) -> serde_json::Value { + serde_json::json!({ + "session_id": session_id, + "request_id": uuid::Uuid::new_v4().to_string(), + "method": "auth:login", + "params": serde_json::json!({ "token": token }).to_string(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn auth_request_uses_stable_session_and_unique_request() { + let first = auth_request("maker-session", "jwt"); + let second = auth_request("maker-session", "jwt"); + + assert_eq!(first["session_id"], "maker-session"); + assert_eq!(first["method"], "auth:login"); + assert_ne!(first["request_id"], second["request_id"]); + assert!(first["params"].as_str().unwrap().contains("jwt")); + } + + #[test] + fn parses_acceptance_and_rejection_responses() { + let accepted: OrderResponse = serde_json::from_value(serde_json::json!({ + "code": 0, + "message": "success", + "request_id": "request-1" + })) + .unwrap(); + assert!(accepted.accepted()); + + let rejected: OrderResponse = serde_json::from_value(serde_json::json!({ + "code": 400, + "message": "alo order rejected", + "request_id": "request-2" + })) + .unwrap(); + assert!(!rejected.accepted()); + } + + #[test] + fn test_constructor_keeps_session_id() { + let stream = OrderResponseStream::with_url_and_token( + "ws://localhost.invalid", + "jwt", + "maker-session", + ); + assert_eq!(stream.session_id(), "maker-session"); + } +}