From b31948bdb3454afeb8a736c331a7a21c437fcdb6 Mon Sep 17 00:00:00 2001 From: BossX Date: Fri, 10 Jul 2026 09:24:25 +0800 Subject: [PATCH 1/2] fix: align maker order API contracts --- Cargo.lock | 1 + Cargo.toml | 1 + crates/standx-cli/Cargo.toml | 2 +- crates/standx-cli/src/commands/maker/cycle.rs | 1 + crates/standx-cli/src/commands/order.rs | 1 + crates/standx-sdk/Cargo.toml | 1 + crates/standx-sdk/src/auth/mod.rs | 54 ++-- crates/standx-sdk/src/client/order.rs | 287 ++++++++++++------ crates/standx-sdk/src/models.rs | 2 + 9 files changed, 231 insertions(+), 119 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fd51f16..599cedf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1898,6 +1898,7 @@ dependencies = [ "tokio-tungstenite", "tracing", "url", + "uuid", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 22ab62c..35f0157 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -38,6 +38,7 @@ chrono = { version = "0.4", features = ["serde"] } # Utilities dirs = "6.0" +uuid = { version = "1.7", features = ["v4"] } # Output formatting tabled = "0.18" diff --git a/crates/standx-cli/Cargo.toml b/crates/standx-cli/Cargo.toml index 73cc5d2..1946eed 100644 --- a/crates/standx-cli/Cargo.toml +++ b/crates/standx-cli/Cargo.toml @@ -50,7 +50,7 @@ chrono.workspace = true dirs.workspace = true rpassword = "7.3" regex = "1.10" -uuid = { version = "1.7", features = ["v4"] } +uuid.workspace = true [dev-dependencies] tokio-test = "0.4" diff --git a/crates/standx-cli/src/commands/maker/cycle.rs b/crates/standx-cli/src/commands/maker/cycle.rs index 04754a9..258f466 100644 --- a/crates/standx-cli/src/commands/maker/cycle.rs +++ b/crates/standx-cli/src/commands/maker/cycle.rs @@ -240,6 +240,7 @@ pub(super) async fn maker_cycle( match client .create_order(CreateOrderParams { symbol: symbol.to_string(), + cl_ord_id: None, side: q.side, order_type: OrderType::Limit, quantity: format_decimals(q.qty, cfg.qty_decimals), diff --git a/crates/standx-cli/src/commands/order.rs b/crates/standx-cli/src/commands/order.rs index 04dcc6e..0e9e6de 100644 --- a/crates/standx-cli/src/commands/order.rs +++ b/crates/standx-cli/src/commands/order.rs @@ -45,6 +45,7 @@ pub async fn handle_order(command: OrderCommands) -> Result<()> { let params = CreateOrderParams { symbol, + cl_ord_id: None, side, order_type, quantity: qty, diff --git a/crates/standx-sdk/Cargo.toml b/crates/standx-sdk/Cargo.toml index 23ce94a..e290218 100644 --- a/crates/standx-sdk/Cargo.toml +++ b/crates/standx-sdk/Cargo.toml @@ -44,6 +44,7 @@ chrono.workspace = true # Utilities dirs.workspace = true +uuid.workspace = true hex = "0.4" base64 = "0.22" diff --git a/crates/standx-sdk/src/auth/mod.rs b/crates/standx-sdk/src/auth/mod.rs index 7a44c0e..af2a4ed 100644 --- a/crates/standx-sdk/src/auth/mod.rs +++ b/crates/standx-sdk/src/auth/mod.rs @@ -12,7 +12,6 @@ pub use credentials::Credentials; pub struct StandXSigner { signing_key: SigningKey, verifying_key: VerifyingKey, - request_id: String, } /// Request signature headers @@ -41,41 +40,48 @@ impl StandXSigner { })?); let verifying_key = signing_key.verifying_key(); - let request_id = hex::encode(verifying_key.as_bytes()); Ok(Self { signing_key, verifying_key, - request_id, }) } - /// Get the request ID (hex-encoded public key) - pub fn request_id(&self) -> &str { - &self.request_id - } - /// Get the public key in hex format pub fn pubkey_hex(&self) -> String { hex::encode(self.verifying_key.as_bytes()) } - /// Sign a request - pub fn sign_request(&self, timestamp: u64, payload: &str) -> RequestSignature { + /// Sign a request with a caller-provided request ID. + /// + /// This is useful for deterministic tests and for protocols that need to + /// correlate an externally-created UUID with an asynchronous response. + pub fn sign_request_with_id( + &self, + request_id: &str, + timestamp: u64, + payload: &str, + ) -> RequestSignature { let version = "v1"; - let message = format!("{},{},{},{}", version, self.request_id, timestamp, payload); + let message = format!("{version},{request_id},{timestamp},{payload}"); let signature = self.signing_key.sign(message.as_bytes()); RequestSignature { version: version.to_string(), - request_id: self.request_id.clone(), + request_id: request_id.to_string(), timestamp, signature: STANDARD.encode(signature.to_bytes()), pubkey: self.pubkey_hex(), } } + /// Sign a request with a fresh UUID request ID. + pub fn sign_request(&self, timestamp: u64, payload: &str) -> RequestSignature { + let request_id = uuid::Uuid::new_v4().to_string(); + self.sign_request_with_id(&request_id, timestamp, payload) + } + /// Sign a request with the current timestamp pub fn sign_request_now(&self, payload: &str) -> RequestSignature { let timestamp = chrono::Utc::now().timestamp_millis() as u64; @@ -96,7 +102,6 @@ mod tests { let signer = StandXSigner::from_base58(&private_key_base58).unwrap(); - assert!(!signer.request_id().is_empty()); assert!(!signer.pubkey_hex().is_empty()); } @@ -115,7 +120,7 @@ mod tests { assert_eq!(sig.timestamp, 1700000000000); assert!(!sig.signature.is_empty()); assert!(!sig.pubkey.is_empty()); - assert_eq!(sig.request_id, signer.request_id()); + assert!(uuid::Uuid::parse_str(&sig.request_id).is_ok()); } #[test] @@ -181,9 +186,10 @@ mod tests { let payload = r#"{"symbol":"BTC-USD","side":"buy"}"#; let timestamp = 1700000000000u64; - // Sign same payload twice - let sig1 = signer.sign_request(timestamp, payload); - let sig2 = signer.sign_request(timestamp, payload); + // Sign the same request twice with a fixed correlation ID. + let request_id = "6f071beb-1b81-4c68-a8bf-66f1589c2146"; + let sig1 = signer.sign_request_with_id(request_id, timestamp, payload); + let sig2 = signer.sign_request_with_id(request_id, timestamp, payload); // Same signer should produce same request_id and pubkey assert_eq!(sig1.request_id, sig2.request_id); @@ -192,4 +198,18 @@ mod tests { // Ed25519 is deterministic, so signatures should be identical assert_eq!(sig1.signature, sig2.signature); } + + #[test] + fn test_sign_request_uses_unique_uuid() { + let signing_key = SigningKey::generate(&mut rand::thread_rng()); + let private_key_base58 = bs58::encode(signing_key.to_bytes()).into_string(); + let signer = StandXSigner::from_base58(&private_key_base58).unwrap(); + + let sig1 = signer.sign_request(1700000000000, "{}"); + let sig2 = signer.sign_request(1700000000000, "{}"); + + assert_ne!(sig1.request_id, sig2.request_id); + assert!(uuid::Uuid::parse_str(&sig1.request_id).is_ok()); + assert!(uuid::Uuid::parse_str(&sig2.request_id).is_ok()); + } } diff --git a/crates/standx-sdk/src/client/order.rs b/crates/standx-sdk/src/client/order.rs index a7148fe..ef2c94e 100644 --- a/crates/standx-sdk/src/client/order.rs +++ b/crates/standx-sdk/src/client/order.rs @@ -9,6 +9,8 @@ use serde_json::json; #[derive(Debug, Clone)] pub struct CreateOrderParams { pub symbol: String, + /// Client-generated idempotency/correlation ID. + pub cl_ord_id: Option, pub side: OrderSide, pub order_type: OrderType, pub quantity: String, @@ -24,6 +26,7 @@ impl Default for CreateOrderParams { fn default() -> Self { Self { symbol: String::new(), + cl_ord_id: None, side: OrderSide::Buy, order_type: OrderType::Limit, quantity: String::new(), @@ -43,49 +46,7 @@ impl StandXClient { pub async fn create_order(&self, params: CreateOrderParams) -> Result { let url = format!("{}/api/new_order", self.base_url); - // Build request body - let order_type = match params.order_type { - OrderType::Market => "market", - OrderType::Limit => "limit", - }; - - let side = match params.side { - OrderSide::Buy => "buy", - OrderSide::Sell => "sell", - }; - - let tif = params - .time_in_force - .map(|t| match t { - TimeInForce::Gtc => "gtc", - TimeInForce::Ioc => "ioc", - TimeInForce::Fok => "fok", - TimeInForce::Alo => "alo", - }) - .unwrap_or("gtc"); - - let mut body = json!({ - "symbol": params.symbol, - "side": side, - "order_type": order_type, - "qty": params.quantity, - "time_in_force": tif, - "reduce_only": params.reduce_only, - }); - - // Add optional fields - if let Some(ref price) = params.price { - body["price"] = json!(price); - } - if let Some(stop_price) = params.stop_price { - body["stop_price"] = json!(stop_price); - } - if let Some(sl_price) = params.sl_price { - body["sl_price"] = json!(sl_price); - } - if let Some(tp_price) = params.tp_price { - body["tp_price"] = json!(tp_price); - } + let body = create_order_body(¶ms); let body_str = body.to_string(); let headers = self.build_auth_headers(Some(&body_str)).await?; @@ -98,34 +59,7 @@ impl StandXClient { .send() .await?; - if !response.status().is_success() { - let status = response.status(); - let text = response.text().await.unwrap_or_default(); - return Err(Error::Api { - code: status.as_u16(), - message: text, - endpoint: Some("/api/new_order".to_string()), - retryable: status.as_u16() >= 500, - }); - } - - let result: serde_json::Value = response.json().await?; - - // Check for API error - if let Some(code) = result.get("code").and_then(|c| c.as_i64()) { - if code != 0 { - let message = result - .get("message") - .and_then(|m| m.as_str()) - .unwrap_or("Order rejected"); - return Err(Error::Api { - code: code as u16, - message: message.to_string(), - endpoint: Some("/api/new_order".to_string()), - retryable: false, - }); - } - } + let result = parse_order_response(response, "/api/new_order").await?; // Build order from response let now = chrono::Utc::now().to_rfc3339(); @@ -135,6 +69,7 @@ impl StandXClient { .and_then(|r| r.as_str()) .unwrap_or("") .to_string(), + cl_ord_id: params.cl_ord_id, symbol: params.symbol, side: params.side, order_type: params.order_type, @@ -150,13 +85,14 @@ impl StandXClient { } /// Cancel an order by ID - pub async fn cancel_order(&self, symbol: &str, order_id: &str) -> Result<()> { + pub async fn cancel_order(&self, _symbol: &str, order_id: &str) -> Result<()> { let url = format!("{}/api/cancel_order", self.base_url); + let order_id = order_id.parse::().map_err(|_| Error::Validation { + field: "order_id".to_string(), + message: format!("expected an integer order ID, got '{order_id}'"), + })?; - let body = json!({ - "symbol": symbol, - "order_id": order_id.parse::().unwrap_or(0), - }); + let body = cancel_order_body(order_id); let body_str = body.to_string(); let headers = self.build_auth_headers(Some(&body_str)).await?; @@ -169,27 +105,19 @@ impl StandXClient { .send() .await?; - if !response.status().is_success() { - let status = response.status(); - let text = response.text().await.unwrap_or_default(); - return Err(Error::Api { - code: status.as_u16(), - message: text, - endpoint: Some("/api/cancel_order".to_string()), - retryable: status.as_u16() >= 500, - }); - } + parse_order_response(response, "/api/cancel_order").await?; Ok(()) } - /// Cancel all orders for a symbol - pub async fn cancel_all_orders(&self, symbol: &str) -> Result<()> { - let url = format!("{}/api/cancel_orders", self.base_url); + /// Submit one asynchronous batch cancellation by exchange order IDs. + pub async fn cancel_orders(&self, order_ids: &[i64]) -> Result<()> { + if order_ids.is_empty() { + return Ok(()); + } - let body = json!({ - "symbol": symbol, - }); + let url = format!("{}/api/cancel_orders", self.base_url); + let body = cancel_orders_body(order_ids); let body_str = body.to_string(); let headers = self.build_auth_headers(Some(&body_str)).await?; @@ -202,19 +130,124 @@ impl StandXClient { .send() .await?; - if !response.status().is_success() { - let status = response.status(); - let text = response.text().await.unwrap_or_default(); + parse_order_response(response, "/api/cancel_orders").await?; + + Ok(()) + } + + /// Cancel every currently-open order for a symbol using the documented + /// ID-list batch API. The accepted response is asynchronous; callers that + /// require a clean book must still confirm through order events or polling. + pub async fn cancel_all_orders(&self, symbol: &str) -> Result<()> { + let orders = self.get_open_orders(Some(symbol)).await?; + let order_ids = orders + .iter() + .map(|order| { + order.id.parse::().map_err(|_| Error::Validation { + field: "order_id".to_string(), + message: format!("exchange returned non-integer order ID '{}'", order.id), + }) + }) + .collect::>>()?; + + self.cancel_orders(&order_ids).await + } +} + +fn create_order_body(params: &CreateOrderParams) -> serde_json::Value { + let order_type = match params.order_type { + OrderType::Market => "market", + OrderType::Limit => "limit", + }; + let side = match params.side { + OrderSide::Buy => "buy", + OrderSide::Sell => "sell", + }; + let tif = params + .time_in_force + .map(|t| match t { + TimeInForce::Gtc => "gtc", + TimeInForce::Ioc => "ioc", + TimeInForce::Fok => "fok", + TimeInForce::Alo => "alo", + }) + .unwrap_or("gtc"); + + let mut body = json!({ + "symbol": params.symbol, + "side": side, + "order_type": order_type, + "qty": params.quantity, + "time_in_force": tif, + "reduce_only": params.reduce_only, + }); + if let Some(value) = ¶ms.cl_ord_id { + body["cl_ord_id"] = json!(value); + } + if let Some(value) = ¶ms.price { + body["price"] = json!(value); + } + if let Some(value) = ¶ms.stop_price { + body["stop_price"] = json!(value); + } + if let Some(value) = ¶ms.sl_price { + body["sl_price"] = json!(value); + } + if let Some(value) = ¶ms.tp_price { + body["tp_price"] = json!(value); + } + body +} + +fn cancel_order_body(order_id: i64) -> serde_json::Value { + json!({ "order_id": order_id }) +} + +fn cancel_orders_body(order_ids: &[i64]) -> serde_json::Value { + json!({ "order_id_list": order_ids }) +} + +async fn parse_order_response( + response: reqwest::Response, + endpoint: &str, +) -> Result { + let status = response.status(); + if !status.is_success() { + let text = response.text().await.unwrap_or_default(); + return Err(Error::Api { + code: status.as_u16(), + message: text, + endpoint: Some(endpoint.to_string()), + retryable: response_code_is_retryable(status.as_u16()), + }); + } + + let result = response.json::().await?; + ensure_order_response_success(&result, endpoint)?; + Ok(result) +} + +fn ensure_order_response_success(result: &serde_json::Value, endpoint: &str) -> Result<()> { + if let Some(code) = result.get("code").and_then(|value| value.as_i64()) { + if code != 0 { + let code = u16::try_from(code).unwrap_or(u16::MAX); + let message = result + .get("message") + .and_then(|value| value.as_str()) + .unwrap_or("Order request rejected"); return Err(Error::Api { - code: status.as_u16(), - message: text, - endpoint: Some("/api/cancel_orders".to_string()), - retryable: status.as_u16() >= 500, + code, + message: message.to_string(), + endpoint: Some(endpoint.to_string()), + retryable: response_code_is_retryable(code), }); } - - Ok(()) } + Ok(()) +} + +fn response_code_is_retryable(code: u16) -> bool { + code == 429 || code >= 500 } #[cfg(test)] @@ -246,4 +279,56 @@ mod tests { let result = client.create_order(params).await; assert!(result.is_err()); } + + #[test] + fn create_body_includes_client_order_id() { + let params = CreateOrderParams { + symbol: "BTC-USD".to_string(), + cl_ord_id: Some("maker-buy-0-42".to_string()), + side: OrderSide::Buy, + order_type: OrderType::Limit, + quantity: "0.01".to_string(), + price: Some("65000".to_string()), + time_in_force: Some(TimeInForce::Alo), + ..Default::default() + }; + + let body = create_order_body(¶ms); + assert_eq!(body["cl_ord_id"], "maker-buy-0-42"); + assert_eq!(body["time_in_force"], "alo"); + assert_eq!(body["reduce_only"], false); + } + + #[test] + fn response_codes_classify_rate_limit_as_retryable() { + assert!(!response_code_is_retryable(400)); + assert!(!response_code_is_retryable(401)); + assert!(response_code_is_retryable(429)); + assert!(response_code_is_retryable(500)); + } + + #[test] + fn cancellation_bodies_use_documented_order_id_fields() { + assert_eq!(cancel_order_body(42), serde_json::json!({ "order_id": 42 })); + assert_eq!( + cancel_orders_body(&[42, 43]), + serde_json::json!({ "order_id_list": [42, 43] }) + ); + } + + #[test] + fn response_body_errors_are_not_treated_as_success() { + let error = ensure_order_response_success( + &serde_json::json!({ "code": 429, "message": "rate limited" }), + "/api/cancel_orders", + ) + .unwrap_err(); + assert!(error.is_retryable()); + + assert!(ensure_order_response_success( + &serde_json::json!({ "code": 0, "message": "accepted" }), + "/api/cancel_orders", + ) + .is_ok()); + } } diff --git a/crates/standx-sdk/src/models.rs b/crates/standx-sdk/src/models.rs index bab82d4..3fe04e8 100644 --- a/crates/standx-sdk/src/models.rs +++ b/crates/standx-sdk/src/models.rs @@ -425,6 +425,8 @@ pub struct OrderRequest { pub struct Order { #[serde(deserialize_with = "string_or_number_to_string")] pub id: String, + #[serde(default)] + pub cl_ord_id: Option, pub symbol: String, #[serde(deserialize_with = "deserialize_order_side")] pub side: OrderSide, From f56fec051b50aa83f7c7543d7a81ad3cbc389eed Mon Sep 17 00:00:00 2001 From: BossX Date: Fri, 10 Jul 2026 09:33:40 +0800 Subject: [PATCH 2/2] feat: correlate asynchronous maker order responses --- crates/standx-cli/src/commands/maker/cycle.rs | 27 ++- crates/standx-cli/src/commands/maker/mod.rs | 149 ++++++++++++++++- crates/standx-sdk/src/client/mod.rs | 37 +++- crates/standx-sdk/src/lib.rs | 1 + crates/standx-sdk/src/order_response.rs | 158 ++++++++++++++++++ 5 files changed, 360 insertions(+), 12 deletions(-) create mode 100644 crates/standx-sdk/src/order_response.rs 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"); + } +}